From 3c62ef5e16ac46b6d81dee7a0c41a50c6b257bce Mon Sep 17 00:00:00 2001 From: defensivedepth Date: Fri, 2 Oct 2026 13:47:03 -0400 Subject: [PATCH] Cleanup --- .../files/modules/so/securityonion-es.py | 19 +++-- .../files/modules/so/securityonion-es_test.py | 84 +++++++++---------- 2 files changed, 51 insertions(+), 52 deletions(-) diff --git a/salt/elastalert/files/modules/so/securityonion-es.py b/salt/elastalert/files/modules/so/securityonion-es.py index 67ac699f8..2ee0c0145 100644 --- a/salt/elastalert/files/modules/so/securityonion-es.py +++ b/salt/elastalert/files/modules/so/securityonion-es.py @@ -54,12 +54,17 @@ class SecurityOnionESAlerter(Alerter): """ compound_query_key holds the list; query_key is flattened to a string. """ return self.rule.get('compound_query_key') or ([self.rule['query_key']] if self.rule.get('query_key') else []) + @staticmethod + def is_correlation(match): + """ Only correlation rows carry window_start, from the ES|QL stats. """ + return 'window_start' in match + def alert_id(self, match): - """ Stable id: window + group values for correlations, source _id otherwise. """ - keys = self.query_keys() - if keys: - values = '|'.join(str(self.lookup(match, k)) for k in keys) - key = f"{self.rule['detection_public_id']}|{self.to_dt(match['@timestamp']).isoformat()}|{values}" + """ Stable id: window end + group values for correlations, source _id otherwise. """ + if self.is_correlation(match): + # ungrouped rows have a hashed _id that changes with the count + values = ''.join(f"|{self.lookup(match, k)}" for k in self.query_keys()) + key = f"{self.rule['detection_public_id']}|{self.to_dt(match['@timestamp']).isoformat()}{values}" else: key = f"{self.rule['detection_public_id']}|{match.get('_id')}" @@ -135,7 +140,7 @@ class SecurityOnionESAlerter(Alerter): def summary(self, match): """ One-line correlation summary; None for single-event rules. """ - if 'window_start' not in match: + if not self.is_correlation(match): return None start = self.to_dt(match['window_start']) @@ -200,7 +205,7 @@ class SecurityOnionESAlerter(Alerter): } keys = self.query_keys() - if keys: + if keys and self.is_correlation(match): payload["labels"] = { "correlation_group_by": ', '.join(keys), "correlation_group": self.group(match), diff --git a/salt/elastalert/files/modules/so/securityonion-es_test.py b/salt/elastalert/files/modules/so/securityonion-es_test.py index 4915fec90..5cc2c5043 100644 --- a/salt/elastalert/files/modules/so/securityonion-es_test.py +++ b/salt/elastalert/files/modules/so/securityonion-es_test.py @@ -4,6 +4,7 @@ # Elastic License 2.0. import copy +from datetime import datetime, timezone import importlib.util import json import logging @@ -16,9 +17,7 @@ from unittest.mock import MagicMock, patch # Real ElastAlert when installed (so-elastalert); otherwise stand-ins for what the alerter imports. try: import elastalert.alerts # noqa: F401 - HAVE_ELASTALERT = True except ImportError: - HAVE_ELASTALERT = False class Alerter: def __init__(self, rule): @@ -91,7 +90,7 @@ class TestSecurityOnionESAlerter(unittest.TestCase): match['host.name,source.ip'] = 'sa-delta-02-jb, 192.168.198.149' original = copy.deepcopy(match) - payload, url = self.send(rule, match) + payload, _ = self.send(rule, match) self.assertNotIn('host.name,source.ip', payload['event_data']) self.assertEqual(payload['event_data']['host'], {'name': 'sa-delta-02-jb'}) @@ -100,12 +99,8 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertEqual(payload['related'], {'hosts': ['sa-delta-02-jb'], 'ip': ['192.168.198.149']}) self.assertEqual(payload['event']['kind'], 'alert') self.assertEqual(payload['event']['reason'], '3,561 failed network logons to sa-delta-02-jb from 192.168.198.149 in 2 minutes') - self.assertNotIn('summary', payload['rule']) # ElastAlert reuses the match self.assertEqual(match, original) - # id from the group fields, not the compound key - without_key = {k: v for k, v in match.items() if k != 'host.name,source.ip'} - self.assertTrue(url.endswith('/' + es.SecurityOnionESAlerter(rule).alert_id(without_key))) def test_single_query_key(self): rule = dict(BASE_RULE, query_key='user.name', summary_template='%count% failed SOC logins for %user.name%') @@ -132,26 +127,49 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertEqual(payload['related'], {'user': ['admmig', 'svc']}) self.assertEqual(payload['labels']['correlation_group'], "['admmig', 'svc', 'admmig'], not-an-ip, example.com") - def test_group_without_related_fields(self): - rule = dict(BASE_RULE, query_key='dns.highest_registered_domain') - match = dict(correlation_match(), dns={'highest_registered_domain': 'example.com'}) + payload, _ = self.send(dict(BASE_RULE, query_key='dns.highest_registered_domain'), match) - payload, _ = self.send(rule, match) - - self.assertEqual(payload['labels'], {'correlation_group_by': 'dns.highest_registered_domain', 'correlation_group': 'example.com'}) self.assertNotIn('related', payload) def test_plain_rule_has_no_correlation_fields(self): - match = {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc', 'process': {'name': 'whoami.exe'}} + # a query_key alone, as from an override, does not make a correlation + rule = dict(BASE_RULE, query_key='user.name') + # ElastAlert parses @timestamp for EQL hits + match = {'@timestamp': datetime(2026, 9, 30, 16, 50, tzinfo=timezone.utc), '_id': 'abc', + 'process': {'name': 'whoami.exe'}, 'user': {'name': 'josh'}} - payload, url = self.send(dict(BASE_RULE), match) + payload, url = self.send(rule, match) + alerter = es.SecurityOnionESAlerter(rule) self.assertNotIn('labels', payload) self.assertNotIn('related', payload) self.assertNotIn('reason', payload['event']) self.assertEqual(payload['event']['kind'], 'alert') - self.assertEqual(payload['event_data'], match) - self.assertTrue(url.endswith('/' + es.SecurityOnionESAlerter(dict(BASE_RULE)).alert_id(match))) + self.assertEqual(payload['event_data'], dict(match, **{'@timestamp': '2026-09-30T16:50:00+00:00'})) + self.assertTrue(url.endswith('/' + alerter.alert_id(match))) + self.assertNotEqual(alerter.alert_id(match), alerter.alert_id(dict(match, _id='abd'))) + + def test_ungrouped_correlation_id_ignores_row_hash(self): + rule = dict(BASE_RULE, summary_template=None) + first = {k: v for k, v in correlation_match().items() if k not in ('host', 'source')} + # ES|QL hashes the row into _id, so a later count changes it + later = dict(first, event_count=3600, _id='9f2a') + + payload, url = self.send(rule, first) + + alerter = es.SecurityOnionESAlerter(rule) + self.assertEqual(alerter.alert_id(first), alerter.alert_id(later)) + self.assertNotEqual(alerter.alert_id(first), alerter.alert_id(dict(first, **{'@timestamp': '2026-09-30T18:09:00+00:00'}))) + self.assertTrue(url.endswith('/' + alerter.alert_id(first))) + self.assertEqual(payload['event']['reason'], '3,561 events in 2 minutes') + self.assertNotIn('labels', payload) + + def test_grouped_correlation_id_unchanged(self): + """ Ids of alerts already written must not change. """ + rule = dict(BASE_RULE, compound_query_key=['host.name', 'source.ip'], query_key='host.name,source.ip') + key = f"{BASE_RULE['detection_public_id']}|2026-09-30T18:07:53+00:00|sa-delta-02-jb|192.168.198.149" + + self.assertEqual(es.SecurityOnionESAlerter(rule).alert_id(correlation_match()), es.hashlib.sha256(key.encode()).hexdigest()) def send_responses(self, rule, match, responses): """ Run alert() against a sequence of write responses; return the payloads written. """ @@ -174,10 +192,10 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertEqual(second['tags'], ['alert', 'preserve_original_event']) self.assertEqual(first['tags'], ['alert']) self.assertTrue(second['error']['message'].startswith('event_data rejected by Elasticsearch: {"error"')) - self.assertEqual(second['event']['reason'], first['event']['reason']) - self.assertEqual(second['labels'], first['labels']) - self.assertEqual(second['related'], first['related']) - self.assertEqual(second['rule'], first['rule']) + # everything else carries over + self.assertEqual({k: v for k, v in second['event'].items() if k != 'original'}, first['event']) + 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_twice_is_dropped_without_retry(self): rejected = MagicMock(status_code=400, ok=False, text='{"error":{"type":"document_parsing_exception"}}') @@ -188,30 +206,6 @@ class TestSecurityOnionESAlerter(unittest.TestCase): self.assertEqual(len(payloads), 2) - @unittest.skipUnless(HAVE_ELASTALERT, 'needs ElastAlert, as in the so-elastalert container') - def test_group_matches_elastalert_silence_key(self): - from elastalert.elastalert import ElastAlerter - from elastalert.util import ts_to_dt - - cases = [ - (['host.name', 'source.ip'], {'host': {'name': 'sa-delta-02-jb'}, 'source': {'ip': '192.168.198.149'}}), - (['winlog.event_data.TargetUserName', 'source.ip'], {'winlog': {'event_data': {'TargetUserName': ['admmig', 'svc']}}, 'source': {'ip': '10.23.23.9'}}), - (['user.name'], {'user': {'name': "o'brien \"q\" \\ *:?@local.invalid"}}), - ] - for keys, fields in cases: - with self.subTest(keys=keys): - rule = dict(BASE_RULE, timestamp_field='@timestamp', ts_to_dt=ts_to_dt) - if len(keys) > 1: - rule.update(compound_query_key=keys, query_key=','.join(keys)) - else: - rule['query_key'] = keys[0] - hit = {'_id': 'x', '_source': dict(correlation_match(), **copy.deepcopy(fields))} - match = ElastAlerter.process_hits(rule, [hit])[0] - - silence_suffix = ElastAlerter.get_named_key_value(None, rule, match, 'query_key') - - self.assertEqual(es.SecurityOnionESAlerter(rule).group(match), silence_suffix) - if __name__ == '__main__': unittest.main()