Update first seen scoring for non-rolling datasets

Co-Authored-By: Liza Tsibur <[email protected]>

.env

@@ -9,3 +9,5 @@ CONFIG_DIR=./deployment
 CONFIG_FILE=./config.hjson
 SYSLOG_ADDRESS=localhost:514
 DB_ADDRESS=localhost:9000
+LOGGING_ENABLED=true
+LOG_LEVEL=1

.env.production

@@ -9,3 +9,6 @@ CONFIG_FILE=/etc/rita/config.hjson
 SYSLOG_ADDRESS=syslogng:5514
 APP_LOGS=/var/log/rita
 DB_ADDRESS=db:9000
+LOGGING_ENABLED=true
+LOG_LEVEL=1
+

analysis/analysis.go

@@ -283,10 +283,14 @@ func (analyzer *Analyzer) runAnalysis() error {
 			timeSince := relativeTime.Sub(entry.FirstSeenHistorical)
 			daysSinceFirstSeen := float32(timeSince.Hours() / 24)
 
-			if daysSinceFirstSeen <= analyzer.Config.Modifiers.FirstSeenIncreaseThreshold {
-				mixtape.FirstSeenScore = analyzer.Config.Modifiers.FirstSeenScoreIncrease
-			} else if daysSinceFirstSeen >= analyzer.Config.Modifiers.FirstSeenDecreaseThreshold {
-				mixtape.FirstSeenScore = -1 * analyzer.Config.Modifiers.FirstSeenScoreDecrease
+			// Historical First Seen Scoring
+			// only apply to rolling datasets
+			if analyzer.Database.Rolling {
+				if daysSinceFirstSeen <= analyzer.Config.Modifiers.FirstSeenIncreaseThreshold {
+					mixtape.FirstSeenScore = analyzer.Config.Modifiers.FirstSeenScoreIncrease
+				} else if daysSinceFirstSeen >= analyzer.Config.Modifiers.FirstSeenDecreaseThreshold {
+					mixtape.FirstSeenScore = -1 * analyzer.Config.Modifiers.FirstSeenScoreDecrease
+				}
 			}
 
 			// Prevalence Scoring

config/config.go

@@ -130,9 +130,6 @@ type (
 		Modifiers Modifiers `json:"modifiers"`
 
 		ThreatIntel ThreatIntel `json:"threat_intel"`
-
-		LogLevel       int  `json:"log_level"`
-		LoggingEnabled bool `json:"logging_enabled"`
 	}
 )
 
@@ -473,11 +470,6 @@ func (cfg *Config) verifyConfig() error {
 		return fmt.Errorf("the MIME type/URI mismatch score increase must be between 0 and 1, got %v", cfg.Modifiers.MIMETypeMismatchScoreIncrease)
 	}
 
-	// validate log level
-	if cfg.LogLevel < -1 || cfg.LogLevel > 5 {
-		return fmt.Errorf("the LogLevel must be between -1 and 5 (inclusive)")
-	}
-
 	return nil
 }
 
@@ -651,7 +643,5 @@ func defaultConfig() Config {
 			OnlineFeeds:          []string{},
 			CustomFeedsDirectory: "/etc/rita/threat_intel_feeds",
 		},
-		LogLevel:       1,    // INFO level is default
-		LoggingEnabled: true, // enable logging by default
 	}
 }

config/config_test.go

@@ -136,8 +136,6 @@ func TestParseJSON(t *testing.T) {
 					c2_over_dns_direct_conn_score_increase: 0.9,
 					mime_type_mismatch_score_increase: 0.6
 				},
