activecm-rita/database/analysis_tables.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 Float64,
			beacon_threat_score Float64,
			ts_score Float64,
			ds_score Float64,
			dur_score Float64,
			hist_score Float64,
			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 Float64,

			-- STROBE
			strobe_score Float64,

			-- C2 OVER DNS
			subdomain_count UInt64,
			c2_over_dns_score Float64,
			c2_over_dns_direct_conn_score Float64,

			-- THREAT INTEL
			threat_intel Bool,
			threat_intel_score Float64,

			-- **** MODIFIERS ****
			modifier_name LowCardinality(String),
			modifier_score Float64,
			modifier_value String,

			-- PREVALENCE
			prevalence_total UInt64,
			prevalence Float64,
			prevalence_score Float64,
			network_size UInt64,

			first_seen_historical DateTime(),
			first_seen_score Float64,

			-- THREAT INTEL DATA SIZE
			threat_intel_data_size_score Float64,


			-- MISSING HOST HEADER
			missing_host_count UInt64,
			missing_host_header_score Float64

		) 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) 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
		WHERE host != ''
		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
		WHERE host != ''
		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.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
}