ES|QL and Sigma correlation support

update tick interval desc

.github/workflows/pythontest.yml

@@ -6,6 +6,7 @@ on:
       - "salt/sensoroni/files/analyzers/**"
       - "salt/manager/tools/sbin/**"
       - "salt/_beacons/**"
+      - "salt/elastalert/files/modules/**"
       - "salt/telegraf/tools/sbin_jinja/**"
       - "salt/telegraf/defaults.yaml"
       - "salt/telegraf/soc_telegraf.yaml"
@@ -18,7 +19,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.14"]
-        python-code-path: ["salt/sensoroni/files/analyzers", "salt/manager/tools/sbin", "salt/_beacons"]
+        python-code-path: ["salt/sensoroni/files/analyzers", "salt/manager/tools/sbin", "salt/_beacons", "salt/elastalert/files/modules/so"]
 
     steps:
     - uses: actions/checkout@v3

salt/_beacons/postgres_pillar_beacon.py

@@ -131,8 +131,6 @@ def beacon(config):  # noqa: C901
             'setting_id': setting_id,
             'node_id': node_id,
         })
-        log.info('postgres_pillar_beacon: audit_settings id=%d setting_id=%s node_id=%s',
-                 row_id, setting_id, node_id)
         if row_id > max_id:
             max_id = row_id
 

salt/elastalert/defaults.yaml

@@ -10,7 +10,7 @@ elastalert:
     buffer_time:
       minutes: 10
     old_query_limit:
-      minutes: 5
+      minutes: 1440
     es_port: 9200
     es_conn_timeout: 55
     max_query_size: 5000

salt/elastalert/files/modules/so/conftest.py

@@ -0,0 +1,80 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+from datetime import datetime
+import json
+import logging
+import sys
+import types
+
+# stand-ins when ElastAlert isn't installed (CI)
+try:
+    import elastalert.alerts  # noqa: F401
+except ImportError:
+
+    class Alerter:
+        def __init__(self, rule):
+            self.rule = rule
+
+    class DateTimeEncoder(json.JSONEncoder):
+        def default(self, obj):
+            return obj.isoformat() if hasattr(obj, 'isoformat') else json.JSONEncoder.default(self, obj)
+
+    class EAException(Exception):
+        pass
+
+    def lookup_es_key(doc, term):
+        for part in term.split('.'):
+            if not isinstance(doc, dict) or part not in doc:
+                return None
+            doc = doc[part]
+        return doc
+
+    def ts_to_dt(value):
+        return value if isinstance(value, datetime) else datetime.fromisoformat(value)
+
+    def elasticsearch_client(conf):
+        return None  # tests set the alerter's client
+
+    alerts = types.ModuleType('elastalert.alerts')
+    alerts.Alerter = Alerter
+    alerts.DateTimeEncoder = DateTimeEncoder
+    util = types.ModuleType('elastalert.util')
+    util.EAException = EAException
+    util.elastalert_logger = logging.getLogger('elastalert')
+    util.lookup_es_key = lookup_es_key
+    util.ts_to_dt = ts_to_dt
+    util.elasticsearch_client = elasticsearch_client
+    package = types.ModuleType('elastalert')
+    package.alerts = alerts
+    package.util = util
+    sys.modules.update({'elastalert': package, 'elastalert.alerts': alerts, 'elastalert.util': util})
+
+# stand-ins when elasticsearch-py isn't installed (CI)
+try:
+    import elasticsearch.exceptions  # noqa: F401
+except ImportError:
+
+    class ElasticsearchException(Exception):
+        pass
+
+    class TransportError(ElasticsearchException):
+        pass
+
+    class ConnectionError(TransportError):
+        pass
+
+    class ConflictError(TransportError):
+        pass
+
+    class RequestError(TransportError):
+        pass
+
+    exceptions = types.ModuleType('elasticsearch.exceptions')
+    for cls in (ElasticsearchException, TransportError, ConnectionError, ConflictError, RequestError):
+        setattr(exceptions, cls.__name__, cls)
+    es_package = types.ModuleType('elasticsearch')
+    es_package.exceptions = exceptions
+    sys.modules.update({'elasticsearch': es_package, 'elasticsearch.exceptions': exceptions})

salt/elastalert/files/modules/so/securityonion-es.py

@@ -1,63 +1,269 @@
 # -*- coding: utf-8 -*-
 
 # Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
-# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.
 
-
-from time import gmtime, strftime
-import requests,json
-from elastalert.alerts import Alerter
+from datetime import datetime, timezone
+import hashlib
+import ipaddress
+import json
+import re
+import uuid
 
 import urllib3
+from elasticsearch.exceptions import ConflictError, ElasticsearchException, RequestError
+from elastalert.alerts import Alerter, DateTimeEncoder
+from elastalert.util import EAException, elastalert_logger, elasticsearch_client, lookup_es_key, ts_to_dt
+
+# grid runs verify_certs: false; also quiets ElastAlert's own queries
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
+ALERT_INDEX = 'logs-detections.alerts-so'
+# ES error text kept in logs and alerts
+ERROR_TEXT_LIMIT = 500
+# a match missing backend columns (window_start, count, @timestamp); the alert is still written
+MATCH_ERRORS = (KeyError, TypeError, ValueError)
+
+
 class SecurityOnionESAlerter(Alerter):
     """
     Use matched data to create alerts in Elasticsearch.
     """
 
-    required_options = set(['detection_title', 'sigma_level'])
-    optional_fields = ['sigma_category', 'sigma_product', 'sigma_service']
+    required_options = {'detection_title', 'sigma_level'}
+    optional_fields = ['sigma_category', 'sigma_product', 'sigma_service', 'sigma_correlation']
+
+    # count column and default summary per type; stored alert data, so not localized
+    CORRELATION_COUNTS = {
+        'event_count': ('event_count', '%count% events'),
+        'value_count': ('value_count', '%count% distinct values'),
+        'temporal': ('event_type_count', '%count% correlated rules matched'),
+        'value_sum': ('value_sum', 'total %count%'),
+        'value_avg': ('value_avg', 'average %count%'),
+        'value_percentile': ('value_percentile', 'percentile %count%'),
+        'value_median': ('value_median', 'median %count%'),
+    }
+    PLACEHOLDER = re.compile(r'%([^%\s]+)%')
+    # group-by fields copied into ECS related.*
+    RELATED_USERS = {'user.name', 'winlog.event_data.TargetUserName', 'winlog.event_data.SubjectUserName'}
+    RELATED_HOSTS = {'host.name', 'host.hostname', 'winlog.computer_name'}
+
+    def __init__(self, rule):
+        super().__init__(rule)
+        # uses the grid's TLS, auth and timeout settings
+        self.es = elasticsearch_client(rule)
+
+    @property
+    def is_correlation(self):
+        return bool(self.rule.get('sigma_correlation'))
+
+    def query_keys(self):
+        """compound_query_key holds the list; query_key is flattened to a string."""
+        if self.rule.get('compound_query_key'):
+            return self.rule['compound_query_key']
+        if self.rule.get('query_key'):
+            return [self.rule['query_key']]
+        return []
+
+    def alert_id(self, match):
+        """Stable id: window end + group values for correlations, source _id otherwise; random without one."""
+        if self.is_correlation:
+            # ungrouped rows have a hashed _id that changes with the count
+            values = ''.join(f"|{lookup_es_key(match, k)}" for k in self.query_keys())
+            key = f"{self.rule['detection_public_id']}|{ts_to_dt(match['@timestamp']).isoformat()}{values}"
+        elif match.get('_id'):
+            key = f"{self.rule['detection_public_id']}|{match['_id']}"
+        else:
+            return uuid.uuid4().hex
+
+        return hashlib.sha256(key.encode('utf-8')).hexdigest()
+
+    def group(self, match):
+        """Group-by values, joined like ElastAlert's realert key."""
+        return ', '.join(str(lookup_es_key(match, k)) for k in self.query_keys())
+
+    def related_bucket(self, key):
+        if key == 'ip' or key.endswith('.ip'):
+            return 'ip'
+        if key in self.RELATED_USERS or key.endswith('.user.name'):
+            return 'user'
+        if key in self.RELATED_HOSTS:
+            return 'hosts'
+        return None
+
+    @staticmethod
+    def valid_ip(value):
+        try:
+            ipaddress.ip_address(value)
+            return True
+        except ValueError:
+            return False
+
+    def related(self, match):
+        """ECS related.* from group-by values; skips invalid IPs."""
+        related = {}
+        for key in self.query_keys():
+            bucket = self.related_bucket(key)
+            if not bucket:
+                continue
+            # original spellings of a lowercased group
+            value = lookup_es_key(match, f"{key}_spellings")
+            if value is None:
+                value = lookup_es_key(match, key)
+            for v in value if isinstance(value, list) else [value]:
+                if v is None or (bucket == 'ip' and not self.valid_ip(str(v))):
+                    continue
+                # dict: ordered and deduped
+                related.setdefault(bucket, {})[str(v)] = None
+        return {bucket: list(values) for bucket, values in related.items()}
+
+    def event_data(self, match):
+        """The match minus the compound query_key field, which ES would map by its last part."""
+        if not self.rule.get('compound_query_key'):
+            return match
+        return {k: v for k, v in match.items() if k != self.rule['query_key']}
+
+    @staticmethod
+    def format_value(value):
+        if isinstance(value, list):
+            shown = ', '.join(str(v) for v in value[:3])
+            return shown if len(value) <= 3 else f"{shown} and {len(value) - 3} more"
+        return str(value)
+
+    @staticmethod
+    def format_count(value):
+        if isinstance(value, float) and not value.is_integer():
+            return f"{value:,.2f}"
+        if isinstance(value, (int, float)):
+            return f"{int(value):,}"
+        return str(value)
+
+    @staticmethod
+    def format_duration(seconds):
+        for unit, size in (('hour', 3600), ('minute', 60)):
+            if seconds >= 2 * size:
+                return f"{seconds // size} {unit}s"
+        return f"{seconds} second{'' if seconds == 1 else 's'}"
+
+    def summary(self, match):
+        """One-line correlation summary."""
+        column, label = self.CORRELATION_COUNTS.get(self.rule['sigma_correlation'], (None, '%count%'))
+        start = ts_to_dt(match['window_start'])
+        end = ts_to_dt(match['@timestamp'])
+        values = {
+            'count': self.format_count(match.get(column)),
+            'start': start.strftime('%Y-%m-%d %H:%M:%S UTC'),
+            'end': end.strftime('%Y-%m-%d %H:%M:%S UTC'),
+            'duration': self.format_duration(int((end - start).total_seconds())),
+        }
+
+        template = self.rule.get('summary_template')
+        if not template:
+            groups = ', '.join(f"{k} %{k}%" for k in self.query_keys())
+            template = f"{label} for {groups} in %duration%" if groups else f"{label} in %duration%"
+
+        def fill(m):
+            if m[1] in values:
+                return values[m[1]]
+            value = lookup_es_key(match, m[1])
+            # unknown placeholders stay visible so typos show
+            return m[0] if value is None else self.format_value(value)
+
+        return self.PLACEHOLDER.sub(fill, template)
 
     def alert(self, matches):
         for match in matches:
-            timestamp = strftime("%Y-%m-%d"'T'"%H:%M:%S"'.000Z', gmtime())
-            headers = {"Content-Type": "application/json"}
+            try:
+                alert_id = self.alert_id(match)
+            except MATCH_ERRORS as e:
+                elastalert_logger.warning("Writing alert for rule %s without a stable id, so a retry may duplicate it: %r",
+                                          self.rule['detection_public_id'], e)
+                alert_id = uuid.uuid4().hex
+            try:
+                self.write(alert_id, self.payload(match))
+            except ElasticsearchException as e:
+                # EAException makes ElastAlert retry
+                raise EAException(f"Unable to write the alert to Elasticsearch: {str(e)[:ERROR_TEXT_LIMIT]}") from e
 
-            creds = None
-            if 'es_username' in self.rule and 'es_password' in self.rule:
-                creds = (self.rule['es_username'], self.rule['es_password'])
+    def payload(self, match):
+        rule_info = {
+            "name": self.rule['detection_title'],
+            "uuid": self.rule['detection_public_id']
+        }
 
-            # Start building the rule dict
-            rule_info = {
-                "name": self.rule['detection_title'],
-                "uuid": self.rule['detection_public_id']
-            }
+        # Add optional fields if they are present in the rule
+        for field in self.optional_fields:
+            rule_key = field.split('_')[-1]  # Assumes field format "sigma_<key>"
+            if field in self.rule:
+                rule_info[rule_key] = self.rule[field]
 
-            # Add optional fields if they are present in the rule
-            for field in self.optional_fields:
-                rule_key = field.split('_')[-1]  # Assumes field format "sigma_<key>"
-                if field in self.rule:
-                    rule_info[rule_key] = self.rule[field]
+        event_info = {
+            "kind": "alert",
+            "severity": self.rule['event.severity'],
+            "module": self.rule['event.module'],
+            "dataset": self.rule['event.dataset'],
+            "severity_label": self.rule['sigma_level']
+        }
 
-            # Construct the payload with the conditional rule_info
-            payload = {
-                "tags": "alert",
-                "rule": rule_info,
-                "event": {
-                    "severity": self.rule['event.severity'],
-                    "module": self.rule['event.module'],
-                    "dataset": self.rule['event.dataset'],
-                    "severity_label": self.rule['sigma_level']
-                },
-                "sigma_level": self.rule['sigma_level'],
-                "event_data": match,
-                "@timestamp": timestamp
-            }
-            url = f"https://{self.rule['es_host']}:{self.rule['es_port']}/logs-detections.alerts-so/_doc/"
-            requests.post(url, data=json.dumps(payload), headers=headers, verify=False, auth=creds)
+        payload = {
+            "tags": ["alert"],
+            "rule": rule_info,
+            "event": event_info,
+            "sigma_level": self.rule['sigma_level'],
+            "event_data": self.event_data(match),
+            "@timestamp": datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.000Z')
+        }
+
+        if self.is_correlation:
+            keys = self.query_keys()
+            try:
+                # built before any is added, so a failure adds none
+                reason = self.summary(match)
+                labels = {"correlation_group_by": ', '.join(keys), "correlation_group": self.group(match)} if keys else None
+                related = self.related(match)
+            except MATCH_ERRORS as e:
+                elastalert_logger.warning("Writing alert for rule %s without its correlation summary: %r",
+                                          self.rule['detection_public_id'], e)
+            else:
+                payload["event"]["reason"] = reason
+                if labels:
+                    payload["labels"] = labels
+                if related:
+                    payload["related"] = related
+
+        return payload
+
+    def write(self, alert_id, payload):
+        try:
+            self.create(alert_id, payload)
+        except RequestError as e:
+            # mapping rejections come from event_data; retry it as text
+            rejection = str(e)[:ERROR_TEXT_LIMIT]
+            try:
+                self.create(alert_id, self.without_event_data(payload, rejection))
+            except RequestError as again:
+                elastalert_logger.error("Dropping alert %s for rule %s, rejected by Elasticsearch even without its event data: %s; first rejection: %s",
+                                        alert_id, self.rule['detection_public_id'], str(again)[:ERROR_TEXT_LIMIT], rejection)
+                return
+            elastalert_logger.warning("Stored alert %s for rule %s with its event data as text, rejected by Elasticsearch: %s",
+                                      alert_id, self.rule['detection_public_id'], rejection)
+
+    def create(self, alert_id, payload):
+        try:
+            self.es.create(index=ALERT_INDEX, id=alert_id, body=payload)
+        except ConflictError:
+            pass  # a repeat id is already stored
+
+    @staticmethod
+    def without_event_data(payload, rejection):
+        """Moves event_data to event.original; the tag keeps Fleet's final pipeline from dropping it."""
+        fallback = {k: v for k, v in payload.items() if k != 'event_data'}
+        fallback['event'] = dict(payload['event'], original=json.dumps(payload['event_data'], cls=DateTimeEncoder))
+        fallback['error'] = {'message': f"event_data rejected by Elasticsearch: {rejection}"}
+        fallback['tags'] = payload['tags'] + ['preserve_original_event']
+        return fallback
 
     def get_info(self):
         return {'type': 'SecurityOnionESAlerter'}

salt/elastalert/files/modules/so/securityonion-es_test.py

@@ -0,0 +1,250 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+import copy
+from datetime import datetime, timezone
+import importlib.util
+import json
+import os
+import unittest
+from unittest.mock import MagicMock
+
+from elasticsearch.exceptions import ConflictError, ConnectionError, RequestError
+
+spec = importlib.util.spec_from_file_location('securityonion_es', os.path.join(os.path.dirname(__file__), 'securityonion-es.py'))
+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',
+    'sigma_correlation': 'event_count',
+    '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%',
+}
+
+PLAIN_RULE = {k: v for k, v in BASE_RULE.items() if k not in ('sigma_correlation', 'summary_template')}
+
+
+def correlation_match():
+    return {
+        'event_count': 3561,
+        'window_start': '2026-09-30T18:05:10+00:00',
+        '@timestamp': '2026-09-30T18:07:53+00:00',
+        'host': {'name': 'host-01'},
+        'source': {'ip': '192.0.2.10'},
+        '_id': '6d1c',
+        'num_hits': 1,
+        'num_matches': 1,
+    }
+
+
+class TestSecurityOnionESAlerter(unittest.TestCase):
+
+    def creates(self, rule, match, effects=None):
+        """Run alert(); return (body, id) of each create."""
+        alerter = es.SecurityOnionESAlerter(rule)
+        alerter.es = MagicMock()
+        alerter.es.create.side_effect = effects
+        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):
+        """Run alert(); return the payload it wrote and its id."""
+        (payload, alert_id), = self.creates(rule, match)
+        return payload, alert_id
+
+    def test_compound_query_key_left_out_of_event_data(self):
+        rule = dict(BASE_RULE, compound_query_key=['host.name', 'source.ip'], query_key='host.name,source.ip')
+        match = correlation_match()
+        match['host.name,source.ip'] = 'host-01, 192.0.2.10'
+        original = copy.deepcopy(match)
+
+        payload, _ = self.send(rule, match)
+
+        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['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')
+        self.assertEqual(payload['event']['reason'], '3,561 failed network logons to host-01 from 192.0.2.10 in 2 minutes')
+        # ElastAlert reuses the match
+        self.assertEqual(match, original)
+
+    def test_single_query_key(self):
+        rule = dict(BASE_RULE, query_key='user.name', summary_template='%count% failed SOC logins for %user.name%')
+        match = {'event_count': 3, 'window_start': '2026-09-30T16:49:52+00:00', '@timestamp': '2026-09-30T16:50:00+00:00',
+                 'user': {'name': 'user@example.invalid'}}
+
+        payload, _ = self.send(rule, match)
+
+        self.assertEqual(payload['labels'], {'correlation_group_by': 'user.name', 'correlation_group': 'user@example.invalid'})
+        self.assertEqual(payload['related'], {'user': ['user@example.invalid']})
+        self.assertEqual(payload['event']['reason'], '3 failed SOC logins for user@example.invalid')
+        self.assertEqual(payload['event_data'], match)
+
+    def test_related_buckets(self):
+        rule = dict(BASE_RULE, compound_query_key=['winlog.event_data.TargetUserName', 'source.ip', 'dns.highest_registered_domain'],
+                    query_key='winlog.event_data.TargetUserName,source.ip,dns.highest_registered_domain')
+        match = correlation_match()
+        match.update({'winlog': {'event_data': {'TargetUserName': ['admin1', 'svc', 'admin1']}},
+                      'source': {'ip': 'not-an-ip'}, 'dns': {'highest_registered_domain': 'example.com'}})
+
+        payload, _ = self.send(rule, match)
+
+        # deduped; invalid IP skipped; domain stays in the group only
+        self.assertEqual(payload['related'], {'user': ['admin1', 'svc']})
+        self.assertEqual(payload['labels']['correlation_group'], "['admin1', 'svc', 'admin1'], not-an-ip, example.com")
+
+        payload, _ = self.send(dict(BASE_RULE, query_key='dns.highest_registered_domain'), match)
+
+        self.assertNotIn('related', payload)
+
+    def test_related_uses_original_spellings(self):
+        rule = dict(BASE_RULE, query_key='user.name', summary_template=None)
+        match = {'event_count': 3, 'window_start': '2026-09-30T16:49:52+00:00', '@timestamp': '2026-09-30T16:50:00+00:00',
+                 'user': {'name': 'admin', 'name_spellings': ['Admin', 'admin', 'ADMIN']}}
+
+        payload, _ = self.send(rule, match)
+
+        # the group shows the lowercased value; related.user finds every spelling
+        self.assertEqual(payload['labels']['correlation_group'], 'admin')
+        self.assertEqual(payload['related'], {'user': ['Admin', 'admin', 'ADMIN']})
+        self.assertEqual(payload['event']['reason'], '3 events for user.name admin in 8 seconds')
+
+    def test_plain_rule_has_no_correlation_fields(self):
+        # a query_key alone (e.g. from an override) isn't a correlation
+        rule = dict(PLAIN_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': 'user'}}
+
+        payload, alert_id = 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'], 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')))
+        # without an _id, never deduplicated
+        self.assertNotEqual(alerter.alert_id({'_id': None}), alerter.alert_id({'_id': None}))
+
+    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, alert_id = 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.assertEqual(alert_id, alerter.alert_id(first))
+        self.assertEqual(payload['event']['reason'], '3,561 events in 2 minutes')
+        self.assertNotIn('labels', payload)
+
+    def test_temporal_count_column(self):
+        rule = dict(BASE_RULE, sigma_correlation='temporal', summary_template=None)
+        match = {'event_type_count': 2, 'window_start': '2026-09-30T16:49:52+00:00', '@timestamp': '2026-09-30T16:50:00+00:00'}
+
+        payload, _ = self.send(rule, match)
+
+        self.assertEqual(payload['event']['reason'], '2 correlated rules matched in 8 seconds')
+
+    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|host-01|192.0.2.10"
+
+        self.assertEqual(es.SecurityOnionESAlerter(rule).alert_id(correlation_match()), es.hashlib.sha256(key.encode()).hexdigest())
+
+    def test_rejected_event_data_is_kept_as_text(self):
+        rule = dict(BASE_RULE, query_key='user.name', summary_template='%count% failed SOC logins for %user.name%')
+        match = {'event_count': 3, 'window_start': '2026-09-30T16:49:52+00:00', '@timestamp': '2026-09-30T16:50:00+00:00',
+                 'user': {'name': 'user@example.invalid'}}
+        rejected = RequestError(400, 'document_parsing_exception', {})
+
+        (first, _), (second, _) = self.creates(rule, match, [rejected, None])
+
+        self.assertIn('event_data', first)
+        self.assertNotIn('event_data', second)
+        self.assertEqual(json.loads(second['event']['original']), match)
+        self.assertEqual(second['tags'], ['alert', 'preserve_original_event'])
+        self.assertEqual(first['tags'], ['alert'])
+        self.assertTrue(second['error']['message'].startswith('event_data rejected by Elasticsearch: '))
+        self.assertIn('document_parsing_exception', second['error']['message'])
+        # 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 = 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)
+
+    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
+        self.assertEqual(len(self.creates(PLAIN_RULE, match, [ConflictError(409, 'version_conflict_engine_exception', {})])), 1)
+
+    def test_correlation_fields_are_optional(self):
+        rule = dict(BASE_RULE, query_key='source.ip')
+        # 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'):
+            payload, _ = self.send(rule, match)
+
+        self.assertEqual(payload['event']['kind'], 'alert')
+        self.assertNotIn('reason', payload['event'])
+        self.assertNotIn('labels', payload)
+        self.assertNotIn('related', payload)
+
+    def test_unstable_id_still_writes(self):
+        # 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'):
+            payload, alert_id = self.send(BASE_RULE, match)
+
+        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%')
+        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(es.SecurityOnionESAlerter.format_count('n/a'), 'n/a')
+
+    def test_get_info(self):
+        self.assertEqual(es.SecurityOnionESAlerter(PLAIN_RULE).get_info(), {'type': 'SecurityOnionESAlerter'})

salt/elastalert/soc_elastalert.yaml

@@ -120,7 +120,7 @@ elastalert:
         helpLink: elastalert
     old_query_limit:
       minutes:
-        description: Amount of time in minutes between queries to start at the most recently run query.
+        description: How long ElastAlert can be down, in minutes, and still resume each rule where it stopped. After a longer outage, rules restart from now and skip the gap.
         global: True
         helpLink: elastalert
     es_conn_timeout:

salt/elasticsearch/defaults.yaml

@@ -1160,6 +1160,7 @@ elasticsearch:
         - so-fleet_agent_id_verification-1
         - so-logs-mappings
         - so-logs-settings
+        - detections-alerts-mappings
         data_stream:
           allow_custom_routing: false
           hidden: false

salt/elasticsearch/files/ingest/logs-system.auth@custom

@@ -0,0 +1,34 @@
+{
+    "version": 1,
+    "_meta": {
+        "managed_by": "securityonion",
+        "managed": true
+    },
+    "description": "Custom pipeline for the System integration's auth data stream.",
+    "processors": [
+        {
+            "trim": {
+                "description": "Grok leaves a leading space on 'invalid user' names (elastic/integrations#12174) and, before 2.23.2, sudo padding",
+                "field": "user.name",
+                "ignore_missing": true,
+                "ignore_failure": true
+            }
+        },
+        {
+            "trim": {
+                "description": "Appended from the untrimmed user.name",
+                "field": "related.user",
+                "ignore_missing": true,
+                "ignore_failure": true
+            }
+        },
+        {
+            "script": {
+                "description": "Dedupe after trimming",
+                "if": "ctx.related?.user instanceof List",
+                "source": "ctx.related.user = new ArrayList(new LinkedHashSet(ctx.related.user));",
+                "ignore_failure": true
+            }
+        }
+    ]
+}

salt/elasticsearch/templates/component/so/detection-mappings.json

@@ -50,6 +50,18 @@
               "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"
             },

salt/elasticsearch/templates/component/so/detections-alerts-mappings.json

@@ -0,0 +1,149 @@
+{
+  "template": {
+    "mappings": {
+      "properties": {
+        "tags": {
+          "ignore_above": 1024,
+          "type": "keyword"
+        },
+        "sigma_level": {
+          "ignore_above": 1024,
+          "type": "keyword"
+        },
+        "rule": {
+          "properties": {
+            "name": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "uuid": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "category": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "product": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "service": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "correlation": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            }
+          }
+        },
+        "event": {
+          "properties": {
+            "severity": {
+              "type": "long"
+            },
+            "severity_label": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "module": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "dataset": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "kind": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "reason": {
+              "type": "match_only_text",
+              "fields": {
+                "keyword": {
+                  "ignore_above": 1024,
+                  "type": "keyword"
+                }
+              }
+            },
+            "original": {
+              "type": "keyword",
+              "index": false,
+              "doc_values": false
+            }
+          }
+        },
+        "event_data": {
+          "properties": {
+            "@timestamp": {
+              "type": "date"
+            },
+            "window_start": {
+              "type": "date"
+            },
+            "event_count": {
+              "type": "long"
+            },
+            "value_count": {
+              "type": "long"
+            },
+            "event_type_count": {
+              "type": "long"
+            },
+            "value_sum": {
+              "type": "double"
+            },
+            "value_avg": {
+              "type": "double"
+            },
+            "value_percentile": {
+              "type": "double"
+            },
+            "value_median": {
+              "type": "double"
+            }
+          }
+        },
+        "labels": {
+          "properties": {
+            "correlation_group_by": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "correlation_group": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            }
+          }
+        },
+        "related": {
+          "properties": {
+            "ip": {
+              "type": "ip"
+            },
+            "user": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "hosts": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            }
+          }
+        },
+        "error": {
+          "properties": {
+            "message": {
+              "type": "match_only_text"
+            }
+          }
+        }
+      }
+    }
+  },
+  "_meta": {
+    "description": "Fields written by the ElastAlert SecurityOnionESAlerter to logs-detections.alerts-so"
+  }
+}

salt/manager/tools/sbin/so-push-drainer

@@ -19,8 +19,6 @@ is older than debounce_seconds, this script:
   * dispatches a single `salt-run state.orchestrate orch.push_batch --async`
     with the deduped actions list passed as pillar kwargs
   * deletes the contributed intent files on successful dispatch
-  * records the orchestration jid under /opt/so/state/push_dispatched and, on
-    later passes, looks up its result and logs success or per-minion failures
 
 Reactor sls files (push_files, push_pillar) write intents
 but never dispatch directly
@@ -32,7 +30,6 @@ import json
 import logging
 import logging.handlers
 import os
-import re
 import subprocess
 import sys
 import time
@@ -43,22 +40,8 @@ PENDING_DIR = '/opt/so/state/push_pending'
 LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
 LOG_FILE = '/opt/so/log/salt/so-push-drainer.log'
 
-DISPATCHED_DIR = '/opt/so/state/push_dispatched'
-
 HIGHSTATE_SENTINEL = '__highstate__'
 
-RESULT_CHECK_DELAY = 30
-RESULT_RECHECK_MAX = 300
-RESULT_MAX_AGE = 7200
-RESULT_CHECKS_PER_PASS = 5
-TEXT_LIMIT = 500
-
-# Lead-in salt puts on the comment of an orchestration step that raised.
-STEP_RAISED = 'An exception occurred in this state:'
-
-# salt-run --async reports the jid only in a log line (stderr by default).
-JID_RE = re.compile(r'salt/run/(\d{20})')
-
 
 def _make_logger():
     logger = logging.getLogger('so-push-drainer')
@@ -130,189 +113,14 @@ def _dispatch(actions, log):
     except subprocess.CalledProcessError as exc:
         log.error('dispatch failed (rc=%s): stdout=%s stderr=%s',
                   exc.returncode, exc.stdout, exc.stderr)
-        return None
+        return False
     except subprocess.TimeoutExpired:
         log.error('dispatch timed out after 60s')
-        return None
+        return False
     except Exception:
         log.exception('dispatch raised')
-        return None
-    output = '{}\n{}'.format(result.stderr or '', result.stdout or '')
-    match = JID_RE.search(output)
-    if not match:
-        log.warning('dispatch accepted but no jid found, result will not be tracked: output=%s',
-                    _trim(output))
-        return ''
-    log.info('dispatch accepted: jid=%s', match.group(1))
-    return match.group(1)
-
-
-def _trim(value):
-    text = value if isinstance(value, str) else json.dumps(value, default=str)
-    lines = [line.strip() for line in text.splitlines() if line.strip()]
-    if 'Traceback (most recent call last):' in text:
-        # Keep the lead-in and the raised exception; the frames are noise in a log line.
-        lines = [text.split('Traceback (most recent call last):', 1)[0].strip(), lines[-1]]
-    text = ' '.join(line for line in lines if line)
-    return text if len(text) <= TEXT_LIMIT else text[:TEXT_LIMIT] + '...'
-
-
-def _unlink(path, log):
-    try:
-        os.unlink(path)
-    except FileNotFoundError:
-        pass
-    except OSError:
-        log.exception('failed to remove %s', path)
-
-
-def _write_record(path, record, log):
-    try:
-        os.makedirs(DISPATCHED_DIR, exist_ok=True)
-        tmp_path = path + '.tmp'
-        with open(tmp_path, 'w') as f:
-            json.dump(record, f)
-        os.rename(tmp_path, path)
-    except Exception:
-        log.exception('failed to record dispatch %s', record.get('jid'))
-
-
-def _record_dispatch(jid, actions, paths, log):
-    record = {'jid': jid, 'dispatched_at': time.time(), 'actions': actions, 'paths': paths}
-    _write_record(os.path.join(DISPATCHED_DIR, '{}.json'.format(jid)), record, log)
-
-
-def _lookup_jid(jid, log):
-    """Returns the job cache entry for jid, {} while it is still running, or None on error."""
-    cmd = ['salt-run', 'jobs.lookup_jid', jid, '--out=json']
-    try:
-        result = subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
-        return json.loads(result.stdout or '{}')
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, ValueError) as exc:
-        log.warning('lookup of jid %s failed: %s', jid, exc)
-        return None
-
-
-def _minion_failure(minion_ret):
-    if isinstance(minion_ret, dict):
-        return '; '.join(
-            '{}: {}'.format(state.get('__id__', state_key), _trim(state.get('comment', '')))
-            for state_key, state in minion_ret.items()
-            if isinstance(state, dict) and state.get('result') is False
-        )
-    # A state run rejected before it starts (e.g. another state run is in
-    # progress) returns a list of error strings instead of state results.
-    if isinstance(minion_ret, (list, str)):
-        return _trim(minion_ret)
-    return ''
-
-
-def _step_failures(step):
-    if not isinstance(step, dict) or step.get('result') is not False:
-        return []
-    failures = ['{}: {}'.format(step.get('__id__', step.get('name')), _trim(step.get('comment', '')))]
-    changes = step.get('changes')
-    minion_rets = changes.get('ret') if isinstance(changes, dict) else None
-    if isinstance(minion_rets, dict):
-        for minion, minion_ret in minion_rets.items():
-            text = _minion_failure(minion_ret)
-            if text:
-                failures.append('{}: {}'.format(minion, text))
-    return failures
-
-
-def _orch_failures(ret):
-    if not isinstance(ret, dict):
-        return [_trim(ret)]
-    failures = []
-    for job in ret.values():
-        if not isinstance(job, dict):
-            continue
-        job_ret = job.get('return')
-        data = job_ret.get('data') if isinstance(job_ret, dict) else {}
-        if not isinstance(data, dict):
-            if data:
-                failures.append(_trim(data))
-            data = {}
-        for steps in data.values():
-            if not isinstance(steps, dict):
-                failures.append(_trim(steps))
-                continue
-            for step in steps.values():
-                failures.extend(_step_failures(step))
-        if job.get('success') is False and not failures:
-            failures.append('orchestration reported failure: {}'.format(_trim(job.get('return'))))
-    return failures
-
-
-def _failed_steps(ret):
-    steps = []
-    for job in ret.values() if isinstance(ret, dict) else []:
-        job_ret = job.get('return') if isinstance(job, dict) else None
-        data = job_ret.get('data') if isinstance(job_ret, dict) else None
-        for group in data.values() if isinstance(data, dict) else []:
-            if isinstance(group, dict):
-                steps.extend(step for step in group.values() if isinstance(step, dict) and step.get('result') is False)
-    return steps
-
-
-def _result_unknown(ret):
-    # A step that raised (e.g. salt-master restarted while it waited on a queued state run)
-    # never collected the minion's return, so the state run may still have completed.
-    steps = _failed_steps(ret)
-    return bool(steps) and all(str(step.get('comment', '')).startswith(STEP_RAISED) for step in steps)
-
-
-def _recheck_delay(age):
-    return min(RESULT_RECHECK_MAX, max(RESULT_CHECK_DELAY, age / 4))
-
-
-def _check_dispatched(log, now):
-    due = []
-    for path in glob.glob(os.path.join(DISPATCHED_DIR, '*.json')):
-        record = _read_intent(path, log)
-        if not isinstance(record, dict) or not record.get('jid'):
-            _unlink(path, log)
-            continue
-        age = now - record.get('dispatched_at', 0)
-        last_check = record.get('checked_at', record.get('dispatched_at', 0))
-        if now - last_check >= _recheck_delay(age):
-            due.append((last_check, path, record, age))
-    # Least recently checked first, so pushes that are still running can't starve finished ones.
-    for _, path, record, age in sorted(due, key=lambda item: item[:2])[:RESULT_CHECKS_PER_PASS]:
-        jid = record['jid']
-        try:
-            if _report_result(record, age, log):
-                _unlink(path, log)
-            else:
-                record['checked_at'] = now
-                _write_record(path, record, log)
-        except Exception:
-            # Drop the record so one unreadable result can't fail every pass ahead of the drain.
-            log.exception('cannot evaluate result for jid=%s; no longer tracking', jid)
-            _unlink(path, log)
-
-
-def _report_result(record, age, log):
-    """Logs the outcome of a dispatched push. Returns True once the record is finished with."""
-    jid = record['jid']
-    paths = record.get('paths', [])
-    ret = _lookup_jid(jid, log)
-    if not ret:
-        if age > RESULT_MAX_AGE:
-            log.warning('no result for jid=%s after %ds, no longer tracking; paths=%s', jid, age, paths)
-            return True
         return False
