mirror of
https://github.com/activecm/rita
synced 2026-06-08 13:02:45 +00:00
Naomi Kramer
b23387c2e2
* Store zone transfer records Co-Authored-By: moth <25512187+0x6d6f7468@users.noreply.github.com> * Update config * Add tests * Tests, connectivity test * Update tests --------- Co-authored-by: moth <25512187+0x6d6f7468@users.noreply.github.com>
1609 lines
40 KiB
Go
1609 lines
40 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"strconv"
|
|
|
|
clickhouse "github.com/ClickHouse/clickhouse-go/v2"
|
|
)
|
|
|
|
func (db *DB) createMinMaxMaterializedView() error {
|
|
ctx := db.QueryParameters(clickhouse.Parameters{
|
|
"database": db.selected,
|
|
"rolling": strconv.FormatBool(db.Rolling),
|
|
})
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.min_max_conn_mv
|
|
TO metadatabase.min_max AS
|
|
SELECT
|
|
{database:String} as database,
|
|
{rolling:Bool} as rolling,
|
|
true as beacon,
|
|
minSimpleState(ts) as min_ts,
|
|
maxSimpleState(ts) as max_ts
|
|
FROM {database:Identifier}.conn c
|
|
GROUP BY (database)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
// add proxy connections to min_max since their matching conn records get filtered out
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.min_max_http_mv
|
|
TO metadatabase.min_max AS
|
|
SELECT
|
|
{database:String} as database,
|
|
{rolling:Bool} as rolling,
|
|
true as beacon,
|
|
minSimpleState(ts) as min_ts,
|
|
maxSimpleState(ts) as max_ts
|
|
FROM {database:Identifier}.http c
|
|
WHERE method = 'CONNECT'
|
|
GROUP BY (database)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.min_max_openconn_mv
|
|
TO metadatabase.min_max AS
|
|
SELECT
|
|
{database:String} as database,
|
|
{rolling:Bool} as rolling,
|
|
false as beacon,
|
|
minSimpleState(ts) as min_ts,
|
|
maxSimpleState(ts) as max_ts
|
|
FROM {database:Identifier}.openconn c
|
|
GROUP BY (database)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
// add proxy connections to min_max since their matching conn records get filtered out
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.min_max_openhttp_mv
|
|
TO metadatabase.min_max AS
|
|
SELECT
|
|
{database:String} as database,
|
|
{rolling:Bool} as rolling,
|
|
false as beacon,
|
|
minSimpleState(ts) as min_ts,
|
|
maxSimpleState(ts) as max_ts
|
|
FROM {database:Identifier}.openhttp c
|
|
WHERE method = 'CONNECT'
|
|
GROUP BY (database)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.min_max_dns_mv
|
|
TO metadatabase.min_max AS
|
|
SELECT
|
|
{database:String} as database,
|
|
{rolling:Bool} as rolling,
|
|
false as beacon,
|
|
minSimpleState(hour) as min_ts,
|
|
maxSimpleState(hour) as max_ts
|
|
FROM {database:Identifier}.udns c
|
|
GROUP BY (database)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createOpenConnTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openconn_tmp (
|
|
import_time DateTime(),
|
|
filtered Bool,
|
|
import_id FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
missing_host_header Bool,
|
|
missing_host_useragent String,
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
conn_state LowCardinality(String),
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
icmp_type Int64,
|
|
icmp_code Int64,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
missed_bytes UInt64,
|
|
zeek_history String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (filtered, dst_nuid, src_nuid, src, dst, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createConnTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.conn_tmp (
|
|
import_time DateTime(),
|
|
filtered Bool,
|
|
import_id FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
missing_host_header Bool,
|
|
missing_host_useragent String,
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
conn_state LowCardinality(String),
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
icmp_type Int64,
|
|
icmp_code Int64,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
missed_bytes UInt64,
|
|
zeek_history String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (filtered, dst_nuid, src_nuid, src, dst, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createConnTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.conn (
|
|
import_time DateTime(),
|
|
import_id FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
missing_host_header Bool,
|
|
missing_host_useragent String,
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
conn_state LowCardinality(String),
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
icmp_type Int64,
|
|
icmp_code Int64,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
missed_bytes UInt64,
|
|
zeek_history String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (import_id, missing_host_header, dst_nuid, src_nuid, src, dst, hash)
|
|
ORDER BY (import_id, missing_host_header, dst_nuid, src_nuid, src, dst, hash, ts)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createBigOlHistogramTable(ctx context.Context) error {
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.big_ol_histogram (
|
|
import_hour DateTime(),
|
|
hash FixedString(16),
|
|
bucket DateTime(),
|
|
src_ip_bytes SimpleAggregateFunction(sum, UInt64),
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = SummingMergeTree()
|
|
PRIMARY KEY (hash, bucket)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.big_ol_histogram_conn_mv
|
|
TO {database:Identifier}.big_ol_histogram AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
hash,
|
|
toStartOfFifteenMinutes(ts) as bucket,
|
|
sumSimpleState(src_ip_bytes) as src_ip_bytes,
|
|
countState() as count
|
|
FROM {database:Identifier}.conn
|
|
GROUP BY (import_hour, hash, bucket)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.big_ol_histogram_http_mv
|
|
TO {database:Identifier}.big_ol_histogram AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
hash,
|
|
toStartOfFifteenMinutes(ts) as bucket,
|
|
sumSimpleState(src_ip_bytes) as src_ip_bytes,
|
|
countState() as count
|
|
FROM {database:Identifier}.http
|
|
WHERE length(host) > 0
|
|
GROUP BY (import_hour, hash, bucket)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.big_ol_histogram_ssl_mv
|
|
TO {database:Identifier}.big_ol_histogram AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
hash,
|
|
toStartOfFifteenMinutes(ts) as bucket,
|
|
sumSimpleState(src_ip_bytes) as src_ip_bytes,
|
|
countState() as count
|
|
FROM {database:Identifier}.ssl
|
|
GROUP BY (import_hour, hash, bucket)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createUconnTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.uconn (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
count AggregateFunction(count, Int64),
|
|
unique_ts_count AggregateFunction(uniqExact, DateTime()),
|
|
missing_host_header_count AggregateFunction(count, Int64),
|
|
ts_list AggregateFunction(groupArray(86400), UInt32),
|
|
src_ip_bytes_list AggregateFunction(groupArray(86400), UInt64),
|
|
total_src_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_dst_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_src_bytes AggregateFunction(sum, UInt64),
|
|
total_dst_bytes AggregateFunction(sum, UInt64),
|
|
total_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_src_packets AggregateFunction(sum, UInt64),
|
|
total_dst_packets AggregateFunction(sum, UInt64),
|
|
total_duration AggregateFunction(sum, Float64),
|
|
first_seen AggregateFunction(min, DateTime()),
|
|
last_seen AggregateFunction(max, DateTime())
|
|
) ENGINE = AggregatingMergeTree()
|
|
ORDER BY (hour, dst_nuid, src_nuid, src, dst, hash)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.uconn_mv
|
|
TO {database:Identifier}.uconn AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
src,
|
|
dst,
|
|
src_nuid,
|
|
dst_nuid,
|
|
hash,
|
|
src_local,
|
|
dst_local,
|
|
countStateIf(missing_host_header = false) as count, -- count only regular conn entries to avoid inflating the count
|
|
uniqExactState(ts) as unique_ts_count,
|
|
countStateIf(missing_host_header = true) as missing_host_header_count,
|
|
groupArrayStateIf(86400)(toUnixTimestamp(ts), missing_host_header = false) as ts_list,
|
|
groupArrayStateIf(86400)(c.src_ip_bytes, missing_host_header = false) as src_ip_bytes_list,
|
|
sumStateIf(c.src_ip_bytes, missing_host_header = false) as total_src_ip_bytes,
|
|
sumStateIf(c.dst_ip_bytes, missing_host_header = false) as total_dst_ip_bytes,
|
|
sumStateIf(c.src_bytes, missing_host_header = false) as total_src_bytes,
|
|
sumStateIf(c.dst_bytes, missing_host_header = false) as total_dst_bytes,
|
|
sumStateIf(c.src_ip_bytes + c.dst_ip_bytes, missing_host_header = false) as total_ip_bytes,
|
|
sumStateIf(c.src_packets, missing_host_header = false) as total_src_packets,
|
|
sumStateIf(c.dst_packets, missing_host_header = false) as total_dst_packets,
|
|
sumStateIf(duration, missing_host_header = false) as total_duration,
|
|
minState(ts) as first_seen,
|
|
maxState(ts) as last_seen
|
|
FROM {database:Identifier}.conn c
|
|
GROUP BY (import_hour, hour, src, src_nuid, dst, dst_nuid, hash, src_local, dst_local)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// if db.rolling {
|
|
// query += "TTL"
|
|
// }
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createUconnTmpImportTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.uconn_tmp (
|
|
hash FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = SummingMergeTree()
|
|
ORDER BY (hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.uconn_tmp_mv
|
|
TO {database:Identifier}.uconn_tmp AS
|
|
SELECT
|
|
hash AS hash,
|
|
zeek_uid,
|
|
countState() as count
|
|
FROM {database:Identifier}.conn
|
|
GROUP BY (hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createOpenConnHashTmpImportTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openconnhash_tmp (
|
|
hash FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = SummingMergeTree()
|
|
ORDER BY (hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.openconnhash_tmp_mv
|
|
TO {database:Identifier}.openconnhash_tmp AS
|
|
SELECT
|
|
hash AS hash,
|
|
zeek_uid,
|
|
countState() as count
|
|
FROM {database:Identifier}.openconn
|
|
GROUP BY (hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createOpenSNIConnTmpImportTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.opensniconn_tmp (
|
|
conn_type LowCardinality(String),
|
|
hash FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = MergeTree()
|
|
ORDER BY (conn_type, hash)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.opensniconn_ssl_tmp_mv
|
|
TO {database:Identifier}.opensniconn_tmp AS
|
|
SELECT
|
|
'ssl' as conn_type,
|
|
hash AS hash,
|
|
zeek_uid,
|
|
countState() as count
|
|
FROM {database:Identifier}.openssl
|
|
GROUP BY (conn_type, hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.opensniconn_http_tmp_mv
|
|
TO {database:Identifier}.opensniconn_tmp AS
|
|
SELECT
|
|
'http' AS conn_type,
|
|
hash AS hash,
|
|
countState() as count,
|
|
zeek_uid,
|
|
FROM {database:Identifier}.openhttp
|
|
GROUP BY (conn_type, hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createOpenConnTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openconn (
|
|
import_time DateTime(),
|
|
import_id FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
missing_host_header Bool,
|
|
missing_host_useragent String,
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
conn_state LowCardinality(String),
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
icmp_type Int64,
|
|
icmp_code Int64,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
missed_bytes UInt64,
|
|
zeek_history String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (missing_host_header, dst_nuid, src_nuid, src, dst, hash, zeek_uid)
|
|
ORDER BY (missing_host_header, dst_nuid, src_nuid, src, dst, hash, zeek_uid, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createHTTPTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.http_tmp (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
multi_request Bool,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
trans_depth UInt64,
|
|
method LowCardinality(String),
|
|
host String,
|
|
uri String,
|
|
referrer String,
|
|
http_version String,
|
|
useragent String,
|
|
origin String,
|
|
status_code UInt64,
|
|
status_msg String,
|
|
info_code UInt64,
|
|
info_msg String,
|
|
username String,
|
|
password String,
|
|
src_fuids Array(String),
|
|
src_file_names Array(String),
|
|
src_mime_types Array(String),
|
|
dst_fuids Array(String),
|
|
dst_file_names Array(String),
|
|
dst_mime_types Array(String)
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, host, dst, zeek_uid )
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createOpenHTTPTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openhttp_tmp (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
multi_request Bool,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
trans_depth UInt64,
|
|
method LowCardinality(String),
|
|
host String,
|
|
uri String,
|
|
referrer String,
|
|
http_version String,
|
|
useragent String,
|
|
origin String,
|
|
status_code UInt64,
|
|
status_msg String,
|
|
info_code UInt64,
|
|
info_msg String,
|
|
username String,
|
|
password String,
|
|
src_fuids Array(String),
|
|
src_file_names Array(String),
|
|
src_mime_types Array(String),
|
|
dst_fuids Array(String),
|
|
dst_file_names Array(String),
|
|
dst_mime_types Array(String)
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, host, dst, zeek_uid )
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createHTTPTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.http (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
multi_request Bool,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
trans_depth UInt64,
|
|
method LowCardinality(String),
|
|
host String,
|
|
uri String,
|
|
referrer String,
|
|
http_version String,
|
|
useragent String,
|
|
origin String,
|
|
status_code UInt64,
|
|
status_msg String,
|
|
info_code UInt64,
|
|
info_msg String,
|
|
username String,
|
|
password String,
|
|
src_fuids Array(String),
|
|
src_file_names Array(String),
|
|
src_mime_types Array(String),
|
|
dst_fuids Array(String),
|
|
dst_file_names Array(String),
|
|
dst_mime_types Array(String)
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, host, dst, hash)
|
|
ORDER BY (dst_nuid, src_nuid, src, host, dst, hash, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createOpenHTTPTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openhttp (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
multi_request Bool,
|
|
trans_depth UInt64,
|
|
method LowCardinality(String),
|
|
host String,
|
|
uri String,
|
|
referrer String,
|
|
http_version String,
|
|
useragent String,
|
|
origin String,
|
|
status_code UInt64,
|
|
status_msg String,
|
|
info_code UInt64,
|
|
info_msg String,
|
|
username String,
|
|
password String,
|
|
src_fuids Array(String),
|
|
src_file_names Array(String),
|
|
src_mime_types Array(String),
|
|
dst_fuids Array(String),
|
|
dst_file_names Array(String),
|
|
dst_mime_types Array(String)
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, host, dst, hash, zeek_uid)
|
|
ORDER BY (dst_nuid, src_nuid, src, host, dst, hash, zeek_uid, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createSSLTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.ssl_tmp (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
version LowCardinality(String),
|
|
cipher String,
|
|
curve LowCardinality(String),
|
|
server_name String,
|
|
resumed Bool,
|
|
next_protocol LowCardinality(String),
|
|
established Bool,
|
|
server_cert_fuids Array(String),
|
|
client_cert_fuids Array(String),
|
|
server_subject String,
|
|
server_issuer String,
|
|
client_subject String,
|
|
client_issuer String,
|
|
validation_status LowCardinality(String),
|
|
ja3 String,
|
|
ja3s String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, server_name, dst, zeek_uid)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createOpenSSLTmpTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openssl_tmp (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
version LowCardinality(String),
|
|
cipher String,
|
|
curve LowCardinality(String),
|
|
server_name String,
|
|
resumed Bool,
|
|
next_protocol LowCardinality(String),
|
|
established Bool,
|
|
server_cert_fuids Array(String),
|
|
client_cert_fuids Array(String),
|
|
server_subject String,
|
|
server_issuer String,
|
|
client_subject String,
|
|
client_issuer String,
|
|
validation_status LowCardinality(String),
|
|
ja3 String,
|
|
ja3s String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, server_name, dst, zeek_uid)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createSSLTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.ssl (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
version LowCardinality(String),
|
|
cipher String,
|
|
curve LowCardinality(String),
|
|
server_name String,
|
|
resumed Bool,
|
|
next_protocol LowCardinality(String),
|
|
established Bool,
|
|
server_cert_fuids Array(String),
|
|
client_cert_fuids Array(String),
|
|
server_subject String,
|
|
server_issuer String,
|
|
client_subject String,
|
|
client_issuer String,
|
|
validation_status LowCardinality(String),
|
|
ja3 String,
|
|
ja3s String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, server_name, dst, hash)
|
|
ORDER BY (dst_nuid, src_nuid, src, server_name, dst, hash, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
// protocol analysis for ssl connection pair
|
|
func (db *DB) createTLSProtoTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.tls_proto (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
ja3 String,
|
|
version String,
|
|
validation_status String,
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = AggregatingMergeTree()
|
|
ORDER BY (hour, hash, ja3, version, validation_status);
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.tls_proto_mv
|
|
TO {database:Identifier}.tls_proto AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
ja3,
|
|
version,
|
|
validation_status,
|
|
countState() as count
|
|
FROM {database:Identifier}.ssl s
|
|
GROUP BY (import_hour, hour, hash, ja3, ja3s, version, validation_status);
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// protocol table for http connection pair
|
|
func (db *DB) createHTTPProtoTable(ctx context.Context) error {
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.http_proto (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
useragent String,
|
|
method LowCardinality(String),
|
|
referrer String,
|
|
uri String,
|
|
dst_mime_types AggregateFunction(groupUniqArray, String),
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = AggregatingMergeTree()
|
|
ORDER BY (hour, hash, method, useragent, referrer, uri);
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.http_proto_mv
|
|
TO {database:Identifier}.http_proto AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
useragent,
|
|
method,
|
|
referrer,
|
|
uri,
|
|
groupUniqArrayArrayState(h.dst_mime_types) as dst_mime_types,
|
|
countState() as count
|
|
FROM {database:Identifier}.http h
|
|
GROUP BY (import_hour, hour, hash, method, referrer, useragent, uri);
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createUSNIConnTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.usni (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
fqdn String,
|
|
http Bool,
|
|
proxy Bool,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
count AggregateFunction(count, UInt64),
|
|
proxy_count AggregateFunction(count, UInt64),
|
|
unique_ts_count AggregateFunction(uniqExact, DateTime()),
|
|
ts_list AggregateFunction(groupArray(86400), UInt32),
|
|
src_ip_bytes_list AggregateFunction(groupArray(86400), UInt64),
|
|
total_src_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_dst_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_src_bytes AggregateFunction(sum, UInt64),
|
|
total_dst_bytes AggregateFunction(sum, UInt64),
|
|
total_ip_bytes AggregateFunction(sum, UInt64),
|
|
total_src_packets AggregateFunction(sum, UInt64),
|
|
total_dst_packets AggregateFunction(sum, UInt64),
|
|
total_duration AggregateFunction(sum, Float64),
|
|
server_ips AggregateFunction(groupUniqArray(10), IPv6),
|
|
proxy_ips AggregateFunction(groupUniqArray(10), IPv6),
|
|
first_seen AggregateFunction(min, DateTime()),
|
|
last_seen AggregateFunction(max, DateTime())
|
|
)
|
|
ENGINE = AggregatingMergeTree()
|
|
ORDER BY (hour, http, src, src_nuid, src_local, fqdn, hash)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = db.Conn.Exec(ctx, `
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.usni_ssl_mv
|
|
TO {database:Identifier}.usni AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
src,
|
|
dst,
|
|
src_nuid,
|
|
dst_nuid,
|
|
server_name as fqdn,
|
|
src_local,
|
|
dst_local,
|
|
false as http,
|
|
false as proxy,
|
|
countState() as count,
|
|
uniqExactState(ts) as unique_ts_count,
|
|
groupArrayState(86400)(toUnixTimestamp(ts)) as ts_list,
|
|
groupArrayState(86400)(s.src_ip_bytes) as src_ip_bytes_list,
|
|
sumState(s.src_ip_bytes) as total_src_ip_bytes,
|
|
sumState(s.dst_ip_bytes) as total_dst_ip_bytes,
|
|
sumState(s.src_bytes) as total_src_bytes,
|
|
sumState(s.dst_bytes) as total_dst_bytes,
|
|
sumState(s.src_ip_bytes + s.dst_ip_bytes) as total_ip_bytes,
|
|
sumState(s.src_packets) as total_src_packets,
|
|
sumState(s.dst_packets) as total_dst_packets,
|
|
sumState(duration) as total_duration,
|
|
groupUniqArrayState(10)(dst) as server_ips,
|
|
minState(ts) as first_seen,
|
|
maxState(ts) as last_seen
|
|
FROM {database:Identifier}.ssl s
|
|
GROUP BY (import_hour, hour, src, src_nuid, src_local, dst_local, dst, dst_nuid, fqdn, hash);
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.usni_http_mv
|
|
TO {database:Identifier}.usni AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
src,
|
|
dst,
|
|
src_nuid,
|
|
dst_nuid,
|
|
host as fqdn,
|
|
src_local,
|
|
dst_local,
|
|
true as http,
|
|
if(method = 'CONNECT', true, false) as proxy,
|
|
countState() as count,
|
|
countStateIf(proxy = true) AS proxy_count,
|
|
uniqExactState(ts) as unique_ts_count,
|
|
groupArrayState(86400)(toUnixTimestamp(ts)) as ts_list,
|
|
groupArrayState(86400)(h.src_ip_bytes) as src_ip_bytes_list,
|
|
sumState(h.src_ip_bytes) as total_src_ip_bytes,
|
|
sumState(h.dst_ip_bytes) as total_dst_ip_bytes,
|
|
sumState(h.src_bytes) as total_src_bytes,
|
|
sumState(h.dst_bytes) as total_dst_bytes,
|
|
sumState(h.src_ip_bytes + h.dst_ip_bytes) as total_ip_bytes,
|
|
sumState(h.src_packets) as total_src_packets,
|
|
sumState(h.dst_packets) as total_dst_packets,
|
|
sumState(duration) as total_duration,
|
|
groupUniqArrayStateIf(10)(dst, method != 'CONNECT') as server_ips,
|
|
groupUniqArrayStateIf(10)(dst, method = 'CONNECT') as proxy_ips,
|
|
minState(ts) as first_seen,
|
|
maxState(ts) as last_seen
|
|
FROM {database:Identifier}.http h
|
|
WHERE h.multi_request == false AND length(h.host) > 0
|
|
GROUP BY (import_hour, hour, src, src_nuid, src_local, dst_local, dst, dst_nuid, fqdn, hash, proxy);
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createOpenSSLTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.openssl (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
duration Float64,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
src_bytes UInt64,
|
|
src_ip_bytes UInt64,
|
|
dst_bytes UInt64,
|
|
dst_ip_bytes UInt64,
|
|
src_packets UInt64,
|
|
dst_packets UInt64,
|
|
conn_state LowCardinality(String),
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
version LowCardinality(String),
|
|
cipher String,
|
|
curve LowCardinality(String),
|
|
server_name String,
|
|
resumed Bool,
|
|
next_protocol LowCardinality(String),
|
|
established Bool,
|
|
server_cert_fuids Array(String),
|
|
client_cert_fuids Array(String),
|
|
server_subject String,
|
|
server_issuer String,
|
|
client_subject String,
|
|
client_issuer String,
|
|
validation_status LowCardinality(String),
|
|
ja3 String,
|
|
ja3s String
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, server_name, dst, hash, zeek_uid)
|
|
ORDER BY (dst_nuid, src_nuid, src, server_name, dst, hash, zeek_uid, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createSNIConnTmpImportTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.sniconn_tmp (
|
|
conn_type LowCardinality(String),
|
|
hash FixedString(16),
|
|
zeek_uid FixedString(16),
|
|
count AggregateFunction(count, UInt64)
|
|
-- uids AggregateFunction(groupUniqArray, FixedString(16))
|
|
) ENGINE = MergeTree()
|
|
ORDER BY (conn_type, hash)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.sniconn_ssl_tmp_mv
|
|
TO {database:Identifier}.sniconn_tmp AS
|
|
SELECT
|
|
'ssl' as conn_type,
|
|
hash AS hash,
|
|
zeek_uid,
|
|
countState() as count
|
|
FROM {database:Identifier}.ssl
|
|
GROUP BY (conn_type, hash, zeek_uid)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.sniconn_http_tmp_mv
|
|
TO {database:Identifier}.sniconn_tmp AS
|
|
SELECT
|
|
'http' AS conn_type,
|
|
hash AS hash,
|
|
countState() as count,
|
|
zeek_uid,
|
|
FROM {database:Identifier}.http
|
|
GROUP BY (conn_type, hash, zeek_uid)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createDNSTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.dns (
|
|
import_time DateTime(),
|
|
zeek_uid FixedString(16),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
transaction_id UInt64,
|
|
round_trip_time Float64,
|
|
query String,
|
|
query_class_code UInt64,
|
|
query_class_name LowCardinality(String),
|
|
query_type_code UInt64,
|
|
query_type_name LowCardinality(String),
|
|
response_code UInt64,
|
|
response_code_name LowCardinality(String),
|
|
authoritative_answer Bool,
|
|
recursion_desired Bool,
|
|
recursion_available Bool,
|
|
z UInt64,
|
|
answers Array(String),
|
|
ttls Array(UInt32), -- TODO: should we change this to Array(Float64)?
|
|
rejected Bool
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, query, dst, hash)
|
|
ORDER BY (dst_nuid, src_nuid, src, query, dst, hash, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createUDNSTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.udns (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
tld String,
|
|
fqdn String,
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
visits AggregateFunction(count, UInt64),
|
|
first_seen AggregateFunction(min, DateTime()),
|
|
last_seen AggregateFunction(max, DateTime())
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (hour, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
ORDER BY (hour, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.udns_mv
|
|
TO {database:Identifier}.udns AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
cutToFirstSignificantSubdomain(query) as tld,
|
|
query as fqdn,
|
|
src,
|
|
dst,
|
|
src_nuid,
|
|
dst_nuid,
|
|
src_port,
|
|
dst_port,
|
|
src_local,
|
|
dst_local,
|
|
countState() as visits,
|
|
minState(ts) as first_seen,
|
|
maxState(ts) as last_seen
|
|
FROM {database:Identifier}.dns
|
|
GROUP BY (import_hour, hour, tld, fqdn, src, src_nuid, src_port, dst_port, src_local, dst_local, dst, dst_nuid, hash)
|
|
`)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
func (db *DB) createPDNSRawTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.pdns_raw (
|
|
import_time DateTime(),
|
|
hash FixedString(16),
|
|
ts DateTime(),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_port UInt32,
|
|
dst_port UInt32,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
transaction_id UInt64,
|
|
round_trip_time Float64,
|
|
query String,
|
|
query_class_code UInt64,
|
|
query_class_name LowCardinality(String),
|
|
query_type_code UInt64,
|
|
query_type_name LowCardinality(String),
|
|
response_code UInt64,
|
|
response_code_name LowCardinality(String),
|
|
authoritative_answer Bool,
|
|
recursion_desired Bool,
|
|
recursion_available Bool,
|
|
z UInt64,
|
|
resolved_ip IPv6,
|
|
ttls Array(UInt32),
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (dst_nuid, src_nuid, src, query, dst, hash)
|
|
ORDER BY (dst_nuid, src_nuid, src, query, dst, hash, ts)
|
|
`)
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createPDNSTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.pdns (
|
|
import_day DateTime(),
|
|
hash FixedString(16),
|
|
day DateTime(),
|
|
tld String,
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
src_local Bool,
|
|
dst_local Bool,
|
|
fqdn String,
|
|
resolved_ip IPv6,
|
|
first_seen AggregateFunction(min, DateTime()),
|
|
last_seen AggregateFunction(max, DateTime())
|
|
)
|
|
ENGINE = MergeTree()
|
|
PRIMARY KEY (day, tld, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.pdns_mv
|
|
TO {database:Identifier}.pdns AS
|
|
SELECT
|
|
toStartOfDay(import_time) as import_day,
|
|
toStartOfDay(ts) as day,
|
|
cutToFirstSignificantSubdomain(query) as tld,
|
|
query as fqdn,
|
|
resolved_ip,
|
|
src,
|
|
src_nuid,
|
|
dst,
|
|
dst_nuid,
|
|
hash,
|
|
minState(ts) as first_seen,
|
|
maxState(ts) as last_seen
|
|
FROM {database:Identifier}.pdns_raw
|
|
GROUP BY (import_day, day, tld, fqdn, resolved_ip, src, src_nuid, dst, dst_nuid, hash)
|
|
`)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
func (db *DB) createExplodedDNSTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.exploded_dns (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
tld String,
|
|
fqdn String,
|
|
subdomains AggregateFunction(uniqExact, String),
|
|
visits AggregateFunction(count, UInt64)
|
|
) ENGINE = AggregatingMergeTree()
|
|
PRIMARY KEY (hour, tld, fqdn)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.exploded_dns_mv
|
|
TO {database:Identifier}.exploded_dns AS
|
|
SELECT
|
|
import_hour,
|
|
hour,
|
|
tld,
|
|
uniqExactState(p.fqdn) as subdomains,
|
|
countMergeState(visits) as visits, -- stores the intermediate state of the visits from the udns AggregateFunction
|
|
t.exploded_dns as fqdn
|
|
FROM (SELECT import_hour, hour, fqdn, tld, visits FROM {database:Identifier}.udns) as p
|
|
LEFT JOIN (
|
|
-- join the different parts of the array with dots and reverse it again so that it's not backwards
|
|
SELECT DISTINCT fqdn, reverse(arrayStringConcat(exploded, '.')) as exploded_dns FROM (
|
|
-- for each part of the fqdn, create a new array of parts of the fqdn starting from the number of levels in the TLD until the end
|
|
SELECT fqdn, arrayJoin(arrayMap(i -> arraySlice(d, 1, i+(tld_levels)), range(length(d)))) as exploded FROM (
|
|
-- reverse fqdn and split on dots, get number of parts of the TLD
|
|
SELECT fqdn, splitByChar('.', reverse(fqdn)) as d, length(splitByChar('.', tld)) as tld_levels FROM (
|
|
-- grab each unique fqdn
|
|
SELECT DISTINCT fqdn, tld FROM {database:Identifier}.udns
|
|
WHERE tld != '' AND NOT endsWith(tld, '.arpa') AND NOT endsWith(tld, '.local')
|
|
)
|
|
)
|
|
)
|
|
) as t
|
|
ON p.fqdn = t.fqdn
|
|
WHERE tld != '' AND NOT endsWith(tld, '.arpa') AND NOT endsWith(tld, '.local')
|
|
GROUP BY (import_hour, hour, t.exploded_dns, tld)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// create temp table
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.dns_tmp (
|
|
tld String,
|
|
count AggregateFunction(count, UInt64)
|
|
) ENGINE = AggregatingMergeTree()
|
|
PRIMARY KEY (tld)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.dns_tmp_mv
|
|
TO {database:Identifier}.dns_tmp AS
|
|
SELECT
|
|
cutToFirstSignificantSubdomain(query) AS tld,
|
|
countState() as count
|
|
FROM {database:Identifier}.dns
|
|
GROUP BY (tld)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (db *DB) createIPToHostnameTable(ctx context.Context) error {
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.ip_to_hostname (
|
|
import_time DateTime(),
|
|
import_id FixedString(16),
|
|
hostname String,
|
|
ip IPv6,
|
|
ttl Int32, -- RFC2181 defines TTLs as int32
|
|
) ENGINE = MergeTree()
|
|
PRIMARY KEY (import_id, hostname, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createSensorDBTables() error {
|
|
ctx := db.QueryParameters(clickhouse.Parameters{
|
|
"database": db.selected,
|
|
})
|
|
|
|
if err := db.createConnTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := db.createSSLTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := db.createHTTPTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := db.createOpenConnTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := db.createOpenSSLTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := db.createOpenHTTPTmpTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.createConnTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.createUconnTable(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
err := db.createUconnTmpImportTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createOpenConnTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createHTTPTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createOpenHTTPTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createSSLTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createOpenSSLTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createUSNIConnTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createTLSProtoTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createHTTPProtoTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createSNIConnTmpImportTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createOpenConnHashTmpImportTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createOpenSNIConnTmpImportTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createBigOlHistogramTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createDNSTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createUDNSTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createPDNSRawTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createPDNSTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createExplodedDNSTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createIPToHostnameTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.createMinMaxMaterializedView(); err != nil {
|
|
return err
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (server *ServerConn) CreateServerDBTables() error {
|
|
err := server.createMetaDatabase()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|