-				log_level: 3,
-				logging_enabled: false,
 			}
 			`),
 			expectedConfig: Config{
@@ -240,8 +238,6 @@ func TestParseJSON(t *testing.T) {
 					OnlineFeeds:          []string{"https://example.com/feed1", "https://example.com/feed2"},
 					CustomFeedsDirectory: "/path/to/custom/feeds",
 				},
-				LogLevel:       3,
-				LoggingEnabled: false,
 			},
 			expectedError: nil,
 		},
@@ -379,9 +375,6 @@ func TestParseJSON(t *testing.T) {
 			require.InDelta(test.expectedConfig.Modifiers.RareSignatureScoreIncrease, cfg.Modifiers.RareSignatureScoreIncrease, 0.00001, "RareSignatureScoreIncrease should match expected value")
 			require.InDelta(test.expectedConfig.Modifiers.C2OverDNSDirectConnScoreIncrease, cfg.Modifiers.C2OverDNSDirectConnScoreIncrease, 0.00001, "C2OverDNSDirectConnScoreIncrease should match expected value")
 			require.InDelta(test.expectedConfig.Modifiers.MIMETypeMismatchScoreIncrease, cfg.Modifiers.MIMETypeMismatchScoreIncrease, 0.00001, "MIMETypeMismatchScoreIncrease should match expected value")
-
-			require.Equal(test.expectedConfig.LogLevel, cfg.LogLevel, "LogLevel should match expected value")
-			require.Equal(test.expectedConfig.LoggingEnabled, cfg.LoggingEnabled, "LoggingEnabled should match expected value")
 		})
 	}
 }
@@ -655,8 +648,6 @@ func TestGetDefaultConfig(t *testing.T) {
 	require.Equal(origConfigVar.Scoring, cfg.Scoring, "config scoring should match expected value")
 	require.Equal(origConfigVar.Modifiers, cfg.Modifiers, "config modifiers should match expected value")
 	require.Equal(origConfigVar.ThreatIntel, cfg.ThreatIntel, "config threat intel should match expected value")
-	require.Equal(origConfigVar.LogLevel, cfg.LogLevel, "config log level should match expected value")
-	require.Equal(origConfigVar.LoggingEnabled, cfg.LoggingEnabled, "config logging enabled should match expected value")
 
 	// match the whole object just in case
 	require.Equal(origConfigVar, cfg, "config should match expected value")

database/analysis_tables.go

@@ -257,13 +257,16 @@ func (db *DB) createRareSignatureTable(ctx context.Context) error {
 			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 (src_nuid, src, signature )
+		PRIMARY KEY (src_nuid, src, dst, dst_nuid, fqdn, signature )
 	`)
 
 	if err != nil {
@@ -278,13 +281,14 @@ func (db *DB) createRareSignatureTable(ctx context.Context) error {
 			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
-		GROUP BY (import_hour, hour, src, src_nuid, signature, is_ja3)
+		GROUP BY (import_hour, hour, src, src_nuid, fqdn, signature, is_ja3)
 	`)
 
 	if err != nil {
@@ -299,13 +303,14 @@ func (db *DB) createRareSignatureTable(ctx context.Context) error {
 			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, signature, is_ja3)
+		GROUP BY (import_hour, hour, src, src_nuid, fqdn, signature, is_ja3)
 	`)
 	if err != nil {
 		return err
@@ -319,12 +324,14 @@ func (db *DB) createRareSignatureTable(ctx context.Context) error {
 			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, signature, is_ja3)
+		GROUP BY (import_hour, hour, src, src_nuid, dst, dst_nuid, signature, is_ja3)
 	`)
 	if err != nil {
 		return err
@@ -333,7 +340,6 @@ func (db *DB) createRareSignatureTable(ctx context.Context) error {
 	return err
 
 }
-
 func (db *DB) createPortInfoTable(ctx context.Context) error {
 
 	if err := db.Conn.Exec(ctx, `--sql

default_config.hjson

@@ -1,7 +1,5 @@
 {
     update_check_enabled: true,
-    log_level: 1,
-    logging_enabled: true,
     threat_intel: {
         // Configuration for custom threat intel feeds
         // Allowed format for the contents of both online feeds and custom file feeds is one IP or domain per line

importer/importer.go

@@ -28,7 +28,7 @@ import (
 	"golang.org/x/time/rate"
 )
 
-var errAllFilesPreviouslyImported = errors.New("all files were previously imported")
+var ErrAllFilesPreviouslyImported = errors.New("all files were previously imported")
 
 type zeekRecord interface {
 	zeektypes.Conn | zeektypes.DNS | zeektypes.HTTP | zeektypes.SSL
@@ -221,7 +221,7 @@ func (importer *Importer) Import(afs afero.Fs, files map[string][]string) error
 
 	// verify that there are still files left to import and set file count
 	if totalFileCount < 1 {
-		return errAllFilesPreviouslyImported
+		return ErrAllFilesPreviouslyImported
 	}
 	importer.TotalFileCount = totalFileCount
 

integration/analysis_view_test.go

@@ -1053,6 +1053,26 @@ func (it *ValidDatasetTestSuite) TestThreatMixtape() {
 	require.NoError(t, err)
 	require.EqualValues(t, 4668, count, "threat mixtape should have one non-modifier row per unique hash, got: %d", count)
 
+	err = it.db.Conn.QueryRow(it.db.GetContext(), `
+		SELECT count() FROM (
+			SELECT hash, count() as c FROM threat_mixtape
+			WHERE modifier_name = 'rare_signature'
+			GROUP BY hash
+		) WHERE c > 1
+	`).Scan(&count)
+	require.NoError(t, err)
+	require.EqualValues(t, 0, count, "threat mixtape should have at most one rare_signature modifier row per unique hash, got: %d", count)
+
+	err = it.db.Conn.QueryRow(it.db.GetContext(), `
+		SELECT count() FROM (
+			SELECT hash, count() as c FROM threat_mixtape
+			WHERE modifier_name = 'mime_type_mismatch'
+			GROUP BY hash
+		) WHERE c > 1
+	`).Scan(&count)
+	require.NoError(t, err)
+	require.EqualValues(t, 0, count, "threat mixtape should have at most one mime_type_mismatch modifier row per unique hash, got: %d", count)
+
 	err = it.db.Conn.QueryRow(it.db.GetContext(), `
 		SELECT count() FROM threat_mixtape
 		WHERE beacon_type != 'dns' AND count != open_count
@@ -1086,10 +1106,10 @@ func (it *ValidDatasetTestSuite) TestThreatMixtape() {
 
 	err = it.db.Conn.QueryRow(chCtx, `
 		SELECT count() FROM threat_mixtape
-		WHERE modifier_name = '' AND first_seen_score != {first_seen_increase_score:Float32}
+		WHERE first_seen_score != 0
 	`).Scan(&count)
 	require.NoError(t, err)
-	require.EqualValues(t, 0, count, "no non-modifier entries should have a historical first seen score other than the increase score")
+	require.EqualValues(t, 0, count, "no entries should have a historical first seen score for a non-rolling dataset")
 
 	err = it.db.Conn.QueryRow(chCtx, `
 		SELECT count(DISTINCT import_id) FROM threat_mixtape

integration/db_test.go

@@ -49,7 +49,7 @@ func (it *ValidDatasetTestSuite) TestPrimaryKeySize() {
 		"openconnhash_tmp": {NumParts: 2, TotalMarks: 40, AvgMarks: 40, TotalPrimaryKeySize: 2000, CompressionRatio: 0.4},
 		"ssl":              {NumParts: 2, TotalMarks: 20, AvgMarks: 20, TotalPrimaryKeySize: 1500, CompressionRatio: 0.7},
 		"ssl_tmp":          {NumParts: 2, TotalMarks: 20, AvgMarks: 20, TotalPrimaryKeySize: 1500, CompressionRatio: 0.7},
-		"threat_mixtape":   {NumParts: 4, TotalMarks: 10, AvgMarks: 5, TotalPrimaryKeySize: 700, CompressionRatio: 0.75},
+		"threat_mixtape":   {NumParts: 4, TotalMarks: 10, AvgMarks: 5, TotalPrimaryKeySize: 700, CompressionRatio: 0.7},
 		"tls_proto":        {NumParts: 2, TotalMarks: 10, AvgMarks: 10, TotalPrimaryKeySize: 600, CompressionRatio: 0.7},
 		"uconn":            {NumParts: 2, TotalMarks: 15, AvgMarks: 15, TotalPrimaryKeySize: 800, CompressionRatio: 0.5},
 		"uconn_tmp":        {NumParts: 2, TotalMarks: 60, AvgMarks: 60, TotalPrimaryKeySize: 2000, CompressionRatio: 0.35},

integration/missing_host_test.go

@@ -107,7 +107,7 @@ func (it *MissingHostSuite) TestThreat() {
 	}
 	expected := expectedResults{
 		src: "73.54.23.243", dst: "64.225.56.201",
-		finalScore: 1.05, beaconScore: 0.963, longConnScore: config.HIGH_CATEGORY_SCORE,
+		finalScore: 0.9000, beaconScore: 0.963, longConnScore: config.HIGH_CATEGORY_SCORE,
 		totalDuration: 270036.631115, count: 331, missingHostCount: 330, missingHostScore: 0.1,
 		totalBytes: 1679442, portProtoService: []string{"80:tcp:http", "22:tcp:ssh"}, prevalence: 1,
 		firstSeen: time.Unix(1713470872, 0).UTC(),
@@ -162,7 +162,7 @@ func (it *MissingHostSuite) TestThreat() {
 		require.InDelta(t, expected.missingHostScore, res.MissingHostHeaderScore, 0.001, "missing host score should match")
 		require.InDelta(t, expected.prevalence, res.Prevalence, 0.001, "prevalence should match")
 		require.InDelta(t, -it.cfg.Modifiers.PrevalenceScoreDecrease, res.PrevalenceScore, 0.001, "prevalence score should equal the prevalence decrease config value")
-		require.InDelta(t, it.cfg.Modifiers.FirstSeenScoreIncrease, res.FirstSeenScore, 0.001, "first seen score should equal the first seen score increase config value")
+		require.EqualValues(t, 0, res.FirstSeenScore, 0.001, "first seen score should equal 0 for a non-rolling dataset")
 		require.EqualValues(t, expected.firstSeen.UTC(), res.FirstSeen, "first seen date should match")
 		require.InDelta(t, it.cfg.Modifiers.MissingHostCountScoreIncrease, res.MissingHostHeaderScore, 0.001, "missing host header score should equal the missing host header increase score config value")
 		require.InDelta(t, it.cfg.Modifiers.RareSignatureScoreIncrease, res.TotalModifierScore, 0.001, "total modifier score should equal the rare signature increase score config value")

integration/modifiers_test.go

@@ -0,0 +1,26 @@
+package integration_test
+
+import "github.com/stretchr/testify/require"
+
+func (it *ValidDatasetTestSuite) TestRareSignaturesModifier() {
+	t := it.T()
+	var count uint64
+	err := it.db.Conn.QueryRow(it.db.GetContext(), `
+		WITH mixtape AS (
+			SELECT DISTINCT src, src_nuid, dst, dst_nuid, fqdn, modifier_value 
+			FROM threat_mixtape 
+			WHERE modifier_name = 'rare_signature'
+		), rare_sigs AS (
+			SELECT src, src_nuid, signature, uniqExactMerge(times_used_dst) as times_used_dst, uniqExactMerge(times_used_fqdn) as times_used_fqdn
+			FROM rare_signatures
+			GROUP BY src, src_nuid, signature
+			HAVING times_used_dst = 1 OR times_used_fqdn = 1
+		)
+		SELECT count() FROM mixtape m
+		LEFT JOIN rare_sigs r ON r.src = m.src AND m.src_nuid = r.src_nuid AND m.modifier_value = r.signature
+		WHERE (fqdn != '' AND times_used_fqdn != 1) OR (fqdn = '' AND times_used_dst != 1)
+	`).Scan(&count)
+	require.NoError(t, err)
+	require.Zero(t, count, "all rare signature entries in the mixtape should actually be used only once according to rare_signatures table")
+
+}

integration/open_sni_test.go

@@ -75,10 +75,11 @@ func (it *OpenSNITestSuite) TestThreats() {
 		prevalenceTotal  int64
 		portProtoService []string
 	}{
-		{src: "10.0.0.238", dst: "::", fqdn: "ce7.stearns.org", finalScore: .40468138, totalDuration: 14737.061150000001, count: 0, proxyCount: 0, openCount: 2, totalBytes: 24106, serverIPs: []net.IP{sslServerIP}, portProtoService: []string{"8443:tcp:ssl"}},
-		{src: "10.0.0.238", dst: "::", fqdn: "ce7.stearns.org:8000", finalScore: 0.26993924, totalDuration: 7376.718848, count: 0, proxyCount: 0, openCount: 1, totalBytes: 8144, serverIPs: []net.IP{sslServerIP}, portProtoService: []string{"8000:tcp:http"}},
-		{src: "10.0.0.238", dst: "34.222.122.143", finalScore: 0.26667413, totalDuration: 7200.403186, count: 0, proxyCount: 0, openCount: 1, totalBytes: 2715618, portProtoService: []string{"64590:tcp:"}},
-		{src: "10.0.0.238", dst: "52.33.59.39", finalScore: 0.26667413, totalDuration: 7200.169165, count: 0, proxyCount: 0, openCount: 1, totalBytes: 4593763, portProtoService: []string{"64004:tcp:"}},
+
+		{src: "10.0.0.238", dst: "::", fqdn: "ce7.stearns.org", finalScore: 0.25468, totalDuration: 14737.061150000001, count: 0, proxyCount: 0, openCount: 2, totalBytes: 24106, serverIPs: []net.IP{sslServerIP}, portProtoService: []string{"8443:tcp:ssl"}},
+		{src: "10.0.0.238", dst: "::", fqdn: "ce7.stearns.org:8000", finalScore: 0.11993, totalDuration: 7376.718848, count: 0, proxyCount: 0, openCount: 1, totalBytes: 8144, serverIPs: []net.IP{sslServerIP}, portProtoService: []string{"8000:tcp:http"}},
+		{src: "10.0.0.238", dst: "34.222.122.143", finalScore: 0.11667, totalDuration: 7200.403186, count: 0, proxyCount: 0, openCount: 1, totalBytes: 2715618, portProtoService: []string{"64590:tcp:"}},
+		{src: "10.0.0.238", dst: "52.33.59.39", finalScore: 0.11667, totalDuration: 7200.169165, count: 0, proxyCount: 0, openCount: 1, totalBytes: 4593763, portProtoService: []string{"64004:tcp:"}},
 	}
 
 	min, _, _, _, err := it.db.GetTrueMinMaxTimestamps()
@@ -115,7 +116,7 @@ func (it *OpenSNITestSuite) TestThreats() {
 		require.EqualValues(t, 2024, year, "first seen year should match")
 		require.EqualValues(t, 01, month, "first seen month should match")
 		require.EqualValues(t, 31, day, "first seen day should match")
-		require.InDelta(t, it.cfg.Modifiers.FirstSeenScoreIncrease, res.FirstSeenScore, 0.001, "first seen score should equal config increase value")
+		require.InDelta(t, 0, res.FirstSeenScore, 0.001, "first seen score should equal 0 for a non-rolling dataset")
 		i++
 	}
 	rows.Close()

integration/proxy_rolling_test.go

@@ -81,7 +81,7 @@ func (it *ProxyRollingTestSuite) TestRollingThreats() {
 	}
 	var results []res
 
-	expectedCounts := []uint64{4, 62}
+	expectedCounts := []uint64{3, 41}
 
 	err := it.db.Conn.Select(it.db.GetContext(), &results, `
 		SELECT analyzed_at, count() as c FROM threat_mixtape
@@ -110,10 +110,8 @@ func (it *ProxyRollingTestSuite) TestRollingThreats() {
 		proxyCount    uint64
 		proxyIPs      []net.IP
 	}{
-		// finalScore: 0.4492
-		// beaconScore: 0.624
-		{src: "10.0.0.111", dst: "::", fqdn: "safebrowsing.googleapis.com:443", finalScore: 0.4492, beaconScore: 0.624, totalDuration: 6.569, count: 46, proxyCount: 46, totalBytes: 308421, proxyIPs: []net.IP{proxyIP}},
-		{src: "10.0.0.238", dst: "75.75.75.75", finalScore: 0.3383, beaconScore: 0.673, totalDuration: 595.72157, count: 1160, totalBytes: 319107},
+		{src: "10.0.0.238", dst: "75.75.75.75", finalScore: 0.18839, beaconScore: 0.673, totalDuration: 595.72157, count: 1160, totalBytes: 319107},
+		{src: "10.0.0.111", dst: "::", fqdn: "safebrowsing.googleapis.com:443", finalScore: 0.1492, beaconScore: 0.624, totalDuration: 6.569, count: 46, proxyCount: 46, totalBytes: 308421, proxyIPs: []net.IP{proxyIP}},
 	}
 
 	min, _, _, err := it.db.GetBeaconMinMaxTimestamps()