Initial Correlations support

FIX: use /sbin/nologin for service accounts

salt/elastalert/config.sls

@@ -21,6 +21,7 @@ elastalert:
     - gid: 933
     - home: /opt/so/conf/elastalert
     - createhome: False
+    - shell: /sbin/nologin
 
 elastalogdir:
   file.directory:

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,13 @@
 # Elastic License 2.0.
 
 
+from datetime import datetime
 from time import gmtime, strftime
+import hashlib
+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,7 +23,99 @@ 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]+)%')
+
+    @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 [])
+
+    def alert_id(self, match):
+        """ Stable id: window + group values for correlations, source _id otherwise. """
+        keys = self.query_keys()
+        if keys:
+            values = '|'.join(str(self.lookup(match, k)) for k in keys)
+            key = f"{self.rule['detection_public_id']}|{self.to_dt(match['@timestamp']).isoformat()}|{values}"
+        else:
+            key = f"{self.rule['detection_public_id']}|{match.get('_id')}"
+
+        return hashlib.sha256(key.encode('utf-8')).hexdigest()
+
+    @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 'window_start' not in 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:
@@ -42,6 +138,10 @@ class SecurityOnionESAlerter(Alerter):
                 if field in self.rule:
                     rule_info[rule_key] = self.rule[field]
 
+            summary = self.summary(match)
+            if summary:
+                rule_info["summary"] = summary
+
             # Construct the payload with the conditional rule_info
             payload = {
                 "tags": "alert",
@@ -56,8 +156,21 @@ class SecurityOnionESAlerter(Alerter):
                 "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)
+            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}")
+            try:
+                response = requests.put(url, data=json.dumps(payload, cls=DateTimeEncoder), headers=headers, verify=False, auth=creds)
+            except requests.RequestException as e:
+                raise EAException(f"Unable to write alert: {e}")
+            if response.status_code == 400:
+                # mapping rejections fail the same way on retry, so drop them
+                elastalert_logger.error("Dropping alert %s for rule %s, rejected by Elasticsearch: %s",
+                                        alert_id, self.rule['detection_public_id'], response.text[:500])
+                continue
+            if response.status_code != 409 and not response.ok:
+                raise EAException(f"Unable to write alert: {response.status_code} {response.text[:500]}")
 
     def get_info(self):
         return {'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/elastic-fleet-package-registry/config.sls

@@ -19,6 +19,7 @@ elastic-agent-pr:
     - gid: 948
     - home: /opt/so/conf/elastic-fleet-pr
     - createhome: False
+    - shell: /sbin/nologin
 
 {% else %}
 

salt/elasticagent/config.sls

@@ -20,6 +20,7 @@ elastic-agent:
     - gid: 949
     - home: /opt/so/conf/elastic-agent
     - createhome: False
+    - shell: /sbin/nologin
 
 elasticagentconfdir:
   file.directory:

salt/elasticfleet/config.sls

@@ -26,6 +26,7 @@ elastic-fleet:
     - gid: 947
     - home: /opt/so/conf/elastic-fleet
     - createhome: False
+    - shell: /sbin/nologin
 
 elasticfleet_sbin:
   file.recurse:

salt/elasticfleet/tools/sbin/so-elastic-fleet-common

@@ -30,6 +30,56 @@ fleet_api() {
     curl -sK /opt/so/conf/elasticsearch/curl.config -L "localhost:5601/api/fleet/${QUERYPATH}" "$@" --retry 3 --retry-delay 10 --fail 2>/dev/null
 }
 
+elastic_fleet_require_agent_policy() {
+    local AGENT_POLICY=$1
+    local POLICY_JSON
+
+    if ! POLICY_JSON=$(fleet_api "agent_policies/$AGENT_POLICY") || [ -z "$POLICY_JSON" ]; then
+        echo "Error: Agent policy '$AGENT_POLICY' was not found or is not visible in the current Kibana space." >&2
+        return 1
+    fi
+
+    if ! jq -e '.item.package_policies | type == "array"' <<<"$POLICY_JSON" >/dev/null 2>&1; then
+        echo "Error: Agent policy '$AGENT_POLICY' was not found or is not visible in the current Kibana space." >&2
+        return 1
+    fi
+
+    echo "$POLICY_JSON"
+}
+
+# Print the single active enrollment token for POLICY_ID.
+# Exit 1: retryable (API failure, invalid response, no active token)
+# Exit 2: multiple active tokens - Shouldn't get into this state without manual intervention
+elastic_fleet_active_enrollment_token() {
+    local POLICY_ID=$1
+    local RESP TOKEN_COUNT API_KEY
+
+    if ! RESP=$(fleet_api "enrollment_api_keys?perPage=100" -H 'kbn-xsrf: true' -H 'Content-Type: application/json'); then
+        echo "Error: Failed to retrieve enrollment tokens for agent policy '$POLICY_ID'." >&2
+        return 1
+    fi
+
+    if ! jq -e '.list' <<<"$RESP" >/dev/null 2>&1; then
+        echo "Error: Invalid enrollment token response for agent policy '$POLICY_ID'." >&2
+        return 1
+    fi
+
+    TOKEN_COUNT=$(jq --arg pid "$POLICY_ID" '[.list[] | select(.policy_id == $pid and .active == true)] | length' <<<"$RESP")
+
+    if [ "${TOKEN_COUNT:-0}" -eq 0 ]; then
+        echo "Error: No active enrollment token found for agent policy '$POLICY_ID'." >&2
+        return 1
+    fi
+
+    if [ "$TOKEN_COUNT" -gt 1 ]; then
+        echo "Error: Found $TOKEN_COUNT active enrollment tokens for agent policy '$POLICY_ID'; expected exactly one." >&2
+        return 2
+    fi
+
+    API_KEY=$(jq -r --arg pid "$POLICY_ID" '.list[] | select(.policy_id == $pid and .active == true) | .api_key' <<<"$RESP")
+    echo "$API_KEY"
+}
+
 # Max number of concurrent Fleet write jobs (create/update). Override via env if needed.
 MAX_FLEET_JOBS=${MAX_FLEET_JOBS:-10}
 
@@ -62,15 +112,7 @@ elastic_fleet_load_integrations_dir() {
     i=0
 
     # Fetch the agent policy a single time; we look up integration ids locally below.
-    if ! POLICY_JSON=$(fleet_api "agent_policies/$AGENT_POLICY"); then
-        echo "Error: Failed to retrieve agent policy '$AGENT_POLICY'."
-        rm -f "$FAIL_FILE"
-        rm -rf "$OUT_DIR"
-        return 1
-    fi
-
-    if ! jq -e '.item.package_policies' <<<"$POLICY_JSON" >/dev/null 2>&1; then
-        echo "Error: Invalid agent policy response for '$AGENT_POLICY'."
+    if ! POLICY_JSON=$(elastic_fleet_require_agent_policy "$AGENT_POLICY"); then
         rm -f "$FAIL_FILE"
         rm -rf "$OUT_DIR"
         return 1
@@ -124,9 +166,15 @@ elastic_fleet_integration_check() {
 
     JSON_STRING=$2
 
-    NAME=$(jq -r .name $JSON_STRING)
+    NAME=$(jq -r .name "$JSON_STRING")
+    INTEGRATION_ID=""
 
-    INTEGRATION_ID=$(/usr/sbin/so-elastic-fleet-agent-policy-view "$AGENT_POLICY" | jq -r '.item.package_policies[] | select(.name=="'"$NAME"'") | .id')
+    local POLICY_JSON
+    if ! POLICY_JSON=$(elastic_fleet_require_agent_policy "$AGENT_POLICY"); then
+        return 1
+    fi
+
+    INTEGRATION_ID=$(jq -r --arg n "$NAME" '.item.package_policies[]? | select(.name==$n) | .id' <<<"$POLICY_JSON")
 
 }
 
@@ -148,7 +196,16 @@ elastic_fleet_integration_remove() {
 
     NAME=$2
 
-    INTEGRATION_ID=$(/usr/sbin/so-elastic-fleet-agent-policy-view "$AGENT_POLICY" | jq -r '.item.package_policies[] | select(.name=="'"$NAME"'") | .id')
+    local POLICY_JSON
+    if ! POLICY_JSON=$(elastic_fleet_require_agent_policy "$AGENT_POLICY"); then
+        return 1
+    fi
+
+    INTEGRATION_ID=$(jq -r --arg n "$NAME" '.item.package_policies[]? | select(.name==$n) | .id' <<<"$POLICY_JSON")
+    if [ -z "$INTEGRATION_ID" ]; then
+        echo "Error: Integration '$NAME' was not found in agent policy '$AGENT_POLICY'." >&2
+        return 1
+    fi
 
     JSON_STRING=$( jq -n \
                     --arg INTEGRATIONID "$INTEGRATION_ID" \

salt/elasticfleet/tools/sbin/so-elastic-fleet-integration-policy-elastic-defend

@@ -13,7 +13,10 @@ ERROR=false
 for INTEGRATION in /opt/so/conf/elastic-fleet/integrations/elastic-defend/*.json
 do
   printf "\n\nInitial Endpoints Policy - Loading $INTEGRATION\n"
-  elastic_fleet_integration_check "endpoints-initial" "$INTEGRATION"
+  if ! elastic_fleet_integration_check "endpoints-initial" "$INTEGRATION"; then
+    ERROR=true
+    continue
+  fi
   if [ -n "$INTEGRATION_ID" ]; then
       printf "\n\nIntegration $NAME exists - Upgrading integration policy\n"
       if ! elastic_fleet_integration_policy_upgrade "$INTEGRATION_ID"; then

salt/elasticfleet/tools/sbin/so-elastic-fleet-integration-policy-elastic-fleet-server

@@ -7,20 +7,35 @@
 . /usr/sbin/so-elastic-fleet-common
 
 # Get all the fleet policies
-json_output=$(curl -s -K /opt/so/conf/elasticsearch/curl.config -L -X GET "localhost:5601/api/fleet/agent_policies" -H 'kbn-xsrf: true')
+if ! json_output=$(fleet_api "agent_policies" -H 'kbn-xsrf: true'); then
+    echo "Error: Failed to retrieve Fleet agent policies." >&2
+    exit 1
+fi
+
+if ! jq -e '.items' <<<"$json_output" >/dev/null 2>&1; then
+    echo "Error: Invalid Fleet agent policies response." >&2
+    exit 1
+fi
 
 # Extract the IDs that start with "FleetServer_"
-POLICY=$(echo "$json_output" | jq -r '.items[] | select(.id | startswith("FleetServer_")) | .id')
+POLICY=$(jq -r '.items[] | select(.id | startswith("FleetServer_")) | .id' <<<"$json_output")
 
 # Iterate over each ID in the POLICY variable
 for POLICYNAME in $POLICY; do
     printf "\nUpdating Policy: $POLICYNAME\n"
 
-    # First get the Integration ID
-    INTEGRATION_ID=$(/usr/sbin/so-elastic-fleet-agent-policy-view "$POLICYNAME" | jq -r '.item.package_policies[] | select(.package.name == "fleet_server") | .id')  
+    if ! POLICY_JSON=$(elastic_fleet_require_agent_policy "$POLICYNAME"); then
+        exit 1
+    fi
+
+    INTEGRATION_ID=$(jq -r '.item.package_policies[]? | select(.package.name == "fleet_server") | .id' <<<"$POLICY_JSON")
+    if [ -z "$INTEGRATION_ID" ]; then
+        echo "Error: fleet_server integration was not found in agent policy '$POLICYNAME'." >&2
+        exit 1
+    fi
 
     # Modify the default integration policy to update the policy_id and an with the correct naming
-	UPDATED_INTEGRATION_POLICY=$(jq --arg policy_id "$POLICYNAME" --arg name "fleet_server-$POLICYNAME" '
+    UPDATED_INTEGRATION_POLICY=$(jq --arg policy_id "$POLICYNAME" --arg name "fleet_server-$POLICYNAME" '
     .policy_id = $policy_id |
     .name = $name' /opt/so/conf/elastic-fleet/integrations/fleet-server/fleet-server.json)
 

salt/elasticfleet/tools/sbin_jinja/so-elastic-agent-gen-installers

@@ -22,12 +22,19 @@ NUM_RUNNING=$(pgrep -cf "/bin/bash /sbin/so-elastic-agent-gen-installers")
 
 for i in {1..30}
 do
-   ENROLLMENTOKEN=$(curl -K /opt/so/conf/elasticsearch/curl.config -L "localhost:5601/api/fleet/enrollment_api_keys?perPage=100" -H 'kbn-xsrf: true' -H 'Content-Type: application/json'  | jq .list | jq -r -c '.[] | select(.policy_id | contains("endpoints-initial")) | .api_key')
+   ENROLLMENTOKEN=$(elastic_fleet_active_enrollment_token "endpoints-initial")
+   TOKEN_RC=$?
+   if [ "$TOKEN_RC" -eq 2 ]; then
+     exit 1
+   fi
    FLEETHOST=$(curl -K /opt/so/conf/elasticsearch/curl.config 'http://localhost:5601/api/fleet/fleet_server_hosts/grid-default' | jq -r  '.item.host_urls[]' | paste -sd ',')
-if [[ $FLEETHOST ]] && [[  $ENROLLMENTOKEN ]]; then break; else sleep 10; fi
+   if [[ -n "$FLEETHOST" ]] && [[ -n "$ENROLLMENTOKEN" ]]; then
+     break
+   fi
+   sleep 10
 done
 
-if [[ -z $FLEETHOST ]] || [[ -z $ENROLLMENTOKEN ]]; then 
+if [[ -z "$FLEETHOST" ]] || [[ -z "$ENROLLMENTOKEN" ]]; then
   printf "\nFleet Host URL, Enrollment Token or Elastic Version empty - exiting..."
   printf "\nFleet Host: $FLEETHOST, Enrollment Token: $ENROLLMENTOKEN\n"
   exit 1
@@ -67,19 +74,25 @@ for GOOS in "${GOTARGETOS[@]}"; do
     GOARCH="amd64"
     if [[ $GOOS == 'darwin/arm64' ]]; then GOOS="darwin" && GOARCH="arm64"; fi
     printf "\n\n### Generating $GOOS/$GOARCH Installer...\n"
-    docker run -e CGO_ENABLED=0 -e GOOS=$GOOS -e GOARCH=$GOARCH \
+    if ! docker run -e CGO_ENABLED=0 -e GOOS=$GOOS -e GOARCH=$GOARCH \
     --mount type=bind,source=/etc/pki/tls/certs/,target=/workspace/files/cert/ \
     --mount type=bind,source=/nsm/elastic-agent-workspace/,target=/workspace/files/elastic-agent/ \
     --mount type=bind,source=/opt/so/saltstack/local/salt/elasticfleet/files/,target=/output/ \
-    {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent-builder:{{ GLOBALS.so_version }} go build -ldflags "-X main.fleetHostURLsList=$FLEETHOST -X main.enrollmentToken=$ENROLLMENTOKEN"  -o /output/so-elastic-agent_${GOOS}_${GOARCH}
+    {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent-builder:{{ GLOBALS.so_version }} go build -ldflags "-X main.fleetHostURLsList=$FLEETHOST -X main.enrollmentToken=$ENROLLMENTOKEN"  -o /output/so-elastic-agent_${GOOS}_${GOARCH}; then
+        printf "\n### ERROR: Failed to generate $GOOS/$GOARCH installer. Exiting...\n"
+        exit 1
+    fi
     printf "\n### $GOOS/$GOARCH Installer Generated...\n"
 done
 
 printf "\n\n### Generating MSI...\n"
 cp /opt/so/saltstack/local/salt/elasticfleet/files/so-elastic-agent_windows_amd64 /opt/so/saltstack/local/salt/elasticfleet/files/so-elastic-agent_windows_amd64.exe
-docker run \
+if ! docker run \
 --mount type=bind,source=/opt/so/saltstack/local/salt/elasticfleet/files/,target=/output/ -w /output \
-{{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent-builder:{{ GLOBALS.so_version }} wixl -o so-elastic-agent_windows_amd64_msi --arch x64 /workspace/so-elastic-agent.wxs
+{{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent-builder:{{ GLOBALS.so_version }} wixl -o so-elastic-agent_windows_amd64_msi --arch x64 /workspace/so-elastic-agent.wxs; then
+    printf "\n### ERROR: Failed to generate MSI. Exiting...\n"
+    exit 1
+fi
 printf "\n### MSI Generated...\n"
 
 # Verify installers were created

salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup

@@ -202,26 +202,9 @@ fi
 ### Finalization ###
 
 # Query for Enrollment Tokens for default policies
-if ENDPOINTSENROLLMENTOKEN_RAW=$(fleet_api "enrollment_api_keys" -H 'kbn-xsrf: true' -H 'Content-Type: application/json'); then
-    ENDPOINTSENROLLMENTOKEN=$(echo "$ENDPOINTSENROLLMENTOKEN_RAW" | jq .list | jq -r -c '.[] | select(.policy_id | contains("endpoints-initial")) | .api_key')
-else
-    echo -e "\nFailed to query for Endpoints enrollment token"
-    exit 1
-fi
-
-if GRIDNODESENROLLMENTOKENGENERAL_RAW=$(fleet_api "enrollment_api_keys" -H 'kbn-xsrf: true' -H 'Content-Type: application/json'); then
-    GRIDNODESENROLLMENTOKENGENERAL=$(echo "$GRIDNODESENROLLMENTOKENGENERAL_RAW" | jq .list | jq -r -c '.[] | select(.policy_id | contains("so-grid-nodes_general")) | .api_key')
-else
-    echo -e "\nFailed to query for Grid nodes - General enrollment token"
-    exit 1
-fi
-
-if GRIDNODESENROLLMENTOKENHEAVY_RAW=$(fleet_api "enrollment_api_keys" -H 'kbn-xsrf: true' -H 'Content-Type: application/json'); then
-    GRIDNODESENROLLMENTOKENHEAVY=$(echo "$GRIDNODESENROLLMENTOKENHEAVY_RAW" | jq .list | jq -r -c '.[] | select(.policy_id | contains("so-grid-nodes_heavy")) | .api_key')
-else
-    echo -e "\nFailed to query for Grid nodes - Heavy enrollment token"
-    exit 1
-fi
+ENDPOINTSENROLLMENTOKEN=$(elastic_fleet_active_enrollment_token "endpoints-initial") || exit 1
+GRIDNODESENROLLMENTOKENGENERAL=$(elastic_fleet_active_enrollment_token "so-grid-nodes_general") || exit 1
+GRIDNODESENROLLMENTOKENHEAVY=$(elastic_fleet_active_enrollment_token "so-grid-nodes_heavy") || exit 1
 
 # Store needed data in minion pillar
 pillar_file=/opt/so/saltstack/local/pillar/minions/{{ GLOBALS.minion_id }}.sls

salt/elasticsearch/config.sls

@@ -32,6 +32,7 @@ elasticsearch:
     - gid: 930
     - home: /opt/so/conf/elasticsearch
     - createhome: False
+    - shell: /sbin/nologin
 
 elasticsearch_sbin:
   file.recurse:

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/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,106 @@
+{
+  "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"
+            },
+            "summary": {
+              "type": "match_only_text",
+              "fields": {
+                "keyword": {
+                  "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"
+            }
+          }
+        },
+        "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"
+            }
+          }
+        }
+      }
+    }
+  },
+  "_meta": {
+    "description": "Fields written by the ElastAlert SecurityOnionESAlerter to logs-detections.alerts-so"
+  }
+}

salt/kafka/config.sls

@@ -21,6 +21,7 @@ kafka_user:
     - gid: 960
     - home: /opt/so/conf/kafka
     - createhome: False
+    - shell: /sbin/nologin
 
 kafka_home_dir:
   file.absent:

salt/kibana/config.sls

@@ -22,6 +22,7 @@ kibana:
     - gid: 932
     - home: /opt/so/conf/kibana
     - createhome: False
+    - shell: /sbin/nologin
 
 # Drop the correct nginx config based on role
 

salt/kratos/config.sls

@@ -27,6 +27,7 @@ kratos:
     - uid: 928
     - gid: 928
     - home: /opt/so/conf/kratos
+    - shell: /sbin/nologin
 
 kratosdir:
   file.directory:

salt/logstash/config.sls

@@ -35,6 +35,7 @@ logstash:
     - uid: 931
     - gid: 931
     - home: /opt/so/conf/logstash
+    - shell: /sbin/nologin
 
 logstash_sbin:
   file.recurse:

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
+
   INSTALLEDVERSION=3.4.0
 }
 

salt/sensoroni/files/analyzers/whoislookup/requirements.txt

@@ -1,2 +1,3 @@
-requests>=2.31.0
-whoisit>=2.7.0
+requests>=2.34.0
+whoisit>=4.0.5
+anyio>=4.15.1

salt/soc/defaults.yaml

@@ -1467,6 +1467,8 @@ soc:
             - emerging_threats_addon
           useEsql: false
           esqlCaseInsensitive: true
+          esqlQueryDelaySeconds: 30
+          esqlCorrelationAllowanceSeconds: 600
         elastic:
           hostUrl:
           remoteHostUrls: []
@@ -1494,6 +1496,9 @@ soc:
           org: Security Onion
           bucket: telegraf/so_short_term
           verifyCert: false
+        notification:
+          dismissedPruneDays: 30
+          enabled: false
         playbook:
           autoUpdateEnabled: true
           playbookImportFrequencySeconds: 86400
@@ -2668,8 +2673,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
@@ -2753,7 +2761,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
@@ -2783,6 +2791,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/files/soc/sigma_so_pipeline.yaml

@@ -14,6 +14,15 @@ transformations:
         - process.args
         - related.ip
         - dns.resolved_ip
+    # Always lowercase, so matched exactly; the backend can then use the indexed ':' operator.
+    - id: case_sensitive_categorization_fields
+      type: set_state
+      key: case_insensitive_exempt_fields
+      val:
+        - tags
+        - event.category
+        - event.type
+        - event.kind
     - id: esql_default_index
       type: set_state
       key: index

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}) %}

salt/soc/soc_soc.yaml

@@ -409,6 +409,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.
@@ -498,6 +510,15 @@ soc:
             forcedType: string
             syntax: json
             storage: db
+          dismissedPruneDays:
+            title: Dismissed Retention Days
+            description: The number of days to retain dismissed notifications. When a notification is dismissed, it will be pruned after this many days. Only one user need dismiss a notification for it to be pruned.
+            forcedType: int
+            global: True
+          enabled:
+            description: Enables or disables the SOC notification module.
+            forcedType: bool
+            global: True
         postgres:
           host:
             description: Hostname or IP address of the PostgreSQL server used by SOC. Defaults to the manager hostname.
@@ -782,6 +803,7 @@ soc:
                 - gemini
                 - openai_responses
                 - openai_chat
+                - openai_embeddings
             - field: apiUrl
               label: API URL
               required: False

salt/suricata/config.sls

@@ -64,6 +64,7 @@ suricata:
     - gid: 940
     - home: /nsm/suricata
     - createhome: False
+    - shell: /sbin/nologin
 
 socoregroupwithsuricata:
   group.present:

salt/zeek/config.sls

@@ -23,6 +23,7 @@ zeek:
     - gid: 937
     - home: /opt/so/conf/zeek
     - createhome: False
+    - shell: /sbin/nologin
 
 # Create some directories
 zeekpolicydir: