Merge branch 'sigma-pipeline-dir' into esql-correlations

EVTX Import cleanup

salt/common/tools/sbin/so-log-check

@@ -241,7 +241,7 @@ if [[ $EXCLUDE_KNOWN_ERRORS == 'Y' ]]; then
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|marked for removal"           # docker container getting recycled
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|tcp 127.0.0.1:6791: bind: address already in use" # so-elastic-fleet agent restarting. Seen starting w/ 8.18.8 https://github.com/elastic/kibana/issues/201459
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|TransformTask\] \[logs-.*user so_kibana lacks the required permissions" # Known issue with integrations starting transform jobs that are explicitly not allowed to start as a system user
-    EXCLUDED_ERRORS="$EXCLUDED_ERRORS|manifest unknown"             # appears in so-dockerregistry log for so-tcpreplay following docker upgrade to 29.2.1-1
+    EXCLUDED_ERRORS="$EXCLUDED_ERRORS|manifest unknown"             # so-dockerregistry logs a tag lookup miss during image copy; not tied to one docker version
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|Could not index event to Elasticsearch.*\"version\" => \"9.0.8\"" # Expected during Elastic upgrade temporarily, as policies referencing older pipelines are updated
 fi
 

salt/docker/init.sls

@@ -18,10 +18,10 @@ dockergroup:
 dockerheldpackages:
   pkg.installed:
     - pkgs:
-      - containerd.io: 2.2.1-1.el9
-      - docker-ce: 3:29.2.1-1.el9
-      - docker-ce-cli: 1:29.2.1-1.el9
-      - docker-ce-rootless-extras: 29.2.1-1.el9
+      - containerd.io: 2.3.6-1.el9
+      - docker-ce: 3:29.8.1-1.el9
+      - docker-ce-cli: 1:29.8.1-1.el9
+      - docker-ce-rootless-extras: 29.8.1-1.el9
     - hold: True
     - update_holds: True
 

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/securityonion-es.py

@@ -6,9 +6,14 @@
 # Elastic License 2.0.
 
 
+from datetime import datetime
 from time import gmtime, strftime
+import hashlib
+import ipaddress
+import re
 import requests,json
-from elastalert.alerts import Alerter
+from elastalert.alerts import Alerter, DateTimeEncoder
+from elastalert.util import EAException, elastalert_logger
 
 import urllib3
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@@ -19,17 +24,152 @@ class SecurityOnionESAlerter(Alerter):
     """
 
     required_options = set(['detection_title', 'sigma_level'])
-    optional_fields = ['sigma_category', 'sigma_product', 'sigma_service']
+    optional_fields = ['sigma_category', 'sigma_product', 'sigma_service', 'sigma_correlation']
+
+    count_labels = {
+        'event_count': '%count% events',
+        'value_count': '%count% distinct values',
+        'event_type_count': '%count% correlated rules matched',
+        'value_sum': 'total %count%',
+        'value_avg': 'average %count%',
+        'value_percentile': 'percentile %count%',
+        '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'}
+
+    @staticmethod
+    def lookup(doc, dotted):
+        """ Resolve a dotted path; ES|QL columns arrive nested. """
+        node = doc
+        for part in dotted.split('.'):
+            if not isinstance(node, dict) or part not in node:
+                return None
+            node = node[part]
+        return node
+
+    def query_keys(self):
+        """ 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 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')}"
+
+        return hashlib.sha256(key.encode('utf-8')).hexdigest()
+
+    def group(self, match):
+        """ Group-by values joined as ElastAlert joins them for the realert silence key. """
+        return ', '.join(str(self.lookup(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; invalid IPs are skipped, as they fail the ip mapping. """
+        related = {}
+        for key in self.query_keys():
+            bucket = self.related_bucket(key)
+            if not bucket:
+                continue
+            value = self.lookup(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
+                related.setdefault(bucket, {})[str(v)] = None
+        return {bucket: list(values) for bucket, values in related.items()}
+
+    def event_data(self, match):
+        """ The match without the compound query_key field, which ES would map by its last part (e.g. .ip). """
+        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'}"
+
+    @staticmethod
+    def to_dt(value):
+        # ES|QL gives ISO strings; ElastAlert parses @timestamp, except on a retried alert.
+        return value if isinstance(value, datetime) else datetime.fromisoformat(value)
+
+    def summary(self, match):
+        """ One-line correlation summary; None for single-event rules. """
+        if not self.is_correlation(match):
+            return None
+
+        start = self.to_dt(match['window_start'])
+        end = self.to_dt(match['@timestamp'])
+        name = next((f for f in self.count_labels if f in match), None)
+        values = {
+            'count': self.format_count(match.get(name)),
+            '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"{self.count_labels.get(name, '%count%')}{' for ' + groups if groups else ''} in %duration%"
+
+        def fill(m):
+            if m[1] in values:
+                return values[m[1]]
+            value = self.lookup(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"}
-
-            creds = None
-            if 'es_username' in self.rule and 'es_password' in self.rule:
-                creds = (self.rule['es_username'], self.rule['es_password'])
-
             # Start building the rule dict
             rule_info = {
                 "name": self.rule['detection_title'],
@@ -42,22 +182,74 @@ class SecurityOnionESAlerter(Alerter):
                 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']
+            }
+
+            reason = self.summary(match)
+            if reason:
+                event_info["reason"] = reason
+
             # Construct the payload with the conditional rule_info
             payload = {
-                "tags": "alert",
+                "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']
-                },
+                "event": event_info,
                 "sigma_level": self.rule['sigma_level'],
-                "event_data": match,
+                "event_data": self.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)
+
+            keys = self.query_keys()
+            if keys and self.is_correlation(match):
+                payload["labels"] = {
+                    "correlation_group_by": ', '.join(keys),
+                    "correlation_group": self.group(match),
+                }
+                related = self.related(match)
+                if related:
+                    payload["related"] = related
+            alert_id = self.alert_id(match)
+            # _create returns 409 on a repeat id; EAException makes ElastAlert retry
+            url = (f"https://{self.rule['es_host']}:{self.rule['es_port']}"
+                   f"/logs-detections.alerts-so/_create/{alert_id}")
+            response = self.put_alert(url, payload)
+            if response.status_code == 400:
+                # mapping rejections come from event_data; retry with it as unindexed text
+                rejection = response.text[:500]
+                payload = self.without_event_data(payload, rejection)
+                response = self.put_alert(url, payload)
+                if response.status_code == 400:
+                    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'], response.text[:500], rejection)
+                    continue
+                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)
+            if response.status_code != 409 and not response.ok:
+                raise EAException(f"Unable to write alert: {response.status_code} {response.text[:500]}")
+
+    def put_alert(self, url, payload):
+        creds = None
+        if 'es_username' in self.rule and 'es_password' in self.rule:
+            creds = (self.rule['es_username'], self.rule['es_password'])
+        try:
+            return requests.put(url, data=json.dumps(payload, cls=DateTimeEncoder),
+                                headers={"Content-Type": "application/json"}, verify=False, auth=creds)
+        except requests.RequestException as e:
+            raise EAException(f"Unable to write alert: {e}")
+
+    @staticmethod
+    def without_event_data(payload, rejection):
+        """ event_data moved to event.original; the tag keeps Fleet's final pipeline from removing 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,211 @@
+# 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 logging
+import os
+import sys
+import types
+import unittest
+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
+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
+
+    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')
+    package = types.ModuleType('elastalert')
+    package.alerts = alerts
+    package.util = util
+    sys.modules.update({'elastalert': package, 'elastalert.alerts': alerts, 'elastalert.util': util})
+
+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,
+    'summary_template': '%count% failed network logons to %host.name% from %source.ip% in %duration%',
+}
+
+
+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': 'sa-delta-02-jb'},
+        'source': {'ip': '192.168.198.149'},
+        '_id': '6d1c',
+        'num_hits': 1,
+        'num_matches': 1,
+    }
+
+
+class TestSecurityOnionESAlerter(unittest.TestCase):
+
+    def send(self, rule, match):
+        """ Run alert() and return the payload it wrote and the URL it wrote to. """
+        alerter = es.SecurityOnionESAlerter(rule)
+        response = MagicMock(status_code=201, ok=True)
+        with patch.object(es.requests, 'put', return_value=response) as put:
+            alerter.alert([match])
+        self.assertEqual(put.call_count, 1)
+        return json.loads(put.call_args.kwargs['data']), put.call_args.args[0]
+
+    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'] = 'sa-delta-02-jb, 192.168.198.149'
+        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': 'sa-delta-02-jb'})
+        self.assertEqual(payload['event_data']['source'], {'ip': '192.168.198.149'})
+        self.assertEqual(payload['labels'], {'correlation_group_by': 'host.name, source.ip', 'correlation_group': 'sa-delta-02-jb, 192.168.198.149'})
+        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')
+        # 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': 'josh@local.invalid'}}
+
+        payload, _ = self.send(rule, match)
+
+        self.assertEqual(payload['labels'], {'correlation_group_by': 'user.name', 'correlation_group': 'josh@local.invalid'})
+        self.assertEqual(payload['related'], {'user': ['josh@local.invalid']})
+        self.assertEqual(payload['event']['reason'], '3 failed SOC logins for josh@local.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': ['admmig', 'svc', 'admmig']}},
+                      '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': ['admmig', 'svc']})
+        self.assertEqual(payload['labels']['correlation_group'], "['admmig', 'svc', 'admmig'], not-an-ip, example.com")
+
+        payload, _ = self.send(dict(BASE_RULE, query_key='dns.highest_registered_domain'), match)
+
+        self.assertNotIn('related', payload)
+
+    def test_plain_rule_has_no_correlation_fields(self):
+        # 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(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.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. """
+        alerter = es.SecurityOnionESAlerter(rule)
+        with patch.object(es.requests, 'put', side_effect=responses) as put:
+            alerter.alert([match])
+        return [json.loads(c.kwargs['data']) for c in put.call_args_list]
+
+    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': 'josh@local.invalid'}}
+        rejected = MagicMock(status_code=400, ok=False, text='{"error":{"type":"document_parsing_exception"}}')
+
+        first, second = self.send_responses(rule, match, [rejected, MagicMock(status_code=201, ok=True)])
+
+        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: {"error"'))
+        # 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"}}')
+        match = {'@timestamp': '2026-09-30T16:50:00+00:00', '_id': 'abc'}
+
+        # no EAException, so no retry
+        payloads = self.send_responses(dict(BASE_RULE), match, [rejected, rejected])
+
+        self.assertEqual(len(payloads), 2)
+
+
+if __name__ == '__main__':
+    unittest.main()

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/elasticfleet/files/integrations/grid-nodes_general/import-evtx-logs.json

@@ -29,7 +29,7 @@
               "\\.gz$"
             ],
             "include_files": [],
-            "processors": "- dissect:\n    tokenizer: \"/nsm/import/%{import.id}/evtx/%{import.file}\"\n    field: \"log.file.path\"\n    target_prefix: \"\"\n- decode_json_fields:\n    fields: [\"message\"]\n    target: \"\"\n- drop_fields:\n    fields: [\"host\"]\n    ignore_missing: true\n- add_fields:\n    target: data_stream\n    fields:\n      type: logs\n      dataset: system.security\n- add_fields:\n    target: event\n    fields:\n      dataset: system.security\n      module: system\n      imported: true\n- add_fields:\n    target: \"@metadata\"\n    fields:\n        pipeline: logs-system.security-2.22.3\n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-Sysmon/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.sysmon_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.sysmon_operational\n          module: windows\n          imported: true\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.sysmon_operational-3.9.0\n- if:\n    equals:\n      winlog.channel: 'Application'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.application\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.application-2.22.3\n- if:\n    equals:\n      winlog.channel: 'System'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: event\n        fields:\n          dataset: system.system\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-system.system-2.22.3\n    \n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-PowerShell/Operational'\n  then: \n    - add_fields:\n        target: data_stream\n        fields:\n          dataset: windows.powershell_operational\n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.powershell_operational\n          module: windows\n    - add_fields:\n        target: \"@metadata\"\n        fields:\n            pipeline: logs-windows.powershell_operational-3.9.0\n- add_fields:\n    target: data_stream\n    fields:\n        dataset: import",
+            "processors": "- dissect:\n    tokenizer: \"/nsm/import/%{import.id}/evtx/%{import.file}\"\n    field: \"log.file.path\"\n    target_prefix: \"\"\n- decode_json_fields:\n    fields: [\"message\"]\n    target: \"\"\n- add_fields:\n    target: event\n    fields:\n      dataset: windows.forwarded\n      module: windows\n      imported: true\n- add_fields:\n    target: \"@metadata\"\n    fields:\n        pipeline: import.evtx\n- if:\n    equals:\n      winlog.channel: 'Security'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: system.security\n          module: system\n- if:\n    equals:\n      winlog.channel: 'Windows PowerShell'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.powershell\n          module: windows\n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-Sysmon/Operational'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.sysmon_operational\n          module: windows\n          imported: true\n- if:\n    equals:\n      winlog.channel: 'Application'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: system.application\n          module: system\n- if:\n    equals:\n      winlog.channel: 'System'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: system.system\n          module: system\n    \n- if:\n    equals:\n      winlog.channel: 'Microsoft-Windows-PowerShell/Operational'\n  then: \n    - add_fields:\n        target: event\n        fields:\n          dataset: windows.powershell_operational\n          module: windows\n- add_fields:\n    target: data_stream\n    fields:\n        type: logs\n        dataset: import",
             "tags": [
               "import"
             ],

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/global@custom