-    failures = _orch_failures(ret)
-    if failures and _result_unknown(ret):
-        log.warning('push result unknown jid=%s paths=%s; the orchestration lost track of the state run, which may '
-                    'still have completed (if not, the change will be applied at the next scheduled highstate): %s',
-                    jid, paths, ' | '.join(failures))
-    elif failures:
-        log.error('push failed jid=%s paths=%s; change will be applied at the next scheduled highstate: %s',
-                  jid, paths, ' | '.join(failures))
-    else:
-        log.info('push succeeded jid=%s paths=%s', jid, paths)
+    log.info('dispatch accepted: %s', (result.stdout or '').strip())
     return True
 
 
@@ -335,9 +143,6 @@ def main():
 
     debounce_seconds = int(push.get('debounce_seconds', 30))
 
-    # Outside the lock: lookups are slow and the reactors take the same lock.
-    _check_dispatched(log, time.time())
-
     os.makedirs(PENDING_DIR, exist_ok=True)
     lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
     try:
@@ -403,16 +208,10 @@ def main():
             len(ready), len(deduped), len(combined_actions),
             debounce_duration, all_paths[:20],
         )
-        for action in deduped:
-            log.info('action: %s tgt=%s', 'highstate' if action.get('highstate') else action.get('state'),
-                     action.get('tgt'))
 
-        jid = _dispatch(deduped, log)
-        if jid is None:
+        if not _dispatch(deduped, log):
             log.warning('dispatch failed; leaving intent files in place for retry')
             return 1
-        if jid:
-            _record_dispatch(jid, deduped, all_paths[:20], log)
 
         for path, _ in ready:
             try:

salt/manager/tools/sbin/so-push-drainer_test.py

@@ -1,501 +0,0 @@
-# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
-# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
-# https://securityonion.net/license; you may not use this file except in compliance with the
-# Elastic License 2.0.
-
-import importlib.util
-import json
-import logging
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-import unittest
-from importlib.machinery import SourceFileLoader
-from unittest.mock import MagicMock, patch
-
-HERE = os.path.dirname(os.path.abspath(__file__))
-SCRIPT = os.path.join(HERE, 'so-push-drainer')
-_loader = SourceFileLoader('so_push_drainer', SCRIPT)
-_spec = importlib.util.spec_from_loader('so_push_drainer', _loader)
-drainer = importlib.util.module_from_spec(_spec)
-
-# salt is not installed where these tests run; the drainer only needs salt.client.Caller.
-# Mocked only while the drainer loads: run from the repo root, 'salt' is this repo's salt/ directory.
-_salt = MagicMock()
-with patch.dict(sys.modules, {'salt': _salt, 'salt.client': _salt.client}):
-    _loader.exec_module(drainer)
-
-MASTER = 'manager.localdomain_master'
-JID = '20260930171554259426'
-ASYNC_STDERR = ('[WARNING ] Running in asynchronous mode. Results of this execution may be collected '
-                'by attaching to the master event bus or by examining the master job cache, if '
-                'configured. This execution is running under tag salt/run/{}\n'.format(JID))
-CONFLICT = ('The function "state.sls" is running as PID 372218 and was started at '
-            '2026, Sep 30 17:15:40.466233 with jid 20260930171540466233')
-
-
-def _orch_ret(steps, success=True):
-    return {MASTER: {
-        'fun': 'runner.state.orchestrate',
-        'jid': JID,
-        'return': {'data': {MASTER: steps}, 'outputter': 'highstate', 'retcode': 0 if success else 1},
-        'success': success,
-    }}
-
-
-REFRESH_STEP = {
-    'salt_|-refresh_pillar_1_|-saltutil.refresh_pillar_|-function': {
-        '__id__': 'refresh_pillar_1', 'result': True,
-        'changes': {'ret': {'manager_standalone': True}},
-        'comment': 'Function ran successfully.',
-    },
-}
-
-CONFLICT_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-        '__id__': 'apply_soc_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': [CONFLICT]}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}), success=False)
-
-STATE_FAIL_RET = _orch_ret({
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'test_|-no_license_|-no_license_|-fail_without_changes': {
-                '__id__': 'hydra.enabled_no_license_detected', 'result': False,
-                'comment': 'This is a feature supported only for customers with a valid license.',
-            },
-            'file_|-hydra_conf_|-/opt/so/conf/hydra_|-managed': {'result': True, 'comment': 'ok'},
-        }}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}, success=False)
-
-RAISED = ('An exception occurred in this state: Traceback (most recent call last):\n'
-          '  File "salt/client/__init__.py", line 1934, in pub\n'
-          '    raise AuthenticationError(err_msg)\n'
-          'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-
-RAISED_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False, 'changes': {}, 'comment': RAISED,
-    },
-}), success=False)
-
-SUCCESS_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_telegraf_1_|-apply_telegraf_1_|-state': {
-        '__id__': 'apply_telegraf_1', 'result': True,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'file_|-tgrafconf_|-/opt/so/conf/telegraf/etc/telegraf.conf_|-managed': {'result': True},
-        }}},
-        'comment': 'States ran successfully.',
-    },
-}))
-
-
-class DrainerTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.tmpdir = tempfile.mkdtemp()
-        self.pending = os.path.join(self.tmpdir, 'push_pending')
-        self.dispatched = os.path.join(self.tmpdir, 'push_dispatched')
-        os.makedirs(self.pending)
-        for name, value in (
-            ('PENDING_DIR', self.pending),
-            ('LOCK_FILE', os.path.join(self.pending, '.lock')),
-            ('DISPATCHED_DIR', self.dispatched),
-            ('LOG_FILE', os.path.join(self.tmpdir, 'log', 'so-push-drainer.log')),
-        ):
-            patcher = patch.object(drainer, name, value)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-        self.log = MagicMock()
-
-    def tearDown(self):
-        shutil.rmtree(self.tmpdir, ignore_errors=True)
-
-    def write_json(self, directory, name, data):
-        os.makedirs(directory, exist_ok=True)
-        path = os.path.join(directory, name)
-        with open(path, 'w') as f:
-            if isinstance(data, str):
-                f.write(data)
-            else:
-                json.dump(data, f)
-        return path
-
-    def logged(self, level):
-        return ' '.join(c.args[0] % c.args[1:] for c in getattr(self.log, level).call_args_list)
-
-
-class TestHelpers(DrainerTestCase):
-
-    def test_make_logger_adds_handler_once(self):
-        logger = logging.getLogger('so-push-drainer')
-
-        def close_handlers():
-            for handler in logger.handlers:
-                handler.close()
-            logger.handlers.clear()
-
-        self.addCleanup(close_handlers)
-        logger.handlers.clear()
-        self.assertIs(drainer._make_logger(), logger)
-        drainer._make_logger()
-        self.assertEqual(len(logger.handlers), 1)
-        self.assertTrue(os.path.isdir(os.path.dirname(drainer.LOG_FILE)))
-
-    def test_load_push_cfg(self):
-        with patch.object(drainer.salt.client, 'Caller') as caller:
-            caller.return_value.cmd.return_value = {'enabled': False}
-            self.assertEqual(drainer._load_push_cfg(), {'enabled': False})
-            caller.return_value.cmd.return_value = 'garbage'
-            self.assertEqual(drainer._load_push_cfg(), {})
-
-    def test_read_intent(self):
-        good = self.write_json(self.pending, 'good.json', {'a': 1})
-        bad = self.write_json(self.pending, 'bad.json', '{nope')
-        self.assertEqual(drainer._read_intent(good, self.log), {'a': 1})
-        self.assertIsNone(drainer._read_intent(bad, self.log))
-        with patch('builtins.open', side_effect=RuntimeError('boom')):
-            self.assertIsNone(drainer._read_intent(good, self.log))
-        self.log.exception.assert_called_once()
-
-    def test_dedupe_actions(self):
-        actions = [
-            'not a dict',
-            {'state': 'soc'},
-            {'state': 'soc', 'tgt': '*'},
-            {'state': 'soc', 'tgt': '*', 'tgt_type': 'compound'},
-            {'highstate': True, 'tgt': '*'},
-            {'state': 'soc', 'tgt': 'node1', 'tgt_type': 'glob'},
-        ]
-        self.assertEqual(drainer._dedupe_actions(actions), [actions[2], actions[4], actions[5]])
-
-    def test_trim(self):
-        self.assertEqual(drainer._trim('  text \n'), 'text')
-        self.assertEqual(drainer._trim(['a']), '["a"]')
-        self.assertEqual(drainer._trim(None), 'null')
-        self.assertEqual(drainer._trim('x' * 600), 'x' * drainer.TEXT_LIMIT + '...')
-
-    def test_trim_traceback(self):
-        comment = ('An exception occurred in this state: Traceback (most recent call last):\n'
-                   '  File "salt/client/__init__.py", line 1934, in pub\n'
-                   '    raise AuthenticationError(err_msg)\n'
-                   'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-        self.assertEqual(drainer._trim(comment), 'An exception occurred in this state: '
-                         'salt.exceptions.AuthenticationError: Authentication error occurred.')
-        self.assertEqual(drainer._trim('line one\n  line two\n'), 'line one line two')
-
-    def test_unlink(self):
-        drainer._unlink(os.path.join(self.tmpdir, 'missing'), self.log)
-        self.log.exception.assert_not_called()
-        drainer._unlink(self.tmpdir, self.log)
-        self.log.exception.assert_called_once()
-
-
-class TestDispatch(DrainerTestCase):
-
-    def run_dispatch(self, **kwargs):
-        with patch.object(drainer.subprocess, 'run', **kwargs) as run:
-            jid = drainer._dispatch([{'state': 'soc', 'tgt': '*'}], self.log)
-        return jid, run
-
-    def test_jid_parsed_from_stderr(self):
-        jid, run = self.run_dispatch(return_value=MagicMock(stdout='', stderr=ASYNC_STDERR))
-        self.assertEqual(jid, JID)
-        cmd = run.call_args[0][0]
-        self.assertEqual(cmd[:3], ['salt-run', 'state.orchestrate', 'orch.push_batch'])
-        self.assertIn('--async', cmd)
-
-    def test_jid_parsed_from_stdout(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout=ASYNC_STDERR, stderr=None))
-        self.assertEqual(jid, JID)
-
-    def test_no_jid(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout='unexpected output', stderr=None))
-        self.assertEqual(jid, '')
-        self.assertIn('output=unexpected output', self.logged('warning'))
-
-    def test_failures_return_none(self):
-        for exc in (subprocess.CalledProcessError(1, 'salt-run', 'out', 'err'),
-                    subprocess.TimeoutExpired('salt-run', 60),
-                    RuntimeError('boom')):
-            jid, _ = self.run_dispatch(side_effect=exc)
-            self.assertIsNone(jid)
-
-    def test_record_dispatch(self):
-        drainer._record_dispatch(JID, [{'state': 'soc'}], ['audit:soc.config.licenseKey'], self.log)
-        with open(os.path.join(self.dispatched, JID + '.json')) as f:
-            record = json.load(f)
-        self.assertEqual(record['jid'], JID)
-        self.assertEqual(record['paths'], ['audit:soc.config.licenseKey'])
-        self.assertIn('dispatched_at', record)
-
-    def test_record_dispatch_errors(self):
-        with patch.object(drainer.os, 'makedirs', side_effect=OSError('ro')):
-            drainer._record_dispatch(JID, [], [], self.log)
-        drainer._record_dispatch(JID, [object()], [], self.log)
-        self.assertEqual(self.log.exception.call_count, 2)
-        self.assertFalse(os.path.exists(os.path.join(self.dispatched, JID + '.json')))
-
-
-class TestResults(DrainerTestCase):
-
-    def test_lookup_jid(self):
-        with patch.object(drainer.subprocess, 'run') as run:
-            run.return_value = MagicMock(stdout=json.dumps(SUCCESS_RET))
-            self.assertEqual(drainer._lookup_jid(JID, self.log), SUCCESS_RET)
-            self.assertEqual(run.call_args[0][0], ['salt-run', 'jobs.lookup_jid', JID, '--out=json'])
-            run.return_value = MagicMock(stdout='')
-            self.assertEqual(drainer._lookup_jid(JID, self.log), {})
-            run.return_value = MagicMock(stdout='not json')
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-            run.side_effect = subprocess.TimeoutExpired('salt-run', 60)
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-
-    def test_minion_failure_shapes(self):
-        self.assertEqual(drainer._minion_failure([CONFLICT]), json.dumps([CONFLICT]))
-        self.assertEqual(drainer._minion_failure('Rendering SLS failed'), 'Rendering SLS failed')
-        self.assertEqual(drainer._minion_failure(True), '')
-        self.assertEqual(drainer._minion_failure({'a': {'result': True}}), '')
-
-    def test_orch_failures_conflict(self):
-        failures = drainer._orch_failures(CONFLICT_RET)
-        self.assertEqual(failures[0], 'apply_soc_1: Run failed on minions: manager_standalone')
-        self.assertIn('manager_standalone', failures[1])
-        self.assertIn('is running as PID 372218', failures[1])
-        self.assertEqual(len(failures), 2)
-
-    def test_orch_failures_failed_state(self):
-        failures = drainer._orch_failures(STATE_FAIL_RET)
-        self.assertEqual(len(failures), 2)
-        self.assertIn('hydra.enabled_no_license_detected: This is a feature', failures[1])
-        self.assertNotIn('hydra_conf', failures[1])
-
-    def test_orch_failures_success(self):
-        self.assertEqual(drainer._orch_failures(SUCCESS_RET), [])
-
-    def test_orch_failures_render_error(self):
-        ret = {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS failed"]'])
-
-    def test_orch_failures_not_a_dict(self):
-        self.assertEqual(drainer._orch_failures(['No minions matched']), ['["No minions matched"]'])
-        self.assertEqual(drainer._orch_failures('Runner error'), ['Runner error'])
-
-    def test_orch_failures_data_not_a_dict(self):
-        ret = {MASTER: {'return': {'data': ["Rendering SLS 'orch.push_batch' failed"]}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS \'orch.push_batch\' failed"]'])
-
-    def test_orch_failures_odd_changes(self):
-        for changes in ('Run failed', {'ret': ['manager_standalone']}):
-            ret = _orch_ret({'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-                '__id__': 'apply_soc_1', 'result': False, 'changes': changes, 'comment': 'Run failed on minions',
-            }}, success=False)
-            self.assertEqual(drainer._orch_failures(ret), ['apply_soc_1: Run failed on minions'])
-
-    def test_orch_failures_unparsed(self):
-        self.assertEqual(drainer._orch_failures({MASTER: 'odd'}), [])
-        ret = {MASTER: {'return': 'Exception occurred', 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['orchestration reported failure: Exception occurred'])
-
-    def test_result_unknown(self):
-        self.assertTrue(drainer._result_unknown(RAISED_RET))
-        for ret in (CONFLICT_RET, STATE_FAIL_RET, SUCCESS_RET, ['No minions matched'], {MASTER: 'odd'},
-                    {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}):
-            self.assertFalse(drainer._result_unknown(ret), ret)
-        mixed = _orch_ret(dict(RAISED_RET[MASTER]['return']['data'][MASTER],
-                               **CONFLICT_RET[MASTER]['return']['data'][MASTER]), success=False)
-        self.assertFalse(drainer._result_unknown(mixed))
-
-    def record(self, jid, age, now):
-        return self.write_json(self.dispatched, jid + '.json', {
-            'jid': jid, 'dispatched_at': now - age, 'actions': [], 'paths': ['audit:' + jid],
-        })
-
-    def test_check_dispatched(self):
-        now = time.time()
-        results = {
-            '1_failed': CONFLICT_RET,
-            '2_ok': SUCCESS_RET,
-            '3_pending': {},
-            '4_expired': None,
-            '6_unknown': RAISED_RET,
-        }
-        young = self.record('0_young', 5, now)
-        paths = {jid: self.record(jid, 60, now) for jid in results}
-        paths['4_expired'] = self.record('4_expired', drainer.RESULT_MAX_AGE + 1, now)
-        bad = self.write_json(self.dispatched, '5_bad.json', '{nope')
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: results[jid]):
-            drainer._check_dispatched(self.log, now)
-
-        self.assertTrue(os.path.exists(young))
-        with open(paths['3_pending']) as f:
-            self.assertEqual(json.load(f)['checked_at'], now)
-        for jid in ('1_failed', '2_ok', '4_expired', '6_unknown'):
-            self.assertFalse(os.path.exists(paths[jid]), jid)
-        self.assertFalse(os.path.exists(bad))
-        self.assertIn('push failed jid=1_failed', self.logged('error'))
-        self.assertIn('is running as PID 372218', self.logged('error'))
-        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
-        self.assertIn('no result for jid=4_expired', self.logged('warning'))
-        self.assertIn('push result unknown jid=6_unknown', self.logged('warning'))
-        self.assertIn('AuthenticationError: Authentication error occurred.', self.logged('warning'))
-        self.assertNotIn('6_unknown', self.logged('error'))
-
-    def test_check_dispatched_survives_bad_result(self):
-        now = time.time()
-        bad = self.record('1_bad', 60, now)
-        good = self.record('2_ok', 60, now)
-
-        def orch_failures(ret):
-            if ret == 'boom':
-                raise ValueError('unexpected shape')
-            return []
-
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: 'boom' if jid == '1_bad' else SUCCESS_RET), \
-                patch.object(drainer, '_orch_failures', side_effect=orch_failures):
-            drainer._check_dispatched(self.log, now)
-        self.assertFalse(os.path.exists(bad))
-        self.assertFalse(os.path.exists(good))
-        self.log.exception.assert_called_once()
-        self.assertIn('jid=1_bad', self.log.exception.call_args[0][0] % self.log.exception.call_args[0][1:])
-        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
-
-    def test_recheck_delay(self):
-        self.assertEqual(drainer._recheck_delay(10), drainer.RESULT_CHECK_DELAY)
-        self.assertEqual(drainer._recheck_delay(400), 100)
-        self.assertEqual(drainer._recheck_delay(drainer.RESULT_MAX_AGE), drainer.RESULT_RECHECK_MAX)
-
-    def test_check_dispatched_limit_rotates(self):
-        now = time.time()
-        limit = drainer.RESULT_CHECKS_PER_PASS
-        jids = ['{:02d}'.format(i) for i in range(limit + 2)]
-        for jid in jids:
-            self.record(jid, 60, now)
-        with patch.object(drainer, '_lookup_jid', return_value={}) as lookup:
-            drainer._check_dispatched(self.log, now)
-            self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[:limit])
-            lookup.reset_mock()
-            drainer._check_dispatched(self.log, now + 40)
-        self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[limit:] + jids[:limit - 2])
-
-    def test_check_dispatched_not_blocked_by_running(self):
-        now = time.time()
-        for i in range(drainer.RESULT_CHECKS_PER_PASS):
-            self.record('1_running{}'.format(i), 600, now)
-        done = self.record('2_done', 60, now)
-
-        def lookup(jid, log):
-            return SUCCESS_RET if jid == '2_done' else {}
-
-        with patch.object(drainer, '_lookup_jid', side_effect=lookup) as lookup_jid:
-            drainer._check_dispatched(self.log, now)
-            self.assertTrue(os.path.exists(done))
-            lookup_jid.reset_mock()
-            drainer._check_dispatched(self.log, now + 15)
-        self.assertEqual([c.args[0] for c in lookup_jid.call_args_list], ['2_done'])
-        self.assertFalse(os.path.exists(done))
-        self.assertIn('push succeeded jid=2_done', self.logged('info'))
-
-
-class TestMain(DrainerTestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.cfg = {'enabled': True, 'debounce_seconds': 30}
-        for name, kwargs in (
-            ('_make_logger', {'return_value': self.log}),
-            ('_load_push_cfg', {'side_effect': lambda: self.cfg}),
-            ('_check_dispatched', {}),
-        ):
-            patcher = patch.object(drainer, name, **kwargs)
-            setattr(self, name, patcher.start())
-            self.addCleanup(patcher.stop)
-
-    def intent(self, name, age=60, actions=None, paths=None):
-        now = time.time()
-        return self.write_json(self.pending, name, {
-            'first_touch': now - age - 5, 'last_touch': now - age,
-            'actions': [{'state': 'soc', 'tgt': '*'}] if actions is None else actions,
-            'paths': paths or ['audit:soc.config.licenseKey'],
-        })
-
-    def test_no_pending_dir(self):
-        shutil.rmtree(self.pending)
-        self.assertEqual(drainer.main(), 0)
-        self._load_push_cfg.assert_not_called()
-
-    def test_cfg_error(self):
-        self._load_push_cfg.side_effect = RuntimeError('no salt')
-        self.assertEqual(drainer.main(), 1)
-
-    def test_disabled(self):
-        self.cfg['enabled'] = False
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_not_called()
-
-    def test_no_intents_still_checks_results(self):
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_called_once()
-
-    def test_debounce_and_broken(self):
-        young = self.intent('young.json', age=1)
-        broken = self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer, '_dispatch') as dispatch:
-            self.assertEqual(drainer.main(), 0)
-        dispatch.assert_not_called()
-        self.assertTrue(os.path.exists(young))
-        self.assertFalse(os.path.exists(broken))
-
-    def test_broken_unlink_error_ignored(self):
-        self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_no_usable_actions(self):
-        path = self.intent('empty.json', actions=[{'state': 'soc'}])
-        self.assertEqual(drainer.main(), 0)
-        self.assertFalse(os.path.exists(path))
-        self.intent('empty.json', actions=[{'state': 'soc'}])
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_dispatch_failure_keeps_intents(self):
-        path = self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=None):
-            self.assertEqual(drainer.main(), 1)
-        self.assertTrue(os.path.exists(path))
-
-    def test_dispatch_records_jid(self):
-        soc = self.intent('pillar_soc.json')
-        hs = self.intent('pillar_global.json', actions=[{'highstate': True, 'tgt': '*'}], paths=['audit:global.x'])
-        with patch.object(drainer, '_dispatch', return_value=JID) as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        self.assertEqual(len(dispatch.call_args[0][0]), 2)
-        record.assert_called_once()
-        self.assertEqual(record.call_args[0][0], JID)
-        self.assertEqual(sorted(record.call_args[0][2]), ['audit:global.x', 'audit:soc.config.licenseKey'])
-        self.assertFalse(os.path.exists(soc))
-        self.assertFalse(os.path.exists(hs))
-        self.assertIn('action: highstate tgt=*', self.logged('info'))
-
-    def test_dispatch_without_jid_not_recorded(self):
-        self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=''), \
-                patch.object(drainer, '_record_dispatch') as record, \
-                patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-        record.assert_not_called()
-        self.log.exception.assert_called_once()
-
-
-if __name__ == '__main__':
-    unittest.main()

