From fed0c5739d67266ec8027c4b8252761fdd123a4a Mon Sep 17 00:00:00 2001 From: defensivedepth Date: Thu, 8 Oct 2026 19:31:29 -0400 Subject: [PATCH] Refactor correlations --- .../files/modules/so/securityonion-es_test.py | 36 ++++++++++++------- .../component/so/detection-mappings.json | 26 +++++++------- salt/manager/tools/sbin/soup | 18 +++++++--- salt/soc/defaults.yaml | 4 +-- 4 files changed, 52 insertions(+), 32 deletions(-) diff --git a/salt/elastalert/files/modules/so/securityonion-es_test.py b/salt/elastalert/files/modules/so/securityonion-es_test.py index 8f4db168d..bfdce060f 100644 --- a/salt/elastalert/files/modules/so/securityonion-es_test.py +++ b/salt/elastalert/files/modules/so/securityonion-es_test.py @@ -18,7 +18,6 @@ es = importlib.util.module_from_spec(spec) spec.loader.exec_module(es) BASE_RULE = { - 'name': 'Many Failed Network Logons To One Host From One Source -- 35a42db6-6629-45af-b8aa-e1fa33c28ef5', 'detection_title': 'Many Failed Network Logons To One Host From One Source', 'detection_public_id': '35a42db6-6629-45af-b8aa-e1fa33c28ef5', 'sigma_level': 'medium', @@ -26,9 +25,6 @@ BASE_RULE = { 'event.severity': 3, 'event.module': 'sigma', 'event.dataset': 'sigma.alert', - 'es_host': 'manager', - 'es_port': 9200, - 'es_conn_timeout': 55, 'summary_template': '%count% failed network logons to %host.name% from %source.ip% in %duration%', } @@ -58,7 +54,6 @@ class TestSecurityOnionESAlerter(unittest.TestCase): alerter.alert([match]) calls = alerter.es.create.call_args_list self.assertTrue(all(c.kwargs['index'] == 'logs-detections.alerts-so' for c in calls)) - # as the client serializes it return [(json.loads(json.dumps(c.kwargs['body'], cls=es.DateTimeEncoder)), c.kwargs['id']) for c in calls] def send(self, rule, match): @@ -77,6 +72,7 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertNotIn('host.name,source.ip', payload['event_data']) self.assertEqual(payload['event_data']['host'], {'name': 'host-01'}) self.assertEqual(payload['event_data']['source'], {'ip': '192.0.2.10'}) + self.assertEqual(payload['rule'], {'name': BASE_RULE['detection_title'], 'uuid': BASE_RULE['detection_public_id'], 'correlation': 'event_count'}) self.assertEqual(payload['labels'], {'correlation_group_by': 'host.name, source.ip', 'correlation_group': 'host-01, 192.0.2.10'}) self.assertEqual(payload['related'], {'hosts': ['host-01'], 'ip': ['192.0.2.10']}) self.assertEqual(payload['event']['kind'], 'alert') @@ -135,6 +131,7 @@ class TestSecurityOnionESAlerter(unittest.TestCase): payload, alert_id = self.send(rule, match) alerter = es.SecurityOnionESAlerter(rule) + self.assertEqual(payload['rule'], {'name': PLAIN_RULE['detection_title'], 'uuid': PLAIN_RULE['detection_public_id']}) self.assertNotIn('labels', payload) self.assertNotIn('related', payload) self.assertNotIn('reason', payload['event']) @@ -142,6 +139,7 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertEqual(payload['event_data'], dict(match, **{'@timestamp': '2026-09-30T16:50:00+00:00'})) self.assertEqual(alert_id, alerter.alert_id(match)) self.assertNotEqual(alerter.alert_id(match), alerter.alert_id(dict(match, _id='abd'))) + self.assertNotEqual(alerter.alert_id(match), es.SecurityOnionESAlerter(dict(rule, detection_public_id='other')).alert_id(match)) # without an _id, never deduplicated self.assertNotEqual(alerter.alert_id({'_id': None}), alerter.alert_id({'_id': None})) @@ -195,21 +193,31 @@ class TestSecurityOnionESAlerter(unittest.TestCase): changed = ('event_data', 'event', 'error', 'tags') self.assertEqual({k: v for k, v in second.items() if k not in changed}, {k: v for k, v in first.items() if k not in changed}) + def test_rejected_dates_are_kept_as_text(self): + match = {'@timestamp': datetime(2026, 9, 30, 16, 50, tzinfo=timezone.utc), '_id': 'abc'} + + _, (second, _) = self.creates(PLAIN_RULE, match, [RequestError(400, 'document_parsing_exception', {}), None]) + + self.assertEqual(json.loads(second['event']['original']), {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc'}) + def test_rejected_twice_is_dropped_without_retry(self): rejected = RequestError(400, 'document_parsing_exception', {}) match = {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc'} - # no EAException, so no retry - self.assertEqual(len(self.creates(PLAIN_RULE, match, [rejected, rejected])), 2) + with self.assertLogs('elastalert', 'ERROR') as logs: + self.assertEqual(len(self.creates(PLAIN_RULE, match, [rejected, rejected])), 2) + + self.assertIn('Dropping alert', logs.output[0]) def test_write_failure_is_retried(self): match = {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc'} - # EAException makes ElastAlert retry the alert with self.assertRaisesRegex(es.EAException, 'Unable to write the alert to Elasticsearch'): self.creates(PLAIN_RULE, match, [ConnectionError('N/A', 'refused', None)]) - # a repeat id is already stored + def test_repeat_id_is_ignored(self): + match = {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc'} + self.assertEqual(len(self.creates(PLAIN_RULE, match, [ConflictError(409, 'version_conflict_engine_exception', {})])), 1) def test_correlation_fields_are_optional(self): @@ -217,9 +225,10 @@ class TestSecurityOnionESAlerter(unittest.TestCase): # no window_start: the summary cannot be built match = {k: v for k, v in correlation_match().items() if k != 'window_start'} - with self.assertLogs('elastalert', 'WARNING'): + with self.assertLogs('elastalert', 'WARNING') as logs: payload, _ = self.send(rule, match) + self.assertIn('without its correlation summary', logs.output[0]) self.assertEqual(payload['event']['kind'], 'alert') self.assertNotIn('reason', payload['event']) self.assertNotIn('labels', payload) @@ -229,21 +238,22 @@ class TestSecurityOnionESAlerter(unittest.TestCase): # no @timestamp: the window end is unknown match = {k: v for k, v in correlation_match().items() if k not in ('@timestamp', 'window_start')} - with self.assertLogs('elastalert', 'WARNING'): + with self.assertLogs('elastalert', 'WARNING') as logs: payload, alert_id = self.send(BASE_RULE, match) + self.assertIn('without a stable id', logs.output[0]) self.assertEqual(len(alert_id), 32) self.assertEqual(payload['event_data'], match) def test_summary_formats_values(self): rule = dict(BASE_RULE, sigma_correlation='value_avg', query_key='source.ip', - summary_template='%count% for %source.ip% to %destination.port%') + summary_template='%count% for %source.ip% to %destination.port% %no.such.field%') match = {'value_avg': 2.5, 'window_start': '2026-09-30T16:49:52+00:00', '@timestamp': '2026-09-30T16:50:00+00:00', 'source': {'ip': '192.0.2.10'}, 'destination': {'port': [22, 80, 443, 8080, 8443]}} payload, _ = self.send(rule, match) - self.assertEqual(payload['event']['reason'], '2.50 for 192.0.2.10 to 22, 80, 443 and 2 more') + self.assertEqual(payload['event']['reason'], '2.50 for 192.0.2.10 to 22, 80, 443 and 2 more %no.such.field%') self.assertEqual(es.SecurityOnionESAlerter.format_count('n/a'), 'n/a') def test_get_info(self): diff --git a/salt/elasticsearch/templates/component/so/detection-mappings.json b/salt/elasticsearch/templates/component/so/detection-mappings.json index 1098e440a..326a8a712 100644 --- a/salt/elasticsearch/templates/component/so/detection-mappings.json +++ b/salt/elasticsearch/templates/component/so/detection-mappings.json @@ -50,18 +50,6 @@ "ignore_above": 1024, "type": "keyword" }, - "ruleType": { - "ignore_above": 1024, - "type": "keyword" - }, - "correlationType": { - "ignore_above": 1024, - "type": "keyword" - }, - "correlationTimespan": { - "ignore_above": 1024, - "type": "keyword" - }, "content": { "type": "text" }, @@ -166,6 +154,20 @@ } } } + }, + "runtime": { + "so_detection.sigmaKind": { + "type": "keyword", + "script": { + "source": "if (doc['so_detection.language'].size() == 0 || doc['so_detection.language'].value != 'sigma') { return; } def d = params._source.so_detection; if (d == null || d.content == null) { return; } String c = d.content; String NL = String.valueOf((char) 10); emit(c.startsWith('correlation:') || c.contains(NL + 'correlation:') ? 'correlation' : 'single');" + } + }, + "so_detection.sigmaCorrelationType": { + "type": "keyword", + "script": { + "source": "if (doc['so_detection.language'].size() == 0 || doc['so_detection.language'].value != 'sigma') { return; } def d = params._source.so_detection; if (d == null || d.content == null) { return; } String c = d.content; String NL = String.valueOf((char) 10); int i = c.startsWith('correlation:') ? 0 : c.indexOf(NL + 'correlation:'); if (i < 0) { return; } i = c.indexOf(NL, i + 1); while (i >= 0 && i + 1 < c.length()) { int e = c.indexOf(NL, i + 1); String line = e < 0 ? c.substring(i + 1) : c.substring(i + 1, e); if (line.length() > 0 && !Character.isWhitespace(line.charAt(0)) && line.charAt(0) != (char) 35) { return; } String t = line.trim(); if (t.startsWith('type:')) { String v = t.substring(5).trim(); int h = v.indexOf(' ' + String.valueOf((char) 35)); if (h >= 0) { v = v.substring(0, h).trim(); } if (v.length() > 1 && (v.charAt(0) == (char) 39 || v.charAt(0) == (char) 34)) { v = v.substring(1, v.length() - 1); } emit(v); return; } i = e; }" + } + } } } }, diff --git a/salt/manager/tools/sbin/soup b/salt/manager/tools/sbin/soup index 5502a378a..c49fd61e0 100755 --- a/salt/manager/tools/sbin/soup +++ b/salt/manager/tools/sbin/soup @@ -1183,11 +1183,6 @@ up_to_3.4.0() { echo "Removing so-kratos, so-hydra and so-soc so they are recreated on the soauth network." docker rm -f so-kratos so-hydra so-soc >> $SOUP_LOG 2>&1 - # backfill the Sigma rule type on existing detections - mkdir -p /opt/so/conf/soc/migrations - echo "0" > /opt/so/conf/soc/migrations/elastalert-migration-3.4.0 - chown -R socore:socore /opt/so/conf/soc/migrations - for template in so-metrics-logstash.node so-metrics-logstash.stack_monitoring.node; do if ! remove_elasticsearch_index_template "$template" "logstash node and node_cel index patterns reversed"; then FINAL_MESSAGE_QUEUE+=("WARNING: Unable to automatically remove the $template index template. Addon integration templates may fail to load until it is removed:") @@ -1264,8 +1259,21 @@ post_to_3.4.0() { rollover_index "$idx" done + add_detection_runtime_fields + set_postversion 3.4.0 } + +add_detection_runtime_fields() { + local template=$DEFAULT_SALT_DIR/salt/elasticsearch/templates/component/so/detection-mappings.json + local runtime + + runtime=$(jq -c '{runtime: .template.mappings.runtime}' "$template") + if ! so-elasticsearch-query so-detection/_mapping -XPUT -d "$runtime" --retry 3 --retry-delay 5 --fail >> $SOUP_LOG 2>&1; then + FINAL_MESSAGE_QUEUE+=("WARNING: Unable to add the Sigma search fields to the so-detection index. Searches on so_detection.sigmaKind will match nothing until they are added:") + FINAL_MESSAGE_QUEUE+=(" - sudo so-elasticsearch-query so-detection/_mapping -XPUT -d \"\$(jq -c '{runtime: .template.mappings.runtime}' $template)\"") + fi +} ### 3.4.0 End ### diff --git a/salt/soc/defaults.yaml b/salt/soc/defaults.yaml index f6a97e878..259d54d04 100644 --- a/salt/soc/defaults.yaml +++ b/salt/soc/defaults.yaml @@ -2681,10 +2681,10 @@ soc: query: "so_detection.language:suricata | groupby so_detection.ruleset so_detection.isEnabled | groupby so_detection.category" description: Show all NIDS Detections, which are run with Suricata - name: "Detection Type - Sigma (Elastalert) - All" - query: "so_detection.language:sigma | groupby so_detection.ruleType | groupby so_detection.ruleset so_detection.isEnabled | groupby so_detection.category | groupby so_detection.product" + query: "so_detection.language:sigma | groupby so_detection.sigmaKind | groupby so_detection.ruleset so_detection.isEnabled | groupby so_detection.category | groupby so_detection.product" description: Show all Sigma Detections, which are run with Elastalert - name: "Detection Type - Sigma (Elastalert) - Correlations" - query: "so_detection.ruleType:correlation | groupby so_detection.correlationType so_detection.isEnabled | groupby so_detection.correlationTimespan | groupby so_detection.ruleset" + query: "so_detection.sigmaKind:correlation | groupby so_detection.sigmaCorrelationType so_detection.isEnabled | groupby so_detection.ruleset" description: Show Sigma correlation Detections - name: "Detection Type - YARA (Strelka)" query: "so_detection.language:yara | groupby so_detection.ruleset so_detection.isEnabled"