@@ -99,7 +99,7 @@
         },
         {
             "set": {
-                "if": "ctx.tags != null && ctx.tags.contains('import')",
+                "if": "ctx.tags != null && ctx.tags.contains('import') && ctx._index != null && ctx._index.startsWith('logs-import-')",
                 "override": true,
                 "field": "data_stream.dataset",
                 "value": "import"
@@ -107,7 +107,7 @@
         },
         {
             "set": {
-                "if": "ctx.tags != null && ctx.tags.contains('import')",
+                "if": "ctx.tags != null && ctx.tags.contains('import') && ctx._index != null && ctx._index.startsWith('logs-import-')",
                 "override": true,
                 "field": "data_stream.namespace",
                 "value": "so"

salt/elasticsearch/files/ingest/import.evtx

@@ -0,0 +1,31 @@
+{
+  "description" : "import.evtx: normalize imported EVTX and reroute to logs-<dataset>-import",
+  "processors" : [
+    { "script":  {
+        "description": "Host from the event, not the importing node",
+        "lang": "painless",
+        "source": "Map host = ['os': ['type': 'windows', 'family': 'windows', 'platform': 'windows']]; def cn = ctx.winlog?.computer_name; if (cn != null && cn.toString().length() > 0) { String name = cn.toString(); int dot = name.indexOf('.'); if (dot > 0) { name = name.substring(0, dot); } host.put('hostname', name); host.put('name', name.toLowerCase()); } ctx.host = host;"
+    } },
+    { "script":  {
+        "description": "String event IDs, as Winlogbeat sends",
+        "lang": "painless",
+        "source": "if (ctx.winlog?.event_id != null) { ctx.winlog.event_id = ctx.winlog.event_id.toString(); } if (ctx.event?.code != null) { ctx.event.code = ctx.event.code.toString(); }"
+    } },
+    { "script":  {
+        "description": "Unnamed <Data> to param1..N, as Winlogbeat",
+        "lang": "painless",
+        "if": "ctx.winlog?.event_data?.Data instanceof Map && ctx.winlog.event_data.Data['#text'] != null",
+        "source": "def t = ctx.winlog.event_data.Data['#text']; List vals = t instanceof List ? t : [t]; for (int i = 0; i < vals.size(); i++) { ctx.winlog.event_data['param' + (i + 1)] = vals.get(i); } ctx.winlog.event_data.remove('Data');"
+    } },
+    { "script":  {
+        "description": "String values and LF line endings, as Winlogbeat",
+        "lang": "painless",
+        "if": "ctx.winlog?.event_data instanceof Map || ctx.winlog?.user_data instanceof Map",
+        "source": "String lf = String.valueOf((char) 10); String crlf = String.valueOf((char) 13) + lf; for (def key : ['event_data', 'user_data']) { def m = ctx.winlog[key]; if (!(m instanceof Map)) { continue; } for (def e : m.entrySet()) { def v = e.getValue(); if (v instanceof String) { e.setValue(v.replace(crlf, lf)); } else if (v instanceof Number || v instanceof Boolean) { e.setValue(v.toString()); } } }"
+    } },
+    { "set":     { "description": "event.kind, as Winlogbeat", "field": "event.kind", "value": "event", "override": false } },
+    { "set":     { "field": "data_stream.dataset", "copy_from": "event.dataset", "override": true, "ignore_empty_value": true } },
+    { "set":     { "field": "data_stream.namespace", "value": "import", "override": true } },
+    { "reroute": { "dataset": "{{data_stream.dataset}}", "namespace": "{{data_stream.namespace}}" } }
+  ]
+}

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-yaml.py

@@ -42,7 +42,8 @@ def loadYaml(filename):
     try:
         with open(filename, "r") as file:
             content = file.read()
-            return yaml.safe_load(content)
+            loaded = yaml.safe_load(content)
+            return loaded if loaded is not None else {}
     except FileNotFoundError:
         print(f"File not found: {filename}", file=sys.stderr)
         sys.exit(1)

salt/manager/tools/sbin/so-yaml_test.py

@@ -95,6 +95,20 @@ class TestRemove(unittest.TestCase):
         expected = "key1:\n  child1: 123\n  child2:\n    deep2: ab\nkey2: false\n"
         self.assertEqual(actual, expected)
 
+    def test_remove_empty_file(self):
+        filename = "/tmp/so-yaml_test-remove-empty.yaml"
+        file = open(filename, "w")
+        file.close()
+
+        code = soyaml.remove([filename, "key1"])
+        self.assertEqual(code, 0)
+
+        file = open(filename, "r")
+        actual = file.read()
+        file.close()
+
+        self.assertEqual(actual, "{}\n")
+
     def test_remove_missing_args(self):
         with patch('sys.exit', new=MagicMock()) as sysmock:
             with patch('sys.stderr', new=StringIO()) as mock_stderr:
@@ -294,6 +308,36 @@ class TestRemove(unittest.TestCase):
         expected = "key1:\n  child1: 123\n  child2:\n    deep1: 45\n    deep2: d\nkey2: false\nkey3:\n- e\n- f\n- g\n"
         self.assertEqual(actual, expected)
 
+    def test_add_empty_file(self):
+        filename = "/tmp/so-yaml_test-add-empty.yaml"
+        file = open(filename, "w")
+        file.close()
+
+        code = soyaml.add([filename, "telegraf.output", "BOTH"])
+        self.assertEqual(code, 0)
+
+        file = open(filename, "r")
+        actual = file.read()
+        file.close()
+
+        expected = "telegraf:\n  output: BOTH\n"
+        self.assertEqual(actual, expected)
+
+    def test_add_empty_file_simple(self):
+        filename = "/tmp/so-yaml_test-add-empty-simple.yaml"
+        file = open(filename, "w")
+        file.close()
+
+        code = soyaml.add([filename, "telegraf", "BOTH"])
+        self.assertEqual(code, 0)
+
+        file = open(filename, "r")
+        actual = file.read()
+        file.close()
+
+        expected = "telegraf: BOTH\n"
+        self.assertEqual(actual, expected)
+
     def test_replace_missing_arg(self):
         with patch('sys.exit', new=MagicMock()) as sysmock:
             with patch('sys.stderr', new=StringIO()) as mock_stderr:
@@ -346,6 +390,21 @@ class TestRemove(unittest.TestCase):
         expected = "key1:\n  child1: 123\n  child2:\n    deep1: 46\nkey2: false\nkey3:\n- e\n- f\n- g\n"
         self.assertEqual(actual, expected)
 
+    def test_replace_empty_file(self):
+        filename = "/tmp/so-yaml_test-replace-empty.yaml"
+        file = open(filename, "w")
+        file.close()
+
+        code = soyaml.replace([filename, "telegraf.output", "BOTH"])
+        self.assertEqual(code, 0)
+
+        file = open(filename, "r")
+        actual = file.read()
+        file.close()
+
+        expected = "telegraf:\n  output: BOTH\n"
+        self.assertEqual(actual, expected)
+
     def test_convert(self):
         self.assertEqual(soyaml.convertType("foo"), "foo")
         self.assertEqual(soyaml.convertType("foo.bar"), "foo.bar")
@@ -506,6 +565,18 @@ class TestRemove(unittest.TestCase):
             self.assertEqual(result, 2)
             self.assertEqual("", mock_stdout.getvalue())
 
+    def test_get_empty_file(self):
+        with patch('sys.stdout', new=StringIO()) as mock_stdout:
+            with patch('sys.stderr', new=StringIO()) as mock_stderr:
+                filename = "/tmp/so-yaml_test-get-empty.yaml"
+                file = open(filename, "w")
+                file.close()
+
+                result = soyaml.get([filename, "telegraf.output"])
+                self.assertEqual(result, 2)
+                self.assertEqual("", mock_stdout.getvalue())
+                self.assertIn("Key 'telegraf.output' not found by so-yaml.py", mock_stderr.getvalue())
+
     def test_get_usage(self):
         with patch('sys.exit', new=MagicMock()) as sysmock:
             with patch('sys.stderr', new=StringIO()) as mock_stderr:
@@ -991,3 +1062,29 @@ class TestLoadYaml(unittest.TestCase):
                     soyaml.loadYaml("/tmp/so-yaml_test-unreadable.yaml")
                     sysmock.assert_called_with(1)
                     self.assertIn("Error reading file", mock_stderr.getvalue())
+
+    def test_load_yaml_empty_file(self):
+        filename = "/tmp/so-yaml_test-load-empty.yaml"
+        file = open(filename, "w")
+        file.close()
+
+        result = soyaml.loadYaml(filename)
+        self.assertEqual(result, {})
+
+    def test_load_yaml_whitespace_only(self):
+        filename = "/tmp/so-yaml_test-load-whitespace.yaml"
+        file = open(filename, "w")
+        file.write("   \n\n  \n")
+        file.close()
+
+        result = soyaml.loadYaml(filename)
+        self.assertEqual(result, {})
+
+    def test_load_yaml_comments_only(self):
+        filename = "/tmp/so-yaml_test-load-comments.yaml"
+        file = open(filename, "w")
+        file.write("# Just a comment\n# Another comment\n")
+        file.close()
+
+        result = soyaml.loadYaml(filename)
+        self.assertEqual(result, {})

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
 
+  # Extract the Sigma rule type for existing detections (Single vs. Correlation)
+  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/repo/client/map.jinja

@@ -9,7 +9,7 @@
        'epel-testing.repo',
        'saltstack.repo',
        'salt-latest.repo',
-       'wazuh.repo'
+       'wazuh.repo',
        'Rocky-Base.repo',
        'Rocky-CR.repo',
        'Rocky-Debuginfo.repo',

salt/soc/config.sls

@@ -120,19 +120,30 @@ crondetectionsbackup:
 
 socsigmafinalpipeline:
   file.managed:
-    - name: /opt/so/conf/soc/sigma_final_pipeline.yaml
+    - name: /opt/so/conf/soc/sigma_pipelines/sigma_final_pipeline.yml
     - source: salt://soc/files/soc/sigma_final_pipeline.yaml
     - user: 939
     - group: 939
     - mode: 600
+    - makedirs: True
 
-socsigmasopipeline:
-  file.managed:
-    - name: /opt/so/conf/soc/sigma_so_pipeline.yaml
-    - source: salt://soc/files/soc/sigma_so_pipeline.yaml
+# sigma-cli loads every *.yml here; clean removes anything else
+socsigmapipelines:
+  file.recurse:
+    - name: /opt/so/conf/soc/sigma_pipelines
+    - source: salt://soc/files/soc/sigma_pipelines
     - user: 939
     - group: 939
-    - mode: 600
+    - file_mode: 600
+    - clean: True
+    - require:
+      - file: socsigmafinalpipeline
+
+socsigmapipelinesold:
+  file.absent:
+    - names:
+      - /opt/so/conf/soc/sigma_final_pipeline.yaml
+      - /opt/so/conf/soc/sigma_so_pipeline.yaml
 
 socsigmaplaybookpipeline:
   file.managed:

salt/soc/defaults.yaml

@@ -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).
+              # The first document is the correlation (id, title, severity); the documents after it are the rules it refers to, all in this Detection.
+              #
+              # Supported types: event_count, value_count, temporal, value_sum, value_avg, value_percentile, value_median.
+              # 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  # matches the 'name' of the document 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%, and group-by or field values.
+              summary: '%count% distinct names queried by %source.ip% in %duration%'
+              level: 'medium'  # info | low | medium | high | critical
+              ---
+              title: 'Base Event'
+              # The correlation refers to this rule by 'name' (or by '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/enabled.sls

@@ -47,9 +47,8 @@ so-soc:
 {% endif %}
       - /opt/so/conf/soc/motd.md:/opt/sensoroni/html/motd.md:ro
       - /opt/so/conf/soc/banner.md:/opt/sensoroni/html/login/banner.md:ro
-      - /opt/so/conf/soc/sigma_so_pipeline.yaml:/opt/sensoroni/sigma_so_pipeline.yaml:ro
+      - /opt/so/conf/soc/sigma_pipelines:/opt/sensoroni/sigma_pipelines:ro
       - /opt/so/conf/soc/sigma_playbook_pipeline.yaml:/opt/sensoroni/sigma_playbook_pipeline.yaml:ro
-      - /opt/so/conf/soc/sigma_final_pipeline.yaml:/opt/sensoroni/sigma_final_pipeline.yaml:ro
       - /opt/so/conf/soc/playbook_placeholder_map.yaml:/opt/sensoroni/playbook_placeholder_map.yaml:ro
       - /opt/so/conf/soc/playbook_placeholder_map_custom.yaml:/opt/sensoroni/playbook_placeholder_map_custom.yaml:ro
       - /opt/so/conf/soc/custom.js:/opt/sensoroni/html/js/custom.js:ro
@@ -107,6 +106,7 @@ so-soc:
       - file: socclientsroles
       - file: socplaybookplaceholdermap
       - file: socplaybookplaceholdermapcustom
+      - file: socsigmapipelines
 
 delete_so-soc_so-status.disabled:
   file.uncomment:

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

@@ -0,0 +1,477 @@
+name: Security Onion ES|QL Pipeline
+# ES|QL query settings
+priority: 92
+transformations:
+    - id: esql_default_index
+      type: set_state
+      key: index
+      val: .ds-logs-*
+    - id: esql_source_metadata
+      type: set_state
+      key: metadata
+      val: "_id, _index, _source"
+    - id: esql_source_keep
+      type: set_state
+      key: keep
+      val: "_id, _index, _source"
+    # unmapped fields read as null instead of failing the query
+    - id: esql_unmapped_fields
+      type: set_state
+      key: unmapped_fields
+      val: nullify
+    # FROM targets per logsource, any namespace; later entries win, correlations get the union
+    - id: esql_index_process_creation
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.process-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-system.security-*
+        - .ds-logs-windows.powershell-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-auditd_manager.auditd-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        category: process_creation
+    - id: esql_index_process_creation_windows
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.process-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-system.security-*
+        - .ds-logs-windows.powershell-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: windows
+        category: process_creation
+    - id: esql_index_process_creation_linux
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.process-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-auditd_manager.auditd-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: linux
+        category: process_creation
+    - id: esql_index_process_creation_macos
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.process-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: macos
+        category: process_creation
+    - id: esql_index_file
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.file-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: file_event
+      - type: logsource
+        category: file_delete
+      - type: logsource
+        category: file_rename
+      - type: logsource
+        category: file_change
+      - type: logsource
+        category: file_access
+    - id: esql_index_file_windows
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.file-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: windows
+        category: file_event
+      - type: logsource
+        product: windows
+        category: file_delete
+      - type: logsource
+        product: windows
+        category: file_rename
+      - type: logsource
+        product: windows
+        category: file_change
+      - type: logsource
+        product: windows
+        category: file_access
+    - id: esql_index_file_linux
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.file-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: linux
+        category: file_event
+      - type: logsource
+        product: linux
+        category: file_delete
+      - type: logsource
+        product: linux
+        category: file_rename
+      - type: logsource
+        product: linux
+        category: file_change
+      - type: logsource
+        product: linux
+        category: file_access
+    - id: esql_index_file_macos
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.file-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: macos
+        category: file_event
+      - type: logsource
+        product: macos
+        category: file_delete
+      - type: logsource
+        product: macos
+        category: file_rename
+      - type: logsource
+        product: macos
+        category: file_change
+      - type: logsource
+        product: macos
+        category: file_access
+    - id: esql_index_registry
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.registry-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: registry_set
+      - type: logsource
+        category: registry_add
+      - type: logsource
+        category: registry_delete
+      - type: logsource
+        category: registry_event
+    - id: esql_index_library
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.library-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: image_load
+      - type: logsource
+        category: driver_load
+    - id: esql_index_endpoint_network
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.network-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: network_connection
+      - type: logsource
+        category: dns_query
+    - id: esql_index_endpoint_network_windows
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.network-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: windows
+        category: network_connection
+      - type: logsource
+        product: windows
+        category: dns_query
+    - id: esql_index_endpoint_network_linux
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.network-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: linux
+        category: network_connection
+      - type: logsource
+        product: linux
+        category: dns_query
+    - id: esql_index_endpoint_network_macos
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-endpoint.events.network-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: macos
+        category: network_connection
+      - type: logsource
+        product: macos
+        category: dns_query
+    - id: esql_index_sysmon_only
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: process_access
+      - type: logsource
+        category: create_remote_thread
+      - type: logsource
+        category: pipe_created
+      - type: logsource
+        category: create_stream_hash
+      - type: logsource
+        category: wmi_event
+      - type: logsource
+        category: raw_access_thread
+      - type: logsource
+        category: process_tampering
+      - type: logsource
+        category: sysmon_status
+      - type: logsource
+        category: sysmon_error
+      - type: logsource
+        category: file_executable_detected
+      - type: logsource
+        category: file_block_executable
+      - type: logsource
+        category: file_block_shredding
+      - type: logsource
+        category: clipboard_capture
+      - type: logsource
+        product: windows
+        service: sysmon
+    - id: esql_index_ps_operational
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-windows.powershell_operational-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: ps_script
+      - type: logsource
+        category: ps_module
+      - type: logsource
+        product: windows
+        service: powershell
+    - id: esql_index_ps_classic
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-windows.powershell-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: ps_classic_start
+      - type: logsource
+        category: ps_classic_provider_start
+      - type: logsource
+        category: ps_classic_script
+      - type: logsource
+        product: windows
+        service: powershell-classic
+    - id: esql_index_win_security
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-system.security-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: windows
+        service: security
+    - id: esql_index_win_system
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-system.system-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: windows
+        service: system
+    - id: esql_index_win_application
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-system.application-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: windows
+        service: application
+    - id: esql_index_linux_auth
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-system.auth-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        product: linux
+        service: auth
+      - type: logsource
+        product: linux
+        service: sshd
+      - type: logsource
+        product: linux
+        service: sudo
+    - id: esql_index_linux_syslog
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-system.syslog-*
+        - .ds-logs-syslog-so-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: linux
+        service: syslog
+    - id: esql_index_linux_auditd
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-auditd_manager.auditd-*
+        - .ds-logs-auditd.log-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: linux
+        service: auditd
+    - id: esql_index_network
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-zeek-so-*
+        - .ds-logs-suricata-so-*
+        - .ds-logs-suricata.alerts-so-*
+        - .ds-logs-endpoint.events.network-*
+        - .ds-logs-windows.sysmon_operational-*
+        - .ds-logs-sysmon_linux.log-*
+        - .ds-logs-windows.forwarded-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        category: network
+    - id: esql_index_so_network
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-zeek-so-*
+        - .ds-logs-suricata-so-*
+        - .ds-logs-import-so-*
+      rule_cond_op: or
+      rule_conditions:
+      - type: logsource
+        category: network
+        service: connection
+      - type: logsource
+        category: network
+        service: dns
+      - type: logsource
+        category: network
+        service: http
+      - type: logsource
+        category: network
+        service: file
+      - type: logsource
+        category: network
+        service: x509
+      - type: logsource
+        category: network
+        service: ssl
+      - type: logsource
+        category: network
+        service: ssh
+      - type: logsource
+        category: dns
+    - id: esql_index_zeek
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-zeek-so-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: zeek
+    - id: esql_index_opencanary
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-idh-so-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: opencanary
+    - id: esql_index_kratos
+      type: set_state
+      key: index
+      val:
+        - .ds-logs-kratos-so-*
+        - .ds-logs-import-so-*
+      rule_conditions:
+      - type: logsource
+        product: kratos
+

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

@@ -14,18 +14,15 @@ transformations:
         - process.args
         - related.ip
         - dns.resolved_ip
-    - id: esql_default_index
+    # Always lowercase, so matched exactly; the backend can then use the indexed ':' operator.
+    - id: case_sensitive_categorization_fields
       type: set_state
-      key: index
-      val: .ds-logs-*
-    - id: esql_source_metadata
-      type: set_state
-      key: metadata
-      val: "_id, _index, _source"
-    - id: esql_source_keep
-      type: set_state
-      key: keep
-      val: "_id, _index, _source"
+      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
@@ -130,6 +127,9 @@ transformations:
       valid_hash_algos: ["MD5", "SHA1", "SHA256", "SHA512", "IMPHASH"]
       field_prefix: "file"
       drop_algo_prefix: False
+      # ecs_windows has already renamed Hashes; pySigma 1.5+ only parses the fields listed here
+      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,18 @@
 {%   do SOCMERGED.config.server.update({'airgapEnabled': false}) %}
 {% endif %}
 
+{# Sigma correlations require ES|QL: load the community correlations and offer correlation authoring only when it is on #}
+{% set use_esql = SOCMERGED.config.server.modules.elastalertengine.useEsql %}
+{% for repo in SOCMERGED.config.server.modules.elastalertengine.rulesRepos %}
+{%   if repo.get('rulesetName') == 'securityonion-resources' and repo.get('folder') in ['sigma', 'sigma/stable'] %}
+{%     do repo.update({'folder': 'sigma' if use_esql else 'sigma/stable'}) %}
+{%   endif %}
+{% endfor %}
+{% if not use_esql %}
+{%   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,7 +401,7 @@ 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
@@ -410,6 +410,18 @@ soc:
             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. 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.
@@ -862,15 +874,14 @@ soc:
             global: True
             forcedType: bool
           automations:
-            template:
-              description: Scheduled automations for the Onion AI assistant, managed from the Agent Studio. Each automation is stored under its own generated ID so its history and rollback are independent of every other automation.
-              global: True
-              advanced: False
-              readonlyUi: True
-              duplicates: True
-              forcedType: string
-              syntax: json
-              helpLink: onion-ai
+            description: Scheduled automations for the Onion AI assistant, managed from the Agent Studio.
+            global: True
+            advanced: True
+            readonlyUi: True
+            storage: db
+            forcedType: string
+            syntax: json
+            helpLink: onion-ai
           agents:
             description: Agent definitions for the Onion AI assistant, managed from the Agent Studio. An entry naming a system agent overrides only the fields an admin may change; everything else comes from the built-in definition.
             global: True

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" | \