salt/manager/tools/sbin/soup

@@ -1183,6 +1183,11 @@ 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:")

salt/orch/push_batch.sls

@@ -3,8 +3,6 @@
 {% set BATCH = AUTOAPPLY.batch %}
 {% set BATCH_WAIT = AUTOAPPLY.batch_wait %}
 
-{# queue must be a top-level salt.state arg (kwarg is ignored); an int is max_queue and still fails on conflict #}
-
 {% for action in actions %}
 {%   if action.get('highstate') %}
 apply_highstate_{{ loop.index }}:
@@ -14,7 +12,8 @@ apply_highstate_{{ loop.index }}:
     - highstate: True
     - batch: {{ action.get('batch', BATCH) }}
     - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
 {%   else %}
 refresh_pillar_{{ loop.index }}:
   salt.function:
@@ -30,7 +29,8 @@ apply_{{ action.state | replace('.', '_') }}_{{ loop.index }}:
       - {{ action.state }}
     - batch: {{ action.get('batch', BATCH) }}
     - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
     - require:
       - salt: refresh_pillar_{{ loop.index }}
 {%   endif %}

salt/reactor/push_pillar.sls

@@ -138,7 +138,6 @@ def run():
     # top level so the reactor is robust to either shape.
     event = data.get('data', data)  # noqa: F821 -- data provided by reactor
     setting_id = event.get('setting_id', '')
-    audit_id = event.get('id')
     node_id = (event.get('node_id') or '').strip()
 
     app = _app_from_setting(setting_id)
@@ -151,8 +150,8 @@ def run():
     if not entry:
         LOG.warning(
             'push_pillar: app "%s" is not in pillar_push_map.yaml; change will be '
-            'picked up at the next scheduled highstate (setting_id=%s audit_id=%s)',
-            app, setting_id, audit_id,
+            'picked up at the next scheduled highstate (setting_id=%s)',
+            app, setting_id,
         )
         return {}
 
@@ -166,12 +165,12 @@ def run():
             'node_{}_{}'.format(node_id, app), actions,
             'audit:{}@{}'.format(setting_id, node_id),
         )
-        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s audit_id=%s)',
-                 app, node_id, setting_id, audit_id)
+        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s)',
+                 app, node_id, setting_id)
         return {}
 
     # Branch B: grid-wide app change -> use the map entry's actions as-is.
     actions = list(entry)  # copy to avoid mutating the cache
     _write_intent('pillar_{}'.format(app), actions, 'audit:{}'.format(setting_id))
-    LOG.info('push_pillar: app intent updated for %s (setting_id=%s audit_id=%s)', app, setting_id, audit_id)
+    LOG.info('push_pillar: app intent updated for %s (setting_id=%s)', app, setting_id)
     return {}

salt/soc/defaults.yaml

@@ -1445,7 +1445,7 @@ soc:
             default:
               - repo: https://github.com/Security-Onion-Solutions/securityonion-resources
                 license: Elastic-2.0
-                folder: sigma/stable
+                folder: sigma
                 community: true
                 rulesetName: securityonion-resources
               - repo: file:///nsm/rules/custom-local-repos/local-sigma
@@ -1455,7 +1455,7 @@ soc:
             airgap:
               - repo: file:///nsm/rules/detect-sigma/repos/securityonion-resources
                 license: Elastic-2.0
-                folder: sigma/stable
+                folder: sigma
                 community: true
                 rulesetName: securityonion-resources
               - repo: file:///nsm/rules/custom-local-repos/local-sigma
@@ -1467,6 +1467,8 @@ soc:
             - emerging_threats_addon
           useEsql: false
           esqlCaseInsensitive: true
+          esqlQueryDelaySeconds: 30
+          esqlCorrelationAllowanceSeconds: 600
         elastic:
           hostUrl:
           remoteHostUrls: []
@@ -2679,8 +2681,11 @@ 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.ruleset so_detection.isEnabled | groupby so_detection.category | groupby so_detection.product"
+              query: "so_detection.language:sigma | groupby so_detection.ruleType | 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"
+              description: Show Sigma correlation Detections
             - name: "Detection Type - YARA (Strelka)"
               query: "so_detection.language:yara | groupby so_detection.ruleset so_detection.isEnabled"
               description: Show all YARA detections, which are used by Strelka
@@ -2764,7 +2769,7 @@ soc:
             elastalert: |
               # This is a Sigma rule template, which uses YAML. Replace all template values with your own values.
               # The id (UUIDv4) is pregenerated and can safely be used.
-              # Click "Convert" to convert the Sigma rule to use Security Onion field mappings within an EQL query
+              # Click "Convert" to convert the Sigma rule to use Security Onion field mappings within a backend query
               #
               # Rule Creation Guide: https://github.com/SigmaHQ/sigma/wiki/Rule-Creation-High%E2%80%90Level-Guide
               # Logsources: https://sigmahq.io/docs/basics/log-sources.html
@@ -2794,6 +2799,58 @@ soc:
                           - ' -priv'
                   condition: all of selection_*
               level: 'high'  # info | low | medium | high | critical
+            elastalert_correlation: |
+              # Sigma correlation rule; requires ES|QL (useEsql).
+              # First document: the correlation. Following documents: the rules it references.
+              #
+              # Types: event_count, value_count, temporal, value_sum, value_avg, value_median, value_percentile.
+              # Correlation Guide: https://sigmahq.io/docs/meta/correlations.html
+              # Logsources: https://sigmahq.io/docs/basics/log-sources.html
+
+              title: 'A Short Capitalized Title With Less Than 50 Characters'
+              id: [publicId]
+              status: 'experimental'
+              description: |
+                  Describe what the correlation finds and, importantly, why relating these
+                  events is more meaningful than either of them alone.
+              references:
+                - 'https://local.invalid'
+              author: '@SecurityOnion'
+              date: '[today]'
+              tags:
+                - detection.threat_hunting
+                - attack.technique_id
+              correlation:
+                  type: value_count
+                  rules:
+                      - example_base_rule  # the 'name' of the rule below
+                  group-by:
+                      - source.ip
+                  # Xs, Xm, Xh, Xd or Xw.
+                  timespan: 10m
+                  condition:
+                      field: dns.query.name
+                      gte: 40
+              falsepositives:
+                - 'Describe the benign activity that also produces this pattern'
+              # Placeholders: %count%, %start%, %end%, %duration%, or any field.
+              summary: '%count% distinct names queried by %source.ip% in %duration%'
+              level: 'medium'  # info | low | medium | high | critical
+              ---
+              title: 'Base Event'
+              # referenced by 'name' (or 'id')
+              name: example_base_rule
+              description: 'The single event that the correlation aggregates.'
+              logsource:
+                category: network
+                service: dns
+              detection:
+                  selection:
+                      dns.query.name|exists: true
+                  condition: selection
+              # Carried into the alert.
+              fields:
+                - dns.query.name
         assistant:
           enabled: false
           investigationPrompt: Investigate Alert ID {socId}

salt/soc/files/soc/sigma_pipelines/sigma_esql_pipeline.yml

@@ -475,3 +475,10 @@ transformations:
       - type: logsource
         product: kratos
 
+# SOC reads the mapped group-by columns from this output
+postprocessing:
+    - id: esql_correlation_group_by
+      type: template
+      template: '{{ {"query": query, "group_by": rule.group_by} | tojson }}'
+      rule_conditions:
+      - type: is_sigma_correlation_rule

salt/soc/files/soc/sigma_pipelines/sigma_so_pipeline.yml

@@ -14,6 +14,15 @@ transformations:
         - process.args
         - related.ip
         - dns.resolved_ip
+    # always lowercase: matched exactly with the indexed ':' operator
+    - id: case_sensitive_categorization_fields
+      type: set_state
+      key: case_insensitive_exempt_fields
+      val:
+        - tags
+        - event.category
+        - event.type
+        - event.kind
     # Not every source maps .caseless; EQL/ES|QL already match case-insensitively.
     - id: caseless_to_parent_fields
       type: field_name_mapping
@@ -118,6 +127,9 @@ transformations:
       valid_hash_algos: ["MD5", "SHA1", "SHA256", "SHA512", "IMPHASH"]
       field_prefix: "file"
       drop_algo_prefix: False
+      # ecs_windows renamed Hashes; pySigma 1.5+ parses only these
+      field_to_parse:
+        - winlog.event_data.Hashes
       field_name_conditions:
         - type: include_fields
           fields:

salt/soc/merged.map.jinja

@@ -8,6 +8,7 @@
 {% from 'elasticsearch/config.map.jinja' import ELASTICSEARCH_NODES %}
 {% from 'manager/map.jinja' import MANAGERMERGED %}
 {% from 'telegraf/map.jinja' import TELEGRAFMERGED %}
+{% from 'elastalert/map.jinja' import ELASTALERTMERGED %}
 {%- set PG_ENTRY = salt['pillar.get']('telegraf:postgres_creds:' ~ grains.id, {}) %}
 {%- set PG_USER = PG_ENTRY.get('user', '') %}
 {%- set PG_PASS = PG_ENTRY.get('pass', '') %}
@@ -63,6 +64,10 @@
 {%   do SOCMERGED.config.server.modules.elastalertengine.update({'enabledSigmaRules': SOCMERGED.config.server.modules.elastalertengine.enabledSigmaRules.default}) %}
 {% endif %}
 
+{# correlation schedules follow ElastAlert's run_every #}
+{% set run_every = ELASTALERTMERGED.config.run_every %}
+{% do SOCMERGED.config.server.modules.elastalertengine.update({'elastAlertRunEverySeconds': run_every.get('minutes', 0) * 60 + run_every.get('seconds', 0)}) %}
+
 {# set elastalertengine.rulesRepos, strelkaengine.rulesRepos, and suricataengine.rulesetSources based on airgap or not #}
 {% if GLOBALS.airgap %}
 {%   do SOCMERGED.config.server.modules.elastalertengine.update({'rulesRepos': SOCMERGED.config.server.modules.elastalertengine.rulesRepos.airgap}) %}
@@ -80,6 +85,12 @@
 {%   do SOCMERGED.config.server.update({'airgapEnabled': false}) %}
 {% endif %}
 
+{# correlation authoring requires ES|QL #}
+{% if not SOCMERGED.config.server.modules.elastalertengine.useEsql %}
+{%   do SOCMERGED.config.server.client.detection.templateDetections.pop('elastalert_correlation', None) %}
+{%   do SOCMERGED.config.server.client.detections.update({'queries': SOCMERGED.config.server.client.detections.queries | rejectattr('name', 'equalto', 'Detection Type - Sigma (Elastalert) - Correlations') | list}) %}
+{% endif %}
+
 {# Define the postgresmetrics module if telegraf is setup to only use Postgres #}
 {% if TELEGRAFMERGED.output != 'INFLUXDB' and PG_USER and PG_PASS %}
 {%   do SOCMERGED.config.server.modules.update({

salt/soc/soc_soc.yaml

@@ -401,15 +401,27 @@ soc:
             advanced: False
             helpLink: sigma
           useEsql:
-            description: "(Pre-release) Use Elasticsearch Piped Query Language (ES|QL) instead of EQL (Elastic Query Language) for Elasticsearch queries. The Sigma converter will output ES|QL instead of EQL, allowing support for correlations."
+            description: "(Pre-release) Use Elasticsearch Piped Query Language (ES|QL) instead of EQL (Elastic Query Language) for Elasticsearch queries. The Sigma converter will output ES|QL instead of EQL, allowing support for correlations. Switching back to EQL is not supported for correlations."
             global: True
             advanced: True
             forcedType: bool
           esqlCaseInsensitive:
-            description: "Match string values case-insensitively when converting Sigma rules. Applies to ES|QL only"
+            description: "Match string values case-insensitively when converting Sigma rules, and group correlation values regardless of case. Applies to ES|QL only"
             global: True
             advanced: True
             forcedType: bool
+          esqlQueryDelaySeconds:
+            description: "Seconds ES|QL rules search behind now, so unsearchable events aren't missed. Delays alerts by the same amount. Set at least the longest index refresh interval. ES|QL only."
+            global: True
+            advanced: True
+            forcedType: int
+            helpLink: sigma
+          esqlCorrelationAllowanceSeconds:
+            description: "Extra seconds of arrivals each correlation run re-reads beyond its timespan, so a burst whose events arrive spread out is still counted together. Correlations below a threshold (lt, lte, eq, neq) and value_avg or value_percentile correlations count only the timespan ending this much plus esqlQueryDelaySeconds ago, so they alert this much later. ES|QL only."
+            global: True
+            advanced: True
+            forcedType: int
+            helpLink: sigma
         elastic:
           index:
             description: Comma-separated list of indices or index patterns (wildcard "*" supported) that SOC will search for records.
@@ -1025,7 +1037,7 @@ soc:
             forcedType: int
           automationSettings:
             tickIntervalSeconds:
-              description: How often, in seconds, the automation scheduler checks for automations that are due to run. Must be greater than 0.
+              description: How often, in seconds, the automation scheduler checks for automations that are due to run. Must be greater than 0. This value is also the default interval for new automations, however admins can override individual automation intervals to a longer value via the Agent Studio.
               global: True
               advanced: True
               forcedType: int

setup/so-verify

@@ -67,7 +67,7 @@ log_has_errors() {
         grep -vE "Reading first line of patchfile" | \
         grep -vE "Command failed with exit code" | \
         grep -vE "Running scope as unit" | \
-        grep -vE "securityonion-resources/sigma/stable" | \
+        grep -vE "securityonion-resources/sigma/" | \
         grep -vE "remove_failed_vm.sls" | \
         grep -vE "failed to copy: httpReadSeeker" | \
         grep -vE "Error response from daemon: failed to resolve reference" | \