mirror of
https://github.com/activecm/rita
synced 2026-06-08 13:02:45 +00:00
Naomi Kramer
752 lines
21 KiB
Go
752 lines
21 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
|
|
clickhouse "github.com/ClickHouse/clickhouse-go/v2"
|
|
)
|
|
|
|
func (db *DB) createThreatMixtapeTable(ctx context.Context) error {
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.threat_mixtape (
|
|
analyzed_at DateTime64(6),
|
|
import_id FixedString(16),
|
|
hash FixedString(16),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
fqdn String,
|
|
server_ips Array(IPv6),
|
|
proxy_ips Array(IPv6),
|
|
total_bytes UInt64,
|
|
last_seen DateTime(),
|
|
port_proto_service Array(String),
|
|
|
|
-- counts
|
|
count UInt64,
|
|
ts_unique UInt64,
|
|
proxy_count UInt64,
|
|
open_count UInt64,
|
|
|
|
-- **** THREAT INDICATORS ****
|
|
-- BEACONING
|
|
beacon_type LowCardinality(String),
|
|
beacon_score Float32,
|
|
beacon_threat_score Float32,
|
|
ts_score Float32,
|
|
ds_score Float32,
|
|
dur_score Float32,
|
|
hist_score Float32,
|
|
ts_intervals Array(Int64),
|
|
ts_interval_counts Array(Int64),
|
|
ds_sizes Array(Int64),
|
|
ds_size_counts Array(Int64),
|
|
|
|
-- LONG CONNECTIONS
|
|
total_duration Float64,
|
|
long_conn_score Float32,
|
|
|
|
-- STROBE
|
|
strobe_score Float32,
|
|
|
|
-- C2 OVER DNS
|
|
subdomain_count UInt64,
|
|
c2_over_dns_score Float32,
|
|
c2_over_dns_direct_conn_score Float32,
|
|
|
|
-- THREAT INTEL
|
|
threat_intel Bool,
|
|
threat_intel_score Float32,
|
|
|
|
-- **** MODIFIERS ****
|
|
modifier_name LowCardinality(String),
|
|
modifier_score Float32,
|
|
modifier_value String,
|
|
|
|
-- PREVALENCE
|
|
prevalence_total UInt64,
|
|
prevalence Float32,
|
|
prevalence_score Float32,
|
|
network_size UInt64,
|
|
|
|
first_seen_historical DateTime(),
|
|
first_seen_score Float32,
|
|
|
|
-- THREAT INTEL DATA SIZE
|
|
threat_intel_data_size_score Float32,
|
|
|
|
|
|
-- MISSING HOST HEADER
|
|
missing_host_count UInt64,
|
|
missing_host_header_score Float32
|
|
|
|
) ENGINE = MergeTree()
|
|
PRIMARY KEY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
ORDER BY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
`)
|
|
return err
|
|
}
|
|
|
|
// func (db *DB) createThreatMixtapeView(ctx context.Context) error {
|
|
|
|
// if err := db.Conn.Exec(ctx, `
|
|
// CREATE TABLE IF NOT EXISTS {database:Identifier}.final_mixtape (
|
|
// analyzed_at DateTime64(6),
|
|
// import_id FixedString(16),
|
|
// hash FixedString(16),
|
|
// src IPv6,
|
|
// dst IPv6,
|
|
// src_nuid UUID,
|
|
// dst_nuid UUID,
|
|
// fqdn String,
|
|
// server_ips AggregateFunction(groupArrayArray, Array(IPv6)),
|
|
// proxy_ips AggregateFunction(groupArrayArray, Array(IPv6)),
|
|
// total_bytes AggregateFunction(sum, UInt64),
|
|
// last_seen AggregateFunction(max, DateTime()),
|
|
// port_proto_service AggregateFunction(groupArrayArray, Array(String)),
|
|
|
|
// -- counts
|
|
// count AggregateFunction(sum, UInt64),
|
|
// ts_unique AggregateFunction(sum, UInt64),
|
|
// proxy_count AggregateFunction(sum, UInt64),
|
|
// open_count AggregateFunction(sum, UInt64),
|
|
|
|
// -- **** THREAT INDICATORS ****
|
|
// -- BEACONING
|
|
// beacon_type LowCardinality(String),
|
|
// beacon_score AggregateFunction(sum, Float32),
|
|
// beacon_threat_score AggregateFunction(sum, Float32),
|
|
// ts_score AggregateFunction(sum, Float32),
|
|
// ds_score AggregateFunction(sum, Float32),
|
|
// dur_score AggregateFunction(sum, Float32),
|
|
// hist_score AggregateFunction(sum, Float32),
|
|
// ts_intervals AggregateFunction(groupArrayArray, Array(Int64)),
|
|
// ts_interval_counts AggregateFunction(groupArrayArray, Array(Int64)),
|
|
// ds_sizes AggregateFunction(groupArrayArray, Array(Int64)),
|
|
// ds_size_counts AggregateFunction(groupArrayArray, Array(Int64)),
|
|
|
|
// -- LONG CONNECTIONS
|
|
// total_duration AggregateFunction(sum, Float64),
|
|
// long_conn_score AggregateFunction(sum, Float32),
|
|
|
|
// -- STROBE
|
|
// strobe_score AggregateFunction(sum, Float32),
|
|
|
|
// -- C2 OVER DNS
|
|
// subdomain_count AggregateFunction(sum, UInt64),
|
|
// c2_over_dns_score AggregateFunction(sum, Float32),
|
|
// c2_over_dns_direct_conn_score AggregateFunction(sum, Float32),
|
|
|
|
// -- THREAT INTEL
|
|
// threat_intel AggregateFunction(max, Bool),
|
|
// threat_intel_score AggregateFunction(sum, Float32),
|
|
|
|
// -- **** MODIFIERS ****
|
|
// -- modifier_name LowCardinality(String),
|
|
// -- modifier_score Float32,
|
|
// -- modifier_value String,
|
|
|
|
// -- PREVALENCE
|
|
// prevalence_total AggregateFunction(sum, UInt64),
|
|
// prevalence AggregateFunction(sum, Float32),
|
|
// prevalence_score AggregateFunction(sum, Float32),
|
|
// network_size AggregateFunction(sum, UInt64),
|
|
|
|
// first_seen_historical AggregateFunction(max, DateTime()),
|
|
// first_seen_score AggregateFunction(sum, Float32),
|
|
|
|
// -- THREAT INTEL DATA SIZE
|
|
// threat_intel_data_size_score AggregateFunction(sum, Float32),
|
|
|
|
// -- MISSING HOST HEADER
|
|
// missing_host_count AggregateFunction(sum, UInt64),
|
|
// missing_host_header_score AggregateFunction(sum, Float32)
|
|
|
|
// ) ENGINE = AggregatingMergeTree()
|
|
// -- FROM {database:Identifier}.threat_mixtape
|
|
// -- GROUP BY analyzed_at, import_id, hash, src, src_nuid, dst, dst_nuid, fqdn, beacon_type
|
|
// PRIMARY KEY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
// ORDER BY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
// `); err != nil {
|
|
// return err
|
|
// }
|
|
|
|
// if err := db.Conn.Exec(ctx, `--sql
|
|
// CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.final_mixtape_mv
|
|
// TO {database:Identifier}.final_mixtape AS
|
|
// SELECT
|
|
// analyzed_at,
|
|
// import_id,
|
|
// hash,
|
|
// src,
|
|
// dst,
|
|
// src_nuid,
|
|
// dst_nuid,
|
|
// fqdn,
|
|
// groupArrayArrayState(server_ips) as server_ips,
|
|
// groupArrayArrayState(proxy_ips) as proxy_ips,
|
|
// sumState(total_bytes) as total_bytes,
|
|
// maxState(last_seen) as last_seen,
|
|
// groupArrayArrayState(port_proto_service) as port_proto_service,
|
|
|
|
// -- counts
|
|
// sumState(count) as count,
|
|
// sumState(ts_unique) as ts_unique,
|
|
// sumState(proxy_count) as proxy_count,
|
|
// sumState(open_count) as open_count,
|
|
|
|
// -- **** THREAT INDICATORS ****
|
|
// -- BEACONING
|
|
// beacon_type,
|
|
// sumState(beacon_score) as beacon_score,
|
|
// sumState(beacon_threat_score) as beacon_threat_score,
|
|
// sumState(ts_score) as ts_score,
|
|
// sumState(ds_score) as ds_score,
|
|
// sumState(dur_score) as dur_score,
|
|
// sumState(hist_score) as hist_score,
|
|
// groupArrayArrayState(ts_intervals) as ts_intervals,
|
|
// groupArrayArrayState(ts_interval_counts) as ts_interval_counts,
|
|
// groupArrayArrayState(ds_sizes) as ds_sizes,
|
|
// groupArrayArrayState(ds_size_counts) as ds_size_counts,
|
|
|
|
// -- LONG CONNECTIONS
|
|
// sumState(total_duration) as total_duration,
|
|
// sumState(long_conn_score) as long_conn_score,
|
|
|
|
// -- STROBE
|
|
// sumState(strobe_score) as strobe_score,
|
|
|
|
// -- C2 OVER DNS
|
|
// sumState(subdomain_count) as subdomain_count,
|
|
// sumState(c2_over_dns_score) as c2_over_dns_score,
|
|
// sumState(c2_over_dns_direct_conn_score) as c2_over_dns_direct_conn_score,
|
|
|
|
// -- THREAT INTEL
|
|
// maxState(threat_intel) as threat_intel,
|
|
// sumState(threat_intel_score) as threat_intel_score,
|
|
|
|
// -- -- **** MODIFIERS ****
|
|
// -- modifier_name LowCardinality(String),
|
|
// -- modifier_score Float32,
|
|
// -- modifier_value String,
|
|
|
|
// -- PREVALENCE
|
|
// sumState(prevalence_total) as prevalence_total,
|
|
// sumState(prevalence) as prevalence,
|
|
// sumState(prevalence_score) as prevalence_score,
|
|
// sumState(network_size) as network_size,
|
|
|
|
// maxState(first_seen_historical) as first_seen_historical,
|
|
// sumState(first_seen_score) as first_seen_score,
|
|
|
|
// -- THREAT INTEL DATA SIZE
|
|
// sumState(threat_intel_data_size_score) as threat_intel_data_size_score,
|
|
|
|
// -- MISSING HOST HEADER
|
|
// sumState(missing_host_count) as missing_host_count,
|
|
// sumState(missing_host_header_score) as missing_host_header_score
|
|
// FROM {database:Identifier}.threat_mixtape
|
|
// GROUP BY analyzed_at, import_id, hash, src, src_nuid, dst, dst_nuid, fqdn, beacon_type
|
|
// -- PRIMARY KEY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
// -- ORDER BY (analyzed_at, dst_nuid, src_nuid, src, fqdn, dst, hash)
|
|
// `); err != nil {
|
|
// return err
|
|
// }
|
|
// return nil
|
|
// }
|
|
|
|
func (db *DB) createFinalMixtape(ctx context.Context) error {
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.final_mixtape (
|
|
analyzed_at DateTime64(6),
|
|
import_id FixedString(16),
|
|
hash FixedString(16),
|
|
src IPv6,
|
|
dst IPv6,
|
|
src_nuid UUID,
|
|
dst_nuid UUID,
|
|
fqdn String,
|
|
|
|
|
|
base_score Float32,
|
|
total_modifier_score Float32,
|
|
final_score Float32,
|
|
|
|
|
|
server_ips Array(IPv6),
|
|
proxy_ips Array(IPv6),
|
|
total_bytes UInt64,
|
|
last_seen DateTime(),
|
|
port_proto_service Array(String),
|
|
|
|
-- counts
|
|
count UInt64,
|
|
ts_unique UInt64,
|
|
proxy_count UInt64,
|
|
open_count UInt64,
|
|
|
|
-- **** THREAT INDICATORS ****
|
|
-- BEACONING
|
|
beacon_type LowCardinality(String),
|
|
beacon_score Float32,
|
|
beacon_threat_score Float32,
|
|
ts_score Float32,
|
|
ds_score Float32,
|
|
dur_score Float32,
|
|
hist_score Float32,
|
|
-- ts_intervals Array(Int64),
|
|
-- ts_interval_counts Array(Int64),
|
|
-- ds_sizes Array(Int64),
|
|
-- ds_size_counts Array(Int64),
|
|
|
|
connection_graph_intervals Array(Int64),
|
|
connection_graph_counts Array(Int64),
|
|
data_size_graph_intervals Array(Int64),
|
|
data_size_graph_counts Array(Int64),
|
|
|
|
-- LONG CONNECTIONS
|
|
total_duration Float64,
|
|
long_conn_score Float32,
|
|
|
|
-- STROBE
|
|
strobe_score Float32,
|
|
|
|
-- C2 OVER DNS
|
|
subdomain_count UInt64,
|
|
c2_over_dns_score Float32,
|
|
c2_over_dns_direct_conn_score Float32,
|
|
|
|
-- THREAT INTEL
|
|
threat_intel Bool,
|
|
threat_intel_score Float32,
|
|
|
|
-- -- **** MODIFIERS ****
|
|
-- modifier_name LowCardinality(String),
|
|
-- modifier_score Float32,
|
|
-- modifier_value String,
|
|
|
|
-- PREVALENCE
|
|
prevalence_total UInt64,
|
|
prevalence Float32,
|
|
prevalence_score Float32,
|
|
network_size UInt64,
|
|
|
|
first_seen_historical DateTime(),
|
|
first_seen_score Float32,
|
|
|
|
-- THREAT INTEL DATA SIZE
|
|
threat_intel_data_size_score Float32,
|
|
|
|
|
|
-- MISSING HOST HEADER
|
|
missing_host_count UInt64,
|
|
missing_host_header_score Float32
|
|
|
|
) ENGINE = MergeTree
|
|
PRIMARY KEY (hash, src, src_nuid, dst_nuid, dst, fqdn)
|
|
ORDER BY (hash, src, src_nuid, dst_nuid, dst, fqdn, final_score)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createHistoricalFirstSeenMaterializedViews(ctx context.Context) error {
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_conn_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
if(src_local = true, dst, src) as ip,
|
|
'' as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.conn
|
|
GROUP BY (fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_openconn_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
if(src_local = true, dst, src) as ip,
|
|
'' as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.openconn
|
|
GROUP BY (fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_ssl_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
'::' as ip,
|
|
server_name as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.ssl
|
|
GROUP BY ( fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_openssl_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
'::' as ip,
|
|
server_name as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.openssl
|
|
GROUP BY ( fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_http_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
'::' as ip,
|
|
host as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.http
|
|
GROUP BY ( fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_openhttp_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
'::' as ip,
|
|
host as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.openhttp
|
|
GROUP BY ( fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.historical_first_seen_dns_mv
|
|
TO metadatabase.historical_first_seen AS
|
|
SELECT
|
|
'::' as ip,
|
|
query as fqdn,
|
|
minSimpleState(ts) as first_seen,
|
|
maxSimpleState(ts) as last_seen
|
|
FROM {database:Identifier}.dns
|
|
GROUP BY ( fqdn, ip)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createMIMETypeURIsTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.mime_type_uris (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
uri String,
|
|
path String,
|
|
extension String,
|
|
mime_type String,
|
|
mismatch_count AggregateFunction(count, UInt64),
|
|
)
|
|
ENGINE = AggregatingMergeTree()
|
|
PRIMARY KEY (hour, hash, uri)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// This view is used to detect MIME type/URI mismatches
|
|
// If a HTTP connection's MIME type matches a MIME type in the metadatabase.valid_mime_types table
|
|
// and its extension does not match the associated values for that MIME type, then it should be added to this table
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.mime_type_uris_mv
|
|
TO {database:Identifier}.mime_type_uris AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
uri,
|
|
path(uri) as path,
|
|
-- get the extension from the path
|
|
CASE
|
|
-- if the path does not contain a . or ends with a ., then the extension is an empty string
|
|
WHEN position(reverse(arrayElement(splitByChar('/', path), -1)), '.') = 0 OR endsWith(path, '.') THEN ''
|
|
-- otherwise, split the last segment by . and take the last element as the extension
|
|
ELSE splitByChar('.', arrayElement(splitByChar('/', path), -1))[-1]
|
|
END AS extension,
|
|
dst_mime_types as mime_type,
|
|
countState() AS mismatch_count
|
|
FROM {database:Identifier}.http h
|
|
-- for each uri, get the extension and join it with the valid mime types,
|
|
-- keeping only the rows where the extension does not match the valid extension
|
|
ARRAY JOIN dst_mime_types
|
|
LEFT SEMI JOIN metadatabase.valid_mime_types v ON dst_mime_types = v.mime_type
|
|
WHERE uri != '/' AND extension != v.extension
|
|
GROUP BY import_hour, hour, hash, uri, path, extension, mime_type
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) createRareSignatureTable(ctx context.Context) error {
|
|
|
|
err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.rare_signatures (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
src IPv6,
|
|
src_nuid UUID,
|
|
dst IPv6,
|
|
dst_nuid UUID,
|
|
fqdn String,
|
|
signature String,
|
|
is_ja3 Bool,
|
|
times_used_dst AggregateFunction(uniqExact, IPv6),
|
|
times_used_fqdn AggregateFunction(uniqExact, String)
|
|
)
|
|
ENGINE = AggregatingMergeTree()
|
|
PRIMARY KEY (hour, src_nuid, src, dst, dst_nuid, fqdn, signature )
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.rare_signatures_http_mv
|
|
TO {database:Identifier}.rare_signatures AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
src,
|
|
src_nuid,
|
|
host AS fqdn,
|
|
useragent as signature,
|
|
false as is_ja3,
|
|
uniqExactState(dst) as times_used_dst,
|
|
uniqExactState(host) as times_used_fqdn
|
|
FROM {database:Identifier}.http
|
|
WHERE length(useragent) > 0 AND length(host) > 0
|
|
GROUP BY (import_hour, hour, src, src_nuid, fqdn, signature, is_ja3)
|
|
`)
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.rare_signatures_ssl_mv
|
|
TO {database:Identifier}.rare_signatures AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
src,
|
|
src_nuid,
|
|
server_name AS fqdn,
|
|
ja3 as signature,
|
|
true as is_ja3,
|
|
uniqExactState(dst) as times_used_dst,
|
|
uniqExactState(server_name) as times_used_fqdn
|
|
FROM {database:Identifier}.ssl
|
|
WHERE length(ja3) > 0
|
|
GROUP BY (import_hour, hour, src, src_nuid, fqdn, signature, is_ja3)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.rare_signatures_missing_host_mv
|
|
TO {database:Identifier}.rare_signatures AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
src,
|
|
src_nuid,
|
|
dst,
|
|
dst_nuid,
|
|
missing_host_useragent as signature,
|
|
false as is_ja3,
|
|
uniqExactState(if(src_local, dst, src)) as times_used_dst
|
|
FROM {database:Identifier}.conn
|
|
WHERE length(missing_host_useragent) > 0 AND missing_host_header = true
|
|
GROUP BY (import_hour, hour, src, src_nuid, dst, dst_nuid, signature, is_ja3)
|
|
`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return err
|
|
|
|
}
|
|
func (db *DB) createPortInfoTable(ctx context.Context) error {
|
|
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE TABLE IF NOT EXISTS {database:Identifier}.port_info (
|
|
import_hour DateTime(),
|
|
hour DateTime(),
|
|
hash FixedString(16),
|
|
src IPv6,
|
|
src_nuid UUID,
|
|
dst IPv6,
|
|
dst_nuid UUID,
|
|
fqdn String,
|
|
dst_port UInt32,
|
|
proto LowCardinality(String),
|
|
service LowCardinality(String),
|
|
icmp_type Int64,
|
|
icmp_code Int64,
|
|
conn_state LowCardinality(String),
|
|
count AggregateFunction(count, UInt64),
|
|
bytes_sent AggregateFunction(sum, UInt64),
|
|
bytes_received AggregateFunction(sum, UInt64)
|
|
)
|
|
ENGINE = AggregatingMergeTree()
|
|
PRIMARY KEY (hour, hash, dst_port, proto, service, conn_state, icmp_type, icmp_code)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
// conn
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.port_info_ip_mv
|
|
TO {database:Identifier}.port_info AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
src,
|
|
src_nuid,
|
|
dst,
|
|
dst_nuid,
|
|
dst_port,
|
|
proto,
|
|
service,
|
|
if(proto = 'icmp', src_port, 0) as icmp_type,
|
|
if(proto = 'icmp', dst_port, 0) as icmp_code,
|
|
conn_state,
|
|
countState() as count,
|
|
sumState(src_ip_bytes) as bytes_sent,
|
|
sumState(dst_ip_bytes) as bytes_received
|
|
FROM {database:Identifier}.conn
|
|
WHERE missing_host_header = false
|
|
GROUP BY (import_hour, hour, hash, src, src_nuid, dst, dst_nuid, dst_port, proto, service, icmp_type, icmp_code, conn_state)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
// http
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.port_info_http_mv
|
|
TO {database:Identifier}.port_info AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
src,
|
|
src_nuid,
|
|
host as fqdn,
|
|
dst_port,
|
|
proto,
|
|
service,
|
|
conn_state,
|
|
countStateIf(multi_request = false) as count, -- only count unique zeek_uids, not each multi-request
|
|
sumState(src_ip_bytes) as bytes_sent,
|
|
sumState(dst_ip_bytes) as bytes_received
|
|
FROM {database:Identifier}.http
|
|
GROUP BY (import_hour, hour, hash, src, src_nuid, fqdn, dst_port, proto, service, conn_state)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
// ssl
|
|
if err := db.Conn.Exec(ctx, `--sql
|
|
CREATE MATERIALIZED VIEW IF NOT EXISTS {database:Identifier}.port_info_ssl_mv
|
|
TO {database:Identifier}.port_info AS
|
|
SELECT
|
|
toStartOfHour(import_time) as import_hour,
|
|
toStartOfHour(ts) as hour,
|
|
hash,
|
|
src,
|
|
src_nuid,
|
|
server_name as fqdn,
|
|
dst_port,
|
|
proto,
|
|
service,
|
|
conn_state,
|
|
countState() as count,
|
|
sumState(src_ip_bytes) as bytes_sent,
|
|
sumState(dst_ip_bytes) as bytes_received
|
|
FROM {database:Identifier}.ssl
|
|
GROUP BY (import_hour, hour, hash, src, src_nuid, fqdn, dst_port, proto, service, conn_state)
|
|
`); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (db *DB) CreateSensorDBAnalysisTables() error {
|
|
ctx := db.QueryParameters(clickhouse.Parameters{
|
|
"database": db.selected,
|
|
})
|
|
|
|
err := db.createThreatMixtapeTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createFinalMixtape(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createRareSignatureTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createMIMETypeURIsTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = db.createPortInfoTable(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// only create historical first seen mvs for rolling datasets
|
|
if db.Rolling {
|
|
err = db.createHistoricalFirstSeenMaterializedViews(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|