add logrotate for virtual_node_manager and so-salt-cloud

Revert "use -verify flag during grid agent install to ensure agent health"

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

@@ -165,6 +165,8 @@ if [[ $EXCLUDE_FALSE_POSITIVE_ERRORS == 'Y' ]]; then
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|upgrading component template"  # false positive (elasticsearch index or template names contain 'error')
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|upgrading composable template" # false positive (elasticsearch composable template names contain 'error')
     EXCLUDED_ERRORS="$EXCLUDED_ERRORS|Error while parsing document for index \[.ds-logs-kratos-so-.*object mapping for \[file\]" # false positive (mapping error occuring BEFORE kratos index has rolled over in 2.4.210)
+    EXCLUDED_ERRORS="$EXCLUDED_ERRORS|No such container"            # false positive (telegraf trying to run stats on an old container)
+    EXCLUDED_ERRORS="$EXCLUDED_ERRORS|passwords do not match"       # false positive (automated hydra test)
 fi
 
 if [[ $EXCLUDE_KNOWN_ERRORS == 'Y' ]]; then

salt/elasticsearch/defaults.yaml

@@ -3958,10 +3958,13 @@ elasticsearch:
         - vulnerability-mappings
         - common-settings
         - common-dynamic-mappings
+        - logs-redis.log@package
+        - logs-redis.log@custom
         data_stream:
           allow_custom_routing: false
           hidden: false
-        ignore_missing_component_templates: []
+        ignore_missing_component_templates:
+        - logs-redis.log@custom
         index_patterns:
         - logs-redis.log*
         priority: 501

salt/elasticsearch/files/ingest/common

@@ -63,7 +63,8 @@
     { "set":             { "if": "ctx.event?.dataset != null && !ctx.event.dataset.contains('.')", "field": "event.dataset", "value": "{{event.module}}.{{event.dataset}}" } },
     { "split":           { "if": "ctx.event?.dataset != null && ctx.event.dataset.contains('.')", "field": "event.dataset", "separator": "\\.", "target_field": "dataset_tag_temp" } },
     { "append":          { "if": "ctx.dataset_tag_temp != null", "field": "tags", "value": "{{dataset_tag_temp.1}}"  } },
-    { "grok":            { "if": "ctx.http?.response?.status_code != null", "field": "http.response.status_code", "patterns": ["%{NUMBER:http.response.status_code:long} %{GREEDYDATA}"]} },
+    { "grok":            { "if": "ctx.http?.response?.status_code instanceof String", "field": "http.response.status_code", "patterns": ["%{NUMBER:http.response.status_code:long}(?:\\s+%{GREEDYDATA})?"], "ignore_failure": true } },
+    { "convert":         { "if": "ctx.http?.response?.status_code != null && !(ctx.http.response.status_code instanceof Number)", "field": "http.response.status_code", "type": "long", "ignore_failure": true } },
     { "set":             { "if": "ctx?.metadata?.kafka != null" , "field": "kafka.id", "value": "{{metadata.kafka.partition}}{{metadata.kafka.offset}}{{metadata.kafka.timestamp}}", "ignore_failure": true } },
     { "remove":          { "field": [ "message2", "type", "fields", "category", "module", "dataset", "dataset_tag_temp", "event.dataset_temp" ], "ignore_missing": true, "ignore_failure": true } },
     { "pipeline": { "name": "global@custom", "ignore_missing_pipeline": true, "description": "[Fleet] Global pipeline for all data streams" } }

salt/elasticsearch/files/ingest/global@custom

@@ -177,12 +177,84 @@
                 "description": "Extract IPs from Elastic Agent events (host.ip) and adds them to related.ip"
             }
         },
+        {
+            "script": {
+                "description": "Snapshot event.ingested into _tmp.event_ingested_pre_fleet before .fleet_final_pipeline-1 overwrites it with ES ingest time",
+                "lang": "painless",
+                "if": "ctx.event?.ingested != null && ctx.event?.created == null",
+                "ignore_failure": true,
+                "source": "ctx.putIfAbsent('_tmp', [:]); ctx._tmp.event_ingested_pre_fleet = ctx.event.ingested;"
+            }
+        },
         {
             "pipeline": {
                 "name": ".fleet_final_pipeline-1",
                 "ignore_missing_pipeline": true
             }
         },
+        {
+            "script": {
+                "description": "Calculate time from Elastic Agent to Logstash.",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_agent != null",
+                "ignore_failure": true,
+                "source": "ZonedDateTime start = ctx._tmp.event_ingested_pre_fleet != null ? ZonedDateTime.parse(ctx._tmp.event_ingested_pre_fleet) : ZonedDateTime.parse(ctx['@timestamp']); ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_elasticagent_to_logstash = ChronoUnit.SECONDS.between(start, ZonedDateTime.parse(ctx._tmp.logstash_from_agent));"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time from Logstash to Redis",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_agent != null && ctx._tmp?.logstash_to_redis != null",
+                "ignore_failure": true,
+                "source": "ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_logstash_to_redis = ChronoUnit.SECONDS.between(ZonedDateTime.parse(ctx._tmp.logstash_from_agent), ZonedDateTime.parse(ctx._tmp.logstash_to_redis));"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time message spends in redis queue (logstash delay in pulling event).",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_to_redis != null && ctx._tmp?.logstash_from_redis != null",
+                "ignore_failure": true,
+                "source": "ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_redis_to_logstash = ChronoUnit.SECONDS.between(ZonedDateTime.parse(ctx._tmp.logstash_to_redis), ZonedDateTime.parse(ctx._tmp.logstash_from_redis));"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time from Logstash to Elasticsearch (after read from Redis).",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_redis != null",
+                "ignore_failure": true,
+                "source": "ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_logstash_to_elasticsearch = ChronoUnit.SECONDS.between(ZonedDateTime.parse(ctx._tmp.logstash_from_redis), metadata().now);"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time from Elastic Agent to Kafka.",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_kafka != null && ctx._tmp?.logstash_from_agent == null",
+                "ignore_failure": true,
+                "source": "ZonedDateTime start = ctx._tmp.event_ingested_pre_fleet != null ? ZonedDateTime.parse(ctx._tmp.event_ingested_pre_fleet) : ZonedDateTime.parse(ctx['@timestamp']); ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_elasticagent_to_kafka = ChronoUnit.SECONDS.between(start, ZonedDateTime.parse(ctx._tmp.logstash_from_kafka));"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time message spends in Kafka queue (logstash delay in pulling event).",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_kafka != null && ctx.metadata?.kafka?.timestamp != null && ctx._tmp?.logstash_from_agent == null",
+                "ignore_failure": true,
+                "source": "ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_kafka_queue = ChronoUnit.SECONDS.between(ZonedDateTime.ofInstant(Instant.ofEpochMilli(Long.parseLong(ctx.metadata.kafka.timestamp.toString())), ZoneId.of('UTC')), ZonedDateTime.parse(ctx._tmp.logstash_from_kafka));"
+            }
+        },
+        {
+            "script": {
+                "description": "Calculate time from Logstash to Elasticsearch (after read from Kafka).",
+                "lang": "painless",
+                "if": "ctx._tmp?.logstash_from_kafka != null && ctx._tmp?.logstash_from_agent == null",
+                "ignore_failure": true,
+                "source": "ctx.event.putIfAbsent('ingestion', [:]); ctx.event.ingestion.latency_kafka_to_elasticsearch = ChronoUnit.SECONDS.between(ZonedDateTime.parse(ctx._tmp.logstash_from_kafka), metadata().now);"
+            }
+        },
         {
             "remove": {
                 "field": "event.agent_id_status",
@@ -202,11 +274,12 @@
                     "event.dataset_temp",
                     "dataset_tag_temp",
                     "module_temp",
-                    "datastream_dataset_temp"
+                    "datastream_dataset_temp",
+                    "_tmp"
                 ],
                 "ignore_missing": true,
                 "ignore_failure": true
             }
         }
     ]
-}
+}

salt/elasticsearch/files/ingest/zeek.ja4d

@@ -0,0 +1,71 @@
+{
+    "description": "zeek.ja4d",
+    "processors": [
+        {
+            "set": {
+                "field": "event.dataset",
+                "value": "ja4d"
+            }
+        },
+        {
+            "remove": {
+                "field": [
+                    "host"
+                ],
+                "ignore_failure": true
+            }
+        },
+        {
+            "json": {
+                "field": "message",
+                "target_field": "message2",
+                "ignore_failure": true
+            }
+        },
+        {
+            "rename": {
+                "field": "message2.ja4d",
+                "target_field": "hash.ja4d",
+                "ignore_missing": true,
+                "if": "ctx?.message2?.ja4d != null && ctx.message2.ja4d.length() > 0"
+            }
+        },
+        {
+            "rename": {
+                "field": "message2.client_mac",
+                "target_field": "host.mac",
+                "ignore_missing": true,
+                "if": "ctx?.message2?.client_mac != null && ctx.message2.client_mac.length() > 0"
+            }
+        },
+        {
+            "rename": {
+                "field": "message2.hostname",
+                "target_field": "host.hostname",
+                "ignore_missing": true,
+                "if": "ctx?.message2?.hostname != null && ctx.message2.hostname.length() > 0"
+            }
+        },
+        {
+            "rename": {
+                "field": "message2.requested_ip",
+                "target_field": "dhcp.requested_address",
+                "ignore_missing": true,
+                "if": "ctx?.message2?.requested_ip != null && ctx.message2.requested_ip.length() > 0"
+            }
+        },
+        {
+            "rename": {
+                "field": "message2.vendor_class_id",
+                "target_field": "zeek.ja4d.vendor_class_id",
+                "ignore_missing": true,
+                "if": "ctx?.message2?.vendor_class_id != null && ctx.message2.vendor_class_id.length() > 0"
+            }
+        },
+        {
+            "pipeline": {
+                "name": "zeek.common"
+            }
+        }
+    ]
+}

salt/logrotate/defaults.yaml

@@ -210,6 +210,26 @@ logrotate:
       - extension .log
       - dateext
       - dateyesterday
+    /opt/so/log/salt/virtual_node_manager:
+      - daily
+      - rotate 14
+      - missingok
+      - copytruncate
+      - compress
+      - create
+      - extension .log
+      - dateext
+      - dateyesterday
+    /opt/so/log/salt/so-salt-cloud:
+      - daily
+      - rotate 14
+      - missingok
+      - copytruncate
+      - compress
+      - create
+      - extension .log
+      - dateext
+      - dateyesterday
     /nsm/idh/*_x_log:
       - daily
       - rotate 14

salt/logrotate/soc_logrotate.yaml

@@ -133,6 +133,20 @@ logrotate:
       multiline: True
       global: True
       forcedType: "[]string"
+    "/opt/so/log/salt/virtual_node_manager":
+      description: List of logrotate options for this file.
+      title: /opt/so/log/salt/virtual_node_manager
+      advanced: True
+      multiline: True
+      global: True
+      forcedType: "[]string"
+    "/opt/so/log/salt/so-salt-cloud":
+      description: List of logrotate options for this file.
+      title: /opt/so/log/salt/so-salt-cloud
+      advanced: True
+      multiline: True
+      global: True
+      forcedType: "[]string"
     "/nsm/idh/*_x_log":
       description: List of logrotate options for this file.
       title: /nsm/idh/*.log

salt/logstash/defaults.yaml

@@ -26,12 +26,12 @@ logstash:
     manager:
       - so/0011_input_endgame.conf
       - so/0012_input_elastic_agent.conf.jinja
-      - so/0013_input_lumberjack_fleet.conf
+      - so/0013_input_lumberjack_fleet.conf.jinja
       - so/9999_output_redis.conf.jinja
     receiver:
       - so/0011_input_endgame.conf
       - so/0012_input_elastic_agent.conf.jinja
-      - so/0013_input_lumberjack_fleet.conf
+      - so/0013_input_lumberjack_fleet.conf.jinja
       - so/9999_output_redis.conf.jinja
     search:
       - so/0900_input_redis.conf.jinja
@@ -69,4 +69,5 @@ logstash:
     pipeline_x_batch_x_size: 125
     pipeline_x_ecs_compatibility: disabled
   dmz_nodes: []
+  latency_metrics: False
 

salt/logstash/pipelines/config/so/0012_input_elastic_agent.conf.jinja

@@ -1,3 +1,4 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
 input {
   elastic_agent {
     port => 5055
@@ -11,10 +12,15 @@ input {
   }
 }
 filter {
-if ![metadata] {
-  mutate {
-    rename => {"@metadata" => "metadata"}
+  {% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+  ruby {
+    code => "event.set('[_tmp][logstash_from_agent]', Time.now().utc.iso8601(3));"
+  }
+  {% endif %}
+  if ![metadata] {
+    mutate {
+      rename => {"@metadata" => "metadata"}
+    }
   }
 }
-}
 

salt/logstash/pipelines/config/so/0013_input_lumberjack_fleet.conf

@@ -1,23 +0,0 @@
-input {
-  elastic_agent {
-    port => 5056
-    tags => [ "elastic-agent", "fleet-lumberjack-input" ]
-    ssl_enabled => true
-    ssl_certificate => "/usr/share/logstash/elasticfleet-lumberjack.crt"
-    ssl_key => "/usr/share/logstash/elasticfleet-lumberjack.key"
-    ecs_compatibility => v8
-    id => "fleet-lumberjack-in"  
-    codec => "json"
-  }
-}
-
-
-filter {
-if ![metadata] {
-  mutate {
-    rename => {"@metadata" => "metadata"}
-  }
-}
-}
-
-

salt/logstash/pipelines/config/so/0013_input_lumberjack_fleet.conf.jinja

@@ -0,0 +1,26 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+input {
+  elastic_agent {
+    port => 5056
+    tags => [ "elastic-agent", "fleet-lumberjack-input" ]
+    ssl_enabled => true
+    ssl_certificate => "/usr/share/logstash/elasticfleet-lumberjack.crt"
+    ssl_key => "/usr/share/logstash/elasticfleet-lumberjack.key"
+    ecs_compatibility => v8
+    id => "fleet-lumberjack-in"
+    codec => "json"
+  }
+}
+
+filter {
+  {% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+  ruby {
+    code => "event.set('[_tmp][logstash_from_fleet]', Time.now().utc.iso8601(3));"
+  }
+  {% endif %}
+  if ![metadata] {
+    mutate {
+      rename => {"@metadata" => "metadata"}
+    }
+  }
+}

salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja

@@ -1,3 +1,4 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
 {%- set kafka_password = salt['pillar.get']('kafka:config:password') %}
 {%- set kafka_trustpass = salt['pillar.get']('kafka:config:trustpass') %}
 {%- set kafka_brokers = salt['pillar.get']('kafka:nodes', {}) %}
@@ -30,6 +31,11 @@ input {
     }
 }
 filter {
+  {% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+  ruby {
+    code => "event.set('[_tmp][logstash_from_kafka]', Time.now().utc.iso8601(3));"
+  }
+  {% endif %}
   if ![metadata] {
     mutate {
       rename => { "@metadata" => "metadata" }

salt/logstash/pipelines/config/so/0900_input_redis.conf.jinja

@@ -1,4 +1,4 @@
-{%- from 'logstash/map.jinja' import LOGSTASH_REDIS_NODES with context %}
+{%- from 'logstash/map.jinja' import LOGSTASH_REDIS_NODES, LOGSTASH_MERGED %}
 {%- set REDIS_PASS = salt['pillar.get']('redis:config:requirepass') %}
 
 {%- for index in range(LOGSTASH_REDIS_NODES|length) %}
@@ -18,3 +18,10 @@ input {
 }
 {%   endfor %}
 {% endfor -%}
+filter {
+  {% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+  ruby {
+    code => "event.set('[_tmp][logstash_from_redis]', Time.now().utc.iso8601(3));"
+  }
+  {% endif %}
+}

salt/logstash/pipelines/config/so/9805_output_elastic_agent.conf.jinja

@@ -1,3 +1,11 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+{% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+filter {
+  ruby {
+    code => "event.set('[_tmp][logstash_to_elasticsearch]', Time.now().utc.iso8601(3));"
+  }
+}
+{% endif %}
 output {
   if "elastic-agent" in [tags] and "so-ip-mappings" in [tags] {
     elasticsearch {

salt/logstash/pipelines/config/so/9806_output_lumberjack_fleet.conf.jinja

@@ -13,13 +13,20 @@ filter {
                     add_tag => "fleet-lumberjack-{{ GLOBALS.hostname }}"
           }
   }
-
-output { 
-    lumberjack { 
-        codec => json 
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+{% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+filter {
+  ruby {
+    code => "event.set('[_tmp][fleet_to_logstash]', Time.now().utc.iso8601(3));"
+  }
+}
+{% endif %}
+output {
+    lumberjack {
+        codec => json
         hosts => {{  FAILOVER_LOGSTASH_NODES }}
         ssl_certificate => "/usr/share/filebeat/ca.crt"
-        port => 5056 
+        port => 5056
         id => "fleet-lumberjack-{{ GLOBALS.hostname }}"
-        } 
+        }
     }

salt/logstash/pipelines/config/so/9999_output_redis.conf.jinja

@@ -1,10 +1,17 @@
+{%- from 'logstash/map.jinja' import LOGSTASH_MERGED %}
 {%- if grains.role in ['so-heavynode', 'so-receiver'] %}
   {%- set HOST = GLOBALS.hostname %}
 {%- else %}
   {%- set HOST = GLOBALS.manager %}
 {%- endif %}
 {%- set REDIS_PASS = salt['pillar.get']('redis:config:requirepass') %}
-
+{% if LOGSTASH_MERGED.get('latency_metrics', False) %}
+filter {
+  ruby {
+    code => "event.set('[_tmp][logstash_to_redis]', Time.now().utc.iso8601(3));"
+  }
+}
+{% endif %}
 output {
 	redis {
 		host => '{{ HOST }}'

salt/logstash/soc_logstash.yaml

@@ -86,3 +86,8 @@ logstash:
     multiline: True
     advanced: True
     forcedType: "[]string"
+  latency_metrics:
+    description: Enable latency metrics within events processed by logstash. Useful for pinpointing log ingest delay.
+    forcedType: bool
+    global: False
+    advanced: True

salt/manager/tools/sbin/so-detections-overrides-import

@@ -0,0 +1,381 @@
+#!/usr/bin/env python3
+
+# 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.
+
+# Imports detection overrides (e.g. from so-detections-backup) into the so-detection
+# index. Reads <publicId>.<ext> files (NDJSON, one override per line) from a source
+# directory, looks up the matching detection by publicId+engine, validates each
+# override against the same rules SOC enforces, dedupes against existing overrides
+# (operational fields only), and appends new ones.
+
+import argparse
+import ipaddress
+import json
+import os
+import re
+import sys
+from datetime import datetime
+
+import requests
+from requests.auth import HTTPBasicAuth
+import urllib3
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+DEFAULT_INDEX = "so-detection"
+AUTH_FILE = "/opt/so/conf/elasticsearch/curl.config"
+ES_URL = "https://localhost:9200"
+
+# Engines we know how to handle and the file extension the backup script writes.
+ENGINES = {
+    "suricata": "txt",
+}
+
+# Standard Suricata variables that ship with Security Onion. Anything else
+# referenced in an override is "custom" and the user needs to make sure it
+# exists in SOC Config before the override will function.
+BUILTIN_SURICATA_VARS = {
+    "$HOME_NET", "$EXTERNAL_NET",
+    "$HTTP_SERVERS", "$DNS_SERVERS", "$SQL_SERVERS", "$SMTP_SERVERS",
+    "$TELNET_SERVERS", "$AIM_SERVERS", "$DC_SERVERS", "$MODBUS_SERVER",
+    "$MODBUS_CLIENT", "$ENIP_CLIENT", "$ENIP_SERVER",
+    "$HTTP_PORTS", "$SHELLCODE_PORTS", "$ORACLE_PORTS", "$SSH_PORTS",
+    "$FTP_PORTS", "$FILE_DATA_PORTS",
+}
+
+VAR_PATTERN = re.compile(r"\$[A-Z_][A-Z0-9_]*")
+
+# Canonical valid values, per securityonion-soc/model/detection.go.
+SURICATA_OVERRIDE_TYPES = {"suppress", "threshold", "modify"}
+SUPPRESS_TRACKS = {"by_src", "by_dst", "by_either"}
+THRESHOLD_TRACKS = {"by_src", "by_dst", "by_both"}
+THRESHOLD_TYPES = {"limit", "threshold", "both"}
+
+STALE_WARNING = """\
+WARNING: so-detections-backup does not remove backup files when overrides are
+deleted via the Security Onion web UI. As a result, files in the source
+directory may represent overrides that were intentionally deleted and should
+NOT be re-imported.
+
+Before continuing, verify that the source directory reflects the overrides you
+actually want imported. Remove any files corresponding to overrides you previously deleted.
+"""
+
+
+def make_session(auth_file):
+    with open(auth_file, "r") as f:
+        for line in f:
+            if line.startswith("user ="):
+                creds = line.split("=", 1)[1].strip().replace('"', "")
+                user, _, password = creds.partition(":")
+                session = requests.Session()
+                session.auth = HTTPBasicAuth(user, password)
+                session.headers.update({"Content-Type": "application/json"})
+                session.verify = False
+                return session
+    raise RuntimeError(f"Could not find 'user =' line in {auth_file}")
+
+
+def find_detection(session, index, public_id, engine):
+    query = {
+        "query": {"bool": {"must": [
+            {"term": {"so_detection.publicId": public_id}},
+            {"term": {"so_detection.engine": engine}},
+        ]}},
+        "size": 2,
+    }
+    r = session.get(f"{ES_URL}/{index}/_search", json=query)
+    r.raise_for_status()
+    hits = r.json().get("hits", {}).get("hits", [])
+    if not hits:
+        return None, None, None
+    if len(hits) > 1:
+        # Shouldn't happen — publicId is unique per engine — but flag it.
+        print(f"  WARN: {len(hits)} detections matched publicId={public_id} engine={engine}; using first")
+    hit = hits[0]
+    existing = hit["_source"].get("so_detection", {}).get("overrides") or []
+    return hit["_id"], hit["_index"], existing
+
+
+def update_overrides(session, doc_index, doc_id, overrides):
+    body = {"doc": {"so_detection": {"overrides": overrides}}}
+    r = session.post(f"{ES_URL}/{doc_index}/_update/{doc_id}", json=body)
+    r.raise_for_status()
+    return r.json()
+
+
+def dedupe_key(override):
+    """Operational fields only, per Override.Equal() in detection.go.
+    Excludes timestamps and isEnabled so re-imports don't appear unique."""
+    t = override.get("type")
+    if t == "suppress":
+        return (t, override.get("track"), override.get("ip"))
+    if t == "threshold":
+        return (t, override.get("thresholdType"), override.get("track"),
+                override.get("count"), override.get("seconds"))
+    if t == "modify":
+        return (t, override.get("regex"), override.get("value"))
+
+
+def _validate_suricata_ip(ip):
+    if not ip:
+        return "ip cannot be empty"
+    if ip.startswith("$"):
+        return None
+    if ip.startswith("[") and ip.endswith("]"):
+        for part in ip[1:-1].split(","):
+            err = _validate_single_ip(part.strip())
+            if err:
+                return f"invalid IP in list: {err}"
+        return None
+    return _validate_single_ip(ip)
+
+
+def _validate_single_ip(ip):
+    try:
+        if "/" in ip:
+            ipaddress.ip_network(ip, strict=False)
+        else:
+            ipaddress.ip_address(ip)
+    except ValueError:
+        return f"invalid IP/CIDR {ip!r}"
+    return None
+
+
+def validate_override(override, engine):
+    """Mirror Override.Validate() from securityonion-soc/model/detection.go.
+    Returns None on success, an error string otherwise."""
+    t = override.get("type")
+    if not t:
+        return "override type is required"
+    if t not in SURICATA_OVERRIDE_TYPES:
+        return f"invalid type {t!r}: must be one of {sorted(SURICATA_OVERRIDE_TYPES)}"
+
+    has = {k: override.get(k) is not None for k in
+           ("regex", "value", "thresholdType", "track", "ip", "count", "seconds", "customFilter")}
+
+    if t == "suppress":
+        if not has["ip"] or not has["track"]:
+            return "suppress requires 'ip' and 'track'"
+        if any(has[k] for k in ("regex", "value", "thresholdType", "count", "seconds", "customFilter")):
+            return "suppress has unnecessary fields"
+        if override["track"] not in SUPPRESS_TRACKS:
+            return f"invalid track {override['track']!r}: must be one of {sorted(SUPPRESS_TRACKS)}"
+        return _validate_suricata_ip(override["ip"])
+
+    if t == "threshold":
+        if not all(has[k] for k in ("thresholdType", "track", "count", "seconds")):
+            return "threshold requires 'thresholdType', 'track', 'count', 'seconds'"
+        if any(has[k] for k in ("regex", "value", "customFilter")):
+            return "threshold has unnecessary fields"
+        if override["thresholdType"] not in THRESHOLD_TYPES:
+            return f"invalid thresholdType {override['thresholdType']!r}: must be one of {sorted(THRESHOLD_TYPES)}"
+        if override["track"] not in THRESHOLD_TRACKS:
+            return f"invalid track {override['track']!r}: must be one of {sorted(THRESHOLD_TRACKS)}"
+        if not isinstance(override["count"], int) or override["count"] <= 0:
+            return f"count must be a positive integer, got {override['count']!r}"
+        if not isinstance(override["seconds"], int) or override["seconds"] <= 0:
+            return f"seconds must be a positive integer, got {override['seconds']!r}"
+        return None
+
+    if t == "modify":
+        if not has["regex"] or not has["value"]:
+            return "modify requires 'regex' and 'value'"
+        if any(has[k] for k in ("thresholdType", "track", "count", "seconds", "customFilter")):
+            return "modify has unnecessary fields"
+        try:
+            re.compile(override["regex"])
+        except re.error as e:
+            return f"invalid regex: {e}"
+        return None
+
+
+def parse_overrides_file(path):
+    """Parse a file written by so-detections-backup.py: NDJSON, one override
+    per line. Returns a list of (override_dict, line_number)."""
+    overrides = []
+    with open(path, "r") as f:
+        for i, line in enumerate(f, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            overrides.append((json.loads(line), i))
+    return overrides
+
+
+def describe(override):
+    """Human-readable summary of the operational fields for a given override type."""
+    t = override.get("type")
+    if t == "suppress":
+        return f"type=suppress track={override.get('track')} ip={override.get('ip')}"
+    if t == "threshold":
+        return (f"type=threshold track={override.get('track')} "
+                f"thresholdType={override.get('thresholdType')} "
+                f"count={override.get('count')} seconds={override.get('seconds')}")
+    if t == "modify":
+        return f"type=modify regex={override.get('regex')!r}"
+
+
+def collect_custom_vars(override):
+    found = set()
+    for value in override.values():
+        if isinstance(value, str):
+            for match in VAR_PATTERN.findall(value):
+                if match not in BUILTIN_SURICATA_VARS:
+                    found.add(match)
+    return found
+
+
+def parse_args():
+    p = argparse.ArgumentParser(
+        description="Import detection overrides into the so-detection index.",
+    )
+    p.add_argument("--source", "-s", required=True,
+                   help="Source directory containing <publicId>.<ext> override files.")
+    p.add_argument("--engine", "-e", default="suricata", choices=list(ENGINES.keys()),
+                   help="Detection engine (default: suricata).")
+    p.add_argument("--dry-run", "-n", action="store_true",
+                   help="Print what would happen without writing to Elasticsearch.")
+    p.add_argument("--no-import-note", action="store_true",
+                   help="Do not prepend '[Imported YYYY-MM-DD] ' to the override note.")
+    p.add_argument("--index", "-i", default=DEFAULT_INDEX,
+                   help=f"Elasticsearch index to update (default: {DEFAULT_INDEX}).")
+    return p.parse_args()
+
+
+def confirm_proceed(args):
+    """Show the stale-backup warning. Dry-run prints it and continues. Real
+    runs require the user typing 'yes' at the prompt."""
+    print(STALE_WARNING)
+    if args.dry_run:
+        print("(dry-run: no acknowledgement required)\n")
+        return True
+    answer = input("Type 'yes' to acknowledge and continue: ").strip().lower()
+    print()
+    return answer == "yes"
+
+
+def main():
+    args = parse_args()
+
+    if not os.path.isdir(args.source):
+        print(f"ERROR: source directory not found: {args.source}", file=sys.stderr)
+        sys.exit(1)
+
+    extension = ENGINES[args.engine]
+    files = sorted(f for f in os.listdir(args.source) if f.endswith(f".{extension}"))
+    if not files:
+        print(f"No *.{extension} files found in {args.source}")
+        sys.exit(0)
+
+    if not confirm_proceed(args):
+        print("Aborted.")
+        sys.exit(1)
+
+    session = make_session(AUTH_FILE)
+    today = datetime.now().strftime("%Y-%m-%d")
+    note_prefix = "" if args.no_import_note else f"[Imported {today}] "
+
+    counts = {"added": 0, "skipped_dedupe": 0, "skipped_not_found": 0, "invalid": 0, "error": 0}
+    custom_vars = set()
+
+    mode = "DRY-RUN" if args.dry_run else "IMPORT"
+    print(f"[{mode}] engine={args.engine} source={args.source} index={args.index}\n")
+
+    for filename in files:
+        public_id = os.path.splitext(filename)[0]
+        path = os.path.join(args.source, filename)
+        print(f"{public_id}:")
+
+        try:
+            new_overrides = parse_overrides_file(path)
+        except (json.JSONDecodeError, OSError) as e:
+            print(f"  ERROR: could not parse {filename}: {e}")
+            counts["error"] += 1
+            continue
+
+        if not new_overrides:
+            print("  SKIP: empty file")
+            continue
+
+        try:
+            doc_id, doc_index, existing = find_detection(session, args.index, public_id, args.engine)
+        except requests.HTTPError as e:
+            print(f"  ERROR: search failed: {e}")
+            counts["error"] += 1
+            continue
+
+        if doc_id is None:
+            print(f"  WARN: no detection found for publicId={public_id} engine={args.engine}; skipping")
+            counts["skipped_not_found"] += len(new_overrides)
+            continue
+
+        existing_keys = {dedupe_key(o) for o in existing}
+        merged = list(existing)
+        added_this_file = 0
+
+        for override, line_no in new_overrides:
+            err = validate_override(override, args.engine)
+            if err:
+                print(f"  INVALID (line {line_no}): {err}")
+                counts["invalid"] += 1
+                continue
+
+            custom_vars.update(collect_custom_vars(override))
+            key = dedupe_key(override)
+            if key in existing_keys:
+                print(f"  SKIP (line {line_no}): duplicate of existing override [{describe(override)}]")
+                counts["skipped_dedupe"] += 1
+                continue
+
+            if note_prefix:
+                override = dict(override)
+                override["note"] = note_prefix + (override.get("note") or "")
+
+            merged.append(override)
+            existing_keys.add(key)
+            added_this_file += 1
+            print(f"  ADD (line {line_no}): {describe(override)}")
+
+        if added_this_file == 0:
+            continue
+
+        if args.dry_run:
+            print(f"  DRY-RUN: would update {doc_index}/{doc_id} "
+                  f"({len(existing)} existing → {len(merged)} total)")
+            counts["added"] += added_this_file
+            continue
+
+        try:
+            update_overrides(session, doc_index, doc_id, merged)
+            print(f"  UPDATED {doc_index}/{doc_id} ({len(existing)} → {len(merged)})")
+            counts["added"] += added_this_file
+        except requests.HTTPError as e:
+            print(f"  ERROR: update failed: {e}")
+            counts["error"] += 1
+
+    print()
+    print("=" * 60)
+    print(f"Summary ({mode}):")
+    print(f"  Overrides added:           {counts['added']}")
+    print(f"  Skipped (already present): {counts['skipped_dedupe']}")
+    print(f"  Skipped (no detection):    {counts['skipped_not_found']}")
+    print(f"  Invalid (failed checks):   {counts['invalid']}")
+    print(f"  Errors:                    {counts['error']}")
+
+    if custom_vars:
+        print()
+        print("WARNING: detected custom Suricata variables in imported overrides:")
+        for v in sorted(custom_vars):
+            print(f"  {v}")
+        print("If any of these are not already defined in SOC Config (Suricata variables),")
+        print("you must add them manually before the rules will function correctly.")
+
+    sys.exit(0 if counts["error"] == 0 and counts["invalid"] == 0 else 1)
+
+
+if __name__ == "__main__":
+    main()

salt/manager/tools/sbin/so-detections-overrides-import_test.py

@@ -0,0 +1,588 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+import importlib.util
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from importlib.machinery import SourceFileLoader
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+import requests
+
+# The script has no .py extension; spec_from_file_location can't auto-detect a
+# loader, so we hand it a SourceFileLoader explicitly. (load_module() is
+# deprecated in 3.14 and slated for removal in 3.15.)
+HERE = os.path.dirname(os.path.abspath(__file__))
+SCRIPT = os.path.join(HERE, "so-detections-overrides-import")
+_loader = SourceFileLoader("so_overrides_import", SCRIPT)
+_spec = importlib.util.spec_from_loader("so_overrides_import", _loader)
+soi = importlib.util.module_from_spec(_spec)
+_loader.exec_module(soi)
+
+
+class TestValidateSuppress(unittest.TestCase):
+    def test_valid(self):
+        self.assertIsNone(soi.validate_override(
+            {"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}, "suricata"))
+
+    def test_valid_var(self):
+        self.assertIsNone(soi.validate_override(
+            {"type": "suppress", "track": "by_either", "ip": "$HOME_NET"}, "suricata"))
+
+    def test_valid_cidr(self):
+        self.assertIsNone(soi.validate_override(
+            {"type": "suppress", "track": "by_dst", "ip": "10.0.0.0/8"}, "suricata"))
+
+    def test_valid_bracket_list(self):
+        self.assertIsNone(soi.validate_override(
+            {"type": "suppress", "track": "by_src", "ip": "[1.2.3.4,10.0.0.0/8]"}, "suricata"))
+
+    def test_missing_ip(self):
+        err = soi.validate_override({"type": "suppress", "track": "by_src"}, "suricata")
+        self.assertIn("requires", err)
+
+    def test_missing_track(self):
+        err = soi.validate_override({"type": "suppress", "ip": "1.2.3.4"}, "suricata")
+        self.assertIn("requires", err)
+
+    def test_invalid_track(self):
+        err = soi.validate_override(
+            {"type": "suppress", "track": "by_both", "ip": "1.2.3.4"}, "suricata")
+        self.assertIn("invalid track", err)
+
+    def test_invalid_ip(self):
+        err = soi.validate_override(
+            {"type": "suppress", "track": "by_src", "ip": "not-an-ip"}, "suricata")
+        self.assertIn("invalid IP", err)
+
+    def test_unnecessary_field(self):
+        err = soi.validate_override(
+            {"type": "suppress", "track": "by_src", "ip": "1.2.3.4", "count": 5}, "suricata")
+        self.assertIn("unnecessary fields", err)
+
+
+class TestValidateThreshold(unittest.TestCase):
+    def test_valid(self):
+        self.assertIsNone(soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "limit", "count": 10, "seconds": 60,
+        }, "suricata"))
+
+    def test_valid_by_both(self):
+        self.assertIsNone(soi.validate_override({
+            "type": "threshold", "track": "by_both",
+            "thresholdType": "both", "count": 1, "seconds": 1,
+        }, "suricata"))
+
+    def test_track_by_either_invalid(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_either",
+            "thresholdType": "limit", "count": 10, "seconds": 60,
+        }, "suricata")
+        self.assertIn("invalid track", err)
+
+    def test_invalid_threshold_type(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "bogus", "count": 10, "seconds": 60,
+        }, "suricata")
+        self.assertIn("invalid thresholdType", err)
+
+    def test_zero_count(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "limit", "count": 0, "seconds": 60,
+        }, "suricata")
+        self.assertIn("count", err)
+
+    def test_negative_seconds(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "limit", "count": 10, "seconds": -1,
+        }, "suricata")
+        self.assertIn("seconds", err)
+
+    def test_missing_field(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "limit", "count": 10,  # missing seconds
+        }, "suricata")
+        self.assertIn("requires", err)
+
+    def test_unnecessary_field(self):
+        err = soi.validate_override({
+            "type": "threshold", "track": "by_src",
+            "thresholdType": "limit", "count": 10, "seconds": 60,
+            "regex": "foo",
+        }, "suricata")
+        self.assertIn("unnecessary fields", err)
+
+
+class TestValidateModify(unittest.TestCase):
+    def test_valid(self):
+        self.assertIsNone(soi.validate_override(
+            {"type": "modify", "regex": r"content:\"foo\"", "value": "content:bar"}, "suricata"))
+
+    def test_invalid_regex(self):
+        err = soi.validate_override(
+            {"type": "modify", "regex": "(unbalanced", "value": "x"}, "suricata")
+        self.assertIn("invalid regex", err)
+
+    def test_missing_value(self):
+        err = soi.validate_override({"type": "modify", "regex": "x"}, "suricata")
+        self.assertIn("requires", err)
+
+    def test_unnecessary_field(self):
+        err = soi.validate_override(
+            {"type": "modify", "regex": "x", "value": "y", "track": "by_src"}, "suricata")
+        self.assertIn("unnecessary fields", err)
+
+
+class TestValidateMisc(unittest.TestCase):
+    def test_unknown_type(self):
+        err = soi.validate_override({"type": "suppresss", "track": "by_src", "ip": "1.2.3.4"}, "suricata")
+        self.assertIn("invalid type", err)
+
+    def test_missing_type(self):
+        err = soi.validate_override({"track": "by_src"}, "suricata")
+        self.assertIn("type is required", err)
+
+
+class TestValidateIP(unittest.TestCase):
+    def test_plain_ipv4(self):
+        self.assertIsNone(soi._validate_suricata_ip("1.2.3.4"))
+
+    def test_plain_ipv6(self):
+        self.assertIsNone(soi._validate_suricata_ip("::1"))
+
+    def test_cidr(self):
+        self.assertIsNone(soi._validate_suricata_ip("10.0.0.0/8"))
+
+    def test_var(self):
+        self.assertIsNone(soi._validate_suricata_ip("$CONCOURSEWORKERS"))
+
+    def test_bracket_list(self):
+        self.assertIsNone(soi._validate_suricata_ip("[1.2.3.4, 10.0.0.0/8]"))
+
+    def test_bracket_list_bad_member(self):
+        err = soi._validate_suricata_ip("[1.2.3.4,nope]")
+        self.assertIn("invalid IP in list", err)
+
+    def test_empty(self):
+        self.assertIn("empty", soi._validate_suricata_ip(""))
+
+    def test_invalid(self):
+        self.assertIn("invalid", soi._validate_suricata_ip("999.999.999.999"))
+
+
+class TestDedupeKey(unittest.TestCase):
+    def test_suppress(self):
+        a = {"type": "suppress", "track": "by_src", "ip": "1.2.3.4", "count": 99}
+        b = {"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}
+        # count is irrelevant for suppress dedupe
+        self.assertEqual(soi.dedupe_key(a), soi.dedupe_key(b))
+
+    def test_suppress_differs_on_ip(self):
+        a = {"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}
+        b = {"type": "suppress", "track": "by_src", "ip": "5.6.7.8"}
+        self.assertNotEqual(soi.dedupe_key(a), soi.dedupe_key(b))
+
+    def test_threshold(self):
+        a = {"type": "threshold", "track": "by_src", "thresholdType": "limit",
+             "count": 10, "seconds": 60, "ip": "ignored"}
+        b = {"type": "threshold", "track": "by_src", "thresholdType": "limit",
+             "count": 10, "seconds": 60}
+        self.assertEqual(soi.dedupe_key(a), soi.dedupe_key(b))
+
+    def test_threshold_differs_on_count(self):
+        a = {"type": "threshold", "track": "by_src", "thresholdType": "limit",
+             "count": 10, "seconds": 60}
+        b = {"type": "threshold", "track": "by_src", "thresholdType": "limit",
+             "count": 20, "seconds": 60}
+        self.assertNotEqual(soi.dedupe_key(a), soi.dedupe_key(b))
+
+    def test_modify(self):
+        a = {"type": "modify", "regex": "x", "value": "y"}
+        b = {"type": "modify", "regex": "x", "value": "y"}
+        self.assertEqual(soi.dedupe_key(a), soi.dedupe_key(b))
+
+
+class TestDescribe(unittest.TestCase):
+    def test_suppress(self):
+        s = soi.describe({"type": "suppress", "track": "by_src", "ip": "1.2.3.4"})
+        self.assertIn("suppress", s)
+        self.assertIn("by_src", s)
+        self.assertIn("1.2.3.4", s)
+
+    def test_threshold_includes_count(self):
+        s = soi.describe({"type": "threshold", "track": "by_src",
+                          "thresholdType": "limit", "count": 10, "seconds": 60})
+        self.assertIn("count=10", s)
+        self.assertIn("seconds=60", s)
+
+    def test_modify(self):
+        s = soi.describe({"type": "modify", "regex": "foo"})
+        self.assertIn("modify", s)
+        self.assertIn("foo", s)
+
+
+class TestParseOverridesFile(unittest.TestCase):
+    def _write(self, content):
+        fd, path = tempfile.mkstemp(suffix=".txt")
+        os.close(fd)
+        with open(path, "w") as f:
+            f.write(content)
+        self.addCleanup(os.unlink, path)
+        return path
+
+    def test_single_line(self):
+        path = self._write('{"type":"suppress","track":"by_src","ip":"1.2.3.4"}')
+        result = soi.parse_overrides_file(path)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0][0]["type"], "suppress")
+        self.assertEqual(result[0][1], 1)
+
+    def test_ndjson(self):
+        path = self._write(
+            '{"type":"suppress","track":"by_src","ip":"1.2.3.4"}\n'
+            '{"type":"suppress","track":"by_dst","ip":"5.6.7.8"}\n'
+        )
+        result = soi.parse_overrides_file(path)
+        self.assertEqual(len(result), 2)
+        self.assertEqual(result[1][1], 2)
+
+    def test_empty(self):
+        path = self._write("")
+        self.assertEqual(soi.parse_overrides_file(path), [])
+
+    def test_blank_lines_skipped(self):
+        path = self._write('\n{"type":"suppress","track":"by_src","ip":"1.2.3.4"}\n\n')
+        result = soi.parse_overrides_file(path)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0][1], 2)  # line number reflects original position
+
+    def test_invalid_raises(self):
+        path = self._write("not json")
+        with self.assertRaises(json.JSONDecodeError):
+            soi.parse_overrides_file(path)
+
+
+class TestCollectCustomVars(unittest.TestCase):
+    def test_finds_custom(self):
+        v = soi.collect_custom_vars({"ip": "$CONCOURSEWORKERS"})
+        self.assertEqual(v, {"$CONCOURSEWORKERS"})
+
+    def test_filters_builtins(self):
+        v = soi.collect_custom_vars({"ip": "$HOME_NET"})
+        self.assertEqual(v, set())
+
+    def test_mixed(self):
+        v = soi.collect_custom_vars({"ip": "[$HOME_NET,$MYNET]"})
+        self.assertEqual(v, {"$MYNET"})
+
+    def test_non_string_fields_ignored(self):
+        v = soi.collect_custom_vars({"count": 10, "isEnabled": True})
+        self.assertEqual(v, set())
+
+
+class TestMakeSession(unittest.TestCase):
+    def _write(self, content):
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        with open(path, "w") as f:
+            f.write(content)
+        self.addCleanup(os.unlink, path)
+        return path
+
+    def test_valid_auth_file(self):
+        path = self._write('user = "admin:secret"\n')
+        session = soi.make_session(path)
+        self.assertEqual(session.auth.username, "admin")
+        self.assertEqual(session.auth.password, "secret")
+        self.assertFalse(session.verify)
+
+    def test_missing_user_line(self):
+        path = self._write("# no user line here\n")
+        with self.assertRaises(RuntimeError):
+            soi.make_session(path)
+
+
+class TestFindDetection(unittest.TestCase):
+    def _session_with_response(self, payload):
+        session = MagicMock()
+        response = MagicMock()
+        response.json.return_value = payload
+        response.raise_for_status.return_value = None
+        session.get.return_value = response
+        return session
+
+    def test_found(self):
+        session = self._session_with_response({"hits": {"hits": [{
+            "_id": "abc", "_index": "so-detection",
+            "_source": {"so_detection": {"overrides": [{"type": "suppress"}]}},
+        }]}})
+        doc_id, idx, existing = soi.find_detection(session, "so-detection", "2049201", "suricata")
+        self.assertEqual(doc_id, "abc")
+        self.assertEqual(idx, "so-detection")
+        self.assertEqual(len(existing), 1)
+
+    def test_not_found(self):
+        session = self._session_with_response({"hits": {"hits": []}})
+        doc_id, idx, existing = soi.find_detection(session, "so-detection", "x", "suricata")
+        self.assertIsNone(doc_id)
+        self.assertIsNone(idx)
+        self.assertIsNone(existing)
+
+    def test_no_overrides_field(self):
+        session = self._session_with_response({"hits": {"hits": [{
+            "_id": "abc", "_index": "so-detection",
+            "_source": {"so_detection": {}},
+        }]}})
+        _, _, existing = soi.find_detection(session, "so-detection", "x", "suricata")
+        self.assertEqual(existing, [])
+
+    def test_multiple_hits_warns(self):
+        session = self._session_with_response({"hits": {"hits": [
+            {"_id": "a", "_index": "i", "_source": {"so_detection": {"overrides": []}}},
+            {"_id": "b", "_index": "i", "_source": {"so_detection": {"overrides": []}}},
+        ]}})
+        with patch("sys.stdout", new=StringIO()) as out:
+            doc_id, _, _ = soi.find_detection(session, "i", "x", "suricata")
+        self.assertEqual(doc_id, "a")
+        self.assertIn("WARN", out.getvalue())
+
+
+class TestUpdateOverrides(unittest.TestCase):
+    def test_posts_to_update_endpoint(self):
+        session = MagicMock()
+        response = MagicMock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {"result": "updated"}
+        session.post.return_value = response
+
+        result = soi.update_overrides(session, "so-detection", "abc", [{"type": "suppress"}])
+
+        self.assertEqual(result, {"result": "updated"})
+        url = session.post.call_args[0][0]
+        self.assertIn("/_update/abc", url)
+        body = session.post.call_args[1]["json"]
+        self.assertEqual(body["doc"]["so_detection"]["overrides"], [{"type": "suppress"}])
+
+
+class TestConfirmProceed(unittest.TestCase):
+    def test_dry_run_skips_prompt(self):
+        args = MagicMock(dry_run=True)
+        with patch("sys.stdout", new=StringIO()):
+            self.assertTrue(soi.confirm_proceed(args))
+
+    def test_yes_input(self):
+        args = MagicMock(dry_run=False)
+        with patch("sys.stdout", new=StringIO()):
+            with patch("builtins.input", return_value="yes"):
+                self.assertTrue(soi.confirm_proceed(args))
+
+    def test_yes_input_case_insensitive(self):
+        args = MagicMock(dry_run=False)
+        with patch("sys.stdout", new=StringIO()):
+            with patch("builtins.input", return_value="YES"):
+                self.assertTrue(soi.confirm_proceed(args))
+
+    def test_no_input_aborts(self):
+        args = MagicMock(dry_run=False)
+        with patch("sys.stdout", new=StringIO()):
+            with patch("builtins.input", return_value="no"):
+                self.assertFalse(soi.confirm_proceed(args))
+
+    def test_empty_input_aborts(self):
+        args = MagicMock(dry_run=False)
+        with patch("sys.stdout", new=StringIO()):
+            with patch("builtins.input", return_value=""):
+                self.assertFalse(soi.confirm_proceed(args))
+
+
+class TestParseArgs(unittest.TestCase):
+    def test_defaults(self):
+        with patch.object(sys, "argv", ["cmd", "--source", "/some/path"]):
+            args = soi.parse_args()
+        self.assertEqual(args.source, "/some/path")
+        self.assertEqual(args.engine, "suricata")
+        self.assertFalse(args.dry_run)
+        self.assertFalse(args.no_import_note)
+        self.assertEqual(args.index, soi.DEFAULT_INDEX)
+
+    def test_all_options(self):
+        argv = ["cmd", "-s", "/x", "-e", "suricata", "-n",
+                "--no-import-note", "-i", "alt-index"]
+        with patch.object(sys, "argv", argv):
+            args = soi.parse_args()
+        self.assertEqual(args.source, "/x")
+        self.assertTrue(args.dry_run)
+        self.assertTrue(args.no_import_note)
+        self.assertEqual(args.index, "alt-index")
+
+
+class TestMain(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmpdir, ignore_errors=True)
+        # Stub make_session so tests don't need /opt/so/conf/elasticsearch/curl.config.
+        p = patch.object(soi, "make_session", return_value=MagicMock())
+        p.start()
+        self.addCleanup(p.stop)
+
+    def _write_file(self, public_id, overrides, ext="txt"):
+        """Write an NDJSON override file. Entries may be dicts or raw strings (for malformed input)."""
+        path = os.path.join(self.tmpdir, f"{public_id}.{ext}")
+        with open(path, "w") as f:
+            for o in overrides:
+                f.write(o if isinstance(o, str) else json.dumps(o))
+                f.write("\n")
+        return path
+
+    def _run_main(self, *extra_argv, input_response="yes"):
+        """Run main() with stdout/stderr captured and input mocked. Returns (stdout, stderr, exit_code)."""
+        argv = ["cmd", "--source", self.tmpdir, *extra_argv]
+        out, err = StringIO(), StringIO()
+        with patch.object(sys, "argv", argv), \
+                patch("sys.stdout", new=out), \
+                patch("sys.stderr", new=err), \
+                patch("builtins.input", return_value=input_response):
+            with self.assertRaises(SystemExit) as cm:
+                soi.main()
+        return out.getvalue(), err.getvalue(), cm.exception.code
+
+    def test_source_dir_missing(self):
+        argv = ["cmd", "--source", "/no/such/path/here"]
+        err = StringIO()
+        with patch.object(sys, "argv", argv), patch("sys.stderr", new=err):
+            with self.assertRaises(SystemExit) as cm:
+                soi.main()
+        self.assertEqual(cm.exception.code, 1)
+        self.assertIn("source directory not found", err.getvalue())
+
+    def test_no_files_found(self):
+        out, _, code = self._run_main()
+        self.assertEqual(code, 0)
+        self.assertIn("No *.txt files found", out)
+
+    def test_user_aborts(self):
+        self._write_file("1001", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        out, _, code = self._run_main(input_response="no")
+        self.assertEqual(code, 1)
+        self.assertIn("Aborted", out)
+
+    def test_parse_error_increments_error(self):
+        # Malformed JSON line — parse_overrides_file raises JSONDecodeError.
+        self._write_file("1002", ["not json"])
+        out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 1)  # invalid+error → non-zero
+        self.assertIn("could not parse", out)
+        self.assertIn("Errors:                    1", out)
+
+    def test_empty_file_skipped(self):
+        # Blank lines only — parse_overrides_file returns []; main reports "empty file" and continues.
+        path = os.path.join(self.tmpdir, "1003.txt")
+        with open(path, "w") as f:
+            f.write("\n\n")
+        out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 0)
+        self.assertIn("empty file", out)
+
+    @patch.object(soi, "find_detection")
+    def test_search_http_error(self, mock_find):
+        mock_find.side_effect = requests.HTTPError("boom")
+        self._write_file("1004", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 1)
+        self.assertIn("search failed", out)
+
+    @patch.object(soi, "find_detection")
+    def test_no_detection_found(self, mock_find):
+        mock_find.return_value = (None, None, None)
+        self._write_file("1005", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 0)
+        self.assertIn("no detection found", out)
+        self.assertIn("Skipped (no detection):    1", out)
+
+    @patch.object(soi, "find_detection")
+    def test_all_duplicates_no_update(self, mock_find):
+        existing = [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}]
+        mock_find.return_value = ("doc1", "so-detection", existing)
+        self._write_file("1006", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 0)
+        self.assertIn("SKIP", out)
+        self.assertNotIn("DRY-RUN: would update", out)  # added_this_file == 0 branch
+
+    @patch.object(soi, "update_overrides")
+    @patch.object(soi, "find_detection")
+    def test_happy_path_full(self, mock_find, mock_update):
+        # Exercises: ADD, dedupe SKIP, INVALID, note prefix, UPDATE, custom-vars warning, exit=1 (invalid present)
+        existing = [{"type": "suppress", "track": "by_src", "ip": "9.9.9.9"}]
+        mock_find.return_value = ("doc1", "so-detection", existing)
+        mock_update.return_value = {"result": "updated"}
+        self._write_file("1007", [
+            {"type": "suppress", "track": "by_src", "ip": "1.2.3.4"},                # ADD
+            {"type": "suppress", "track": "by_src", "ip": "9.9.9.9"},                # SKIP (dupe of existing)
+            {"type": "suppress", "track": "bogus",  "ip": "1.2.3.4"},                # INVALID
+            {"type": "suppress", "track": "by_src", "ip": "$CONCOURSEWORKERS"},      # ADD + custom var
+        ])
+        out, _, code = self._run_main()
+        self.assertEqual(code, 1)  # one invalid -> non-zero
+
+        mock_update.assert_called_once()
+        merged = mock_update.call_args[0][3]
+        self.assertEqual(len(merged), 3)  # 1 existing + 2 new
+        new_notes = [o.get("note", "") for o in merged if o.get("ip") in ("1.2.3.4", "$CONCOURSEWORKERS")]
+        self.assertTrue(all(n.startswith("[Imported ") for n in new_notes))
+
+        self.assertIn("ADD", out)
+        self.assertIn("SKIP", out)
+        self.assertIn("INVALID", out)
+        self.assertIn("UPDATED", out)
+        self.assertIn("$CONCOURSEWORKERS", out)
+
+    @patch.object(soi, "update_overrides")
+    @patch.object(soi, "find_detection")
+    def test_no_import_note_preserves_note(self, mock_find, mock_update):
+        mock_find.return_value = ("doc1", "so-detection", [])
+        mock_update.return_value = {"result": "updated"}
+        self._write_file("1008", [
+            {"type": "suppress", "track": "by_src", "ip": "1.2.3.4", "note": "original"},
+        ])
+        _, _, code = self._run_main("--no-import-note")
+        self.assertEqual(code, 0)
+        merged = mock_update.call_args[0][3]
+        self.assertEqual(merged[0]["note"], "original")  # no prefix applied
+
+    @patch.object(soi, "find_detection")
+    def test_dry_run_skips_update(self, mock_find):
+        mock_find.return_value = ("doc1", "so-detection", [])
+        self._write_file("1009", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        with patch.object(soi, "update_overrides") as mock_update:
+            out, _, code = self._run_main("--dry-run")
+        self.assertEqual(code, 0)
+        mock_update.assert_not_called()
+        self.assertIn("DRY-RUN: would update", out)
+
+    @patch.object(soi, "update_overrides")
+    @patch.object(soi, "find_detection")
+    def test_update_http_error(self, mock_find, mock_update):
+        mock_find.return_value = ("doc1", "so-detection", [])
+        mock_update.side_effect = requests.HTTPError("nope")
+        self._write_file("1010", [{"type": "suppress", "track": "by_src", "ip": "1.2.3.4"}])
+        out, _, code = self._run_main()
+        self.assertEqual(code, 1)
+        self.assertIn("update failed", out)
+
+
+if __name__ == "__main__":
+    unittest.main()

salt/manager/tools/sbin/soup

@@ -556,14 +556,23 @@ check_transform_health_and_reauthorize() {
     # - unhealthy (any non-green health status)
     # - metadata has run_as_kibana_system: false (this fix is specific to transforms started prior to Kibana 9.3.3)
     # - are not orphaned (integration is not somehow missing/corrupt/uninstalled)
+    local tmp_transforms tmp_stats tmp_installed
+    tmp_transforms=$(mktemp)
+    tmp_stats=$(mktemp)
+    tmp_installed=$(mktemp)
+
+    echo "$transforms_doc" > "$tmp_transforms"
+    echo "$stats_doc"      > "$tmp_stats"
+    echo "$installed_doc"  > "$tmp_installed"
+
     local unhealthy_transforms
     unhealthy_transforms=$(jq -c -n \
-        --argjson t "$transforms_doc" \
-        --argjson s "$stats_doc" \
-        --argjson i "$installed_doc" '
-        ($i.items | map({key: .name, value: .version}) | from_entries) as $pkg_ver
-        | ($s.transforms | map({key: .id, value: .health.status}) | from_entries) as $health
-        | [ $t.transforms[]
+        --slurpfile t "$tmp_transforms" \
+        --slurpfile s "$tmp_stats" \
+        --slurpfile i "$tmp_installed" '
+        ($i[0].items | map({key: .name, value: .version}) | from_entries) as $pkg_ver
+        | ($s[0].transforms | map({key: .id, value: .health.status}) | from_entries) as $health
+        | [ $t[0].transforms[]
             | select(._meta.run_as_kibana_system == false)
             | select(($health[.id] // "unknown") != "green")
             | {id, pkg: ._meta.package.name, ver: ($pkg_ver[._meta.package.name])}
@@ -604,6 +613,8 @@ check_transform_health_and_reauthorize() {
         (( total_failures += $(jq 'map(select(.success != true)) | length' <<< "$resp" 2>/dev/null) ))
     done <<< "$unhealthy_transforms"
 
+    rm -f "$tmp_transforms" "$tmp_stats" "$tmp_installed"
+
     if [[ "$total_failures" -gt 0 ]]; then
         echo "Some transform(s) failed to reauthorize."
     fi
@@ -644,6 +655,27 @@ ensure_postgres_secret() {
   chown socore:socore "$secrets_file"
 }
 
+rename_strelka_scan_lnk() {
+  echo "Renaming strelka pillar ScanLNK to ScanLnk."
+  local STRELKA_FILE=/opt/so/saltstack/local/pillar/strelka/soc_strelka.sls
+  local MINIONDIR=/opt/so/saltstack/local/pillar/minions
+  local OLD_KEY=strelka.backend.config.backend.scanners.ScanLNK
+  local NEW_KEY=strelka.backend.config.backend.scanners.ScanLnk
+  local TMP_VALUE_FILE
+  TMP_VALUE_FILE=$(mktemp)
+
+  for pillar_file in "$STRELKA_FILE" "$MINIONDIR"/*.sls; do
+    [[ -f "$pillar_file" ]] || continue
+    # Skip if ScanLNK doesn't exist
+    so-yaml.py get "$pillar_file" "$OLD_KEY" > "$TMP_VALUE_FILE" 2>/dev/null || continue
+    echo "Found 'ScanLNK' key in $pillar_file. Renaming to 'ScanLnk'."
+    so-yaml.py add "$pillar_file" "$NEW_KEY" "file:$TMP_VALUE_FILE"
+    so-yaml.py remove "$pillar_file" "$OLD_KEY"
+  done
+
+  rm -f "$TMP_VALUE_FILE"
+}
+
 up_to_3.1.0() {
   ensure_postgres_local_pillar
   ensure_postgres_secret
@@ -651,7 +683,7 @@ up_to_3.1.0() {
   elasticsearch_backup_index_templates
   # Clear existing component template state file.
   rm -f /opt/so/state/esfleet_component_templates.json
-
+  rename_strelka_scan_lnk
 
   INSTALLEDVERSION=3.1.0
 }

salt/nginx/etc/nginx.conf

@@ -225,6 +225,7 @@ http {
 			limit_req             zone=auth_throttle burst={{ NGINXMERGED.config.throttle_login_burst }} nodelay;
 			limit_req_status      429;
 			proxy_pass            http://{{ GLOBALS.manager }}:4433;
+			proxy_set_header      Connection "Close";
 			proxy_read_timeout    90;
 			proxy_connect_timeout 90;
 			proxy_set_header      Host $host;
@@ -237,6 +238,7 @@ http {
 		location ~ ^/auth/.*?(whoami|logout|settings|errors|webauthn.js) {
 			rewrite               /auth/(.*) /$1 break;
 			proxy_pass            http://{{ GLOBALS.manager }}:4433;
+			proxy_set_header      Connection "Close";
 			proxy_read_timeout    90;
 			proxy_connect_timeout 90;
 			proxy_set_header      Host $host;

salt/orch/delete_hypervisor.sls

@@ -3,7 +3,14 @@
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.
 
-{% set hypervisor = pillar.minion_id %}
+{% set hypervisor = pillar.get('minion_id', '') %}
+
+{% if not hypervisor|regex_match('^([A-Za-z0-9._-]{1,253})$') %}
+{%   do salt.log.error('delete_hypervisor_orch: refusing unsafe minion_id=' ~ hypervisor) %}
+delete_hypervisor_invalid_minion_id:
+  test.fail_without_changes:
+    - name: delete_hypervisor_invalid_minion_id
+{% else %}
 
 ensure_hypervisor_mine_deleted:
   salt.function:
@@ -20,3 +27,5 @@ update_salt_cloud_profile:
     - sls:
       - salt.cloud.config
     - concurrent: True
+
+{% endif %}

salt/orch/vm_pillar_clean.sls

@@ -12,7 +12,14 @@
 {% if 'vrt' in salt['pillar.get']('features', []) %}
 
 {%   do salt.log.debug('vm_pillar_clean_orch: Running') %}
-{%   set vm_name = pillar.get('vm_name') %}
+{%   set vm_name = pillar.get('vm_name', '') %}
+
+{%   if not vm_name|regex_match('^([A-Za-z0-9._-]{1,253})$') %}
+{%     do salt.log.error('vm_pillar_clean_orch: refusing unsafe vm_name=' ~ vm_name) %}
+vm_pillar_clean_invalid_name:
+  test.fail_without_changes:
+    - name: vm_pillar_clean_invalid_name
+{%   else %}
 
 delete_adv_{{ vm_name }}_pillar:
   module.run:
@@ -24,6 +31,8 @@ delete_{{ vm_name }}_pillar:
     - file.remove:
       - path: /opt/so/saltstack/local/pillar/minions/{{ vm_name }}.sls
 
+{%   endif %}
+
 {% else %}
 
 {%   do salt.log.error(

salt/postgres/config.sls

@@ -46,10 +46,10 @@ postgresinitdir:
     - require:
       - file: postgresconfdir
 
-postgresinitusers:
+postgresinitdb:
   file.managed:
-    - name: /opt/so/conf/postgres/init/init-users.sh
-    - source: salt://postgres/files/init-users.sh
+    - name: /opt/so/conf/postgres/init/init-db.sh
+    - source: salt://postgres/files/init-db.sh
     - user: 939
     - group: 939
     - mode: 755

salt/postgres/enabled.sls

@@ -31,7 +31,7 @@ so-postgres:
       - POSTGRES_DB=securityonion
       # Passwords are delivered via mounted 0600 secret files, not plaintext env vars.
       # The upstream postgres image resolves POSTGRES_PASSWORD_FILE; entrypoint.sh and
-      # init-users.sh resolve SO_POSTGRES_PASS_FILE the same way.
+      # init-db.sh resolve SO_POSTGRES_PASS_FILE the same way.
       - POSTGRES_PASSWORD_FILE=/run/secrets/postgres_password
       - SO_POSTGRES_USER={{ SO_POSTGRES_USER }}
       - SO_POSTGRES_PASS_FILE=/run/secrets/so_postgres_pass
@@ -46,7 +46,7 @@ so-postgres:
       - /opt/so/conf/postgres/postgresql.conf:/conf/postgresql.conf:ro
       - /opt/so/conf/postgres/pg_hba.conf:/conf/pg_hba.conf:ro
       - /opt/so/conf/postgres/secrets:/run/secrets:ro
-      - /opt/so/conf/postgres/init/init-users.sh:/docker-entrypoint-initdb.d/init-users.sh:ro
+      - /opt/so/conf/postgres/init/init-db.sh:/docker-entrypoint-initdb.d/init-db.sh:ro
       - /etc/pki/postgres.crt:/conf/postgres.crt:ro
       - /etc/pki/postgres.key:/conf/postgres.key:ro
       - /etc/pki/tls/certs/intca.crt:/conf/ca.crt:ro
@@ -70,7 +70,7 @@ so-postgres:
     - watch:
       - file: postgresconf
       - file: postgreshba
-      - file: postgresinitusers
+      - file: postgresinitdb
       - file: postgres_super_secret
       - file: postgres_app_secret
       - x509: postgres_crt
@@ -78,7 +78,7 @@ so-postgres:
     - require:
       - file: postgresconf
       - file: postgreshba
-      - file: postgresinitusers
+      - file: postgresinitdb
       - file: postgres_super_secret
       - file: postgres_app_secret
       - x509: postgres_crt

salt/postgres/telegraf_users.sls

@@ -39,17 +39,15 @@ postgres_wait_ready:
     - require:
       - docker_container: so-postgres
 
-# Ensure the shared Telegraf database exists. init-users.sh only runs on a
+# Ensure the shared Telegraf database exists. init-db.sh only runs on a
 # fresh data dir, so hosts upgraded onto an existing /nsm/postgres volume
 # would otherwise never get so_telegraf.
 postgres_create_telegraf_db:
   cmd.run:
-    - name: |
-        if ! docker exec so-postgres psql -U postgres -tAc "SELECT 1 FROM pg_database WHERE datname='so_telegraf'" | grep -q 1; then
-          docker exec so-postgres psql -v ON_ERROR_STOP=1 -U postgres -c "CREATE DATABASE so_telegraf"
-        fi
+    - name: /usr/sbin/so-telegraf-postgres create_db
     - require:
       - cmd: postgres_wait_ready
+      - file: postgres_sbin
 
 # Provision the shared group role and schema once. Every per-minion role is a
 # member of so_telegraf, and each Telegraf connection does SET ROLE so_telegraf
@@ -57,68 +55,26 @@ postgres_create_telegraf_db:
 # on first write are owned by the group role and every member can INSERT/SELECT.
 postgres_telegraf_group_role:
   cmd.run:
-    - name: |
-        docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
-        DO $$
-        BEGIN
-            IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'so_telegraf') THEN
-                CREATE ROLE so_telegraf NOLOGIN;
-            END IF;
-        END
-        $$;
-        GRANT CONNECT ON DATABASE so_telegraf TO so_telegraf;
-        CREATE SCHEMA IF NOT EXISTS telegraf AUTHORIZATION so_telegraf;
-        GRANT USAGE, CREATE ON SCHEMA telegraf TO so_telegraf;
-        CREATE SCHEMA IF NOT EXISTS partman;
-        CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman;
-        CREATE EXTENSION IF NOT EXISTS pg_cron;
-        -- Telegraf (running as so_telegraf) calls partman.create_parent()
-        -- on first write of each metric, which needs USAGE on the partman
-        -- schema, EXECUTE on its functions/procedures, and write access to
-        -- partman.part_config so it can register new partitioned parents.
-        GRANT USAGE, CREATE ON SCHEMA partman TO so_telegraf;
-        GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA partman TO so_telegraf;
-        GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA partman TO so_telegraf;
-        GRANT EXECUTE ON ALL PROCEDURES IN SCHEMA partman TO so_telegraf;
-        -- partman creates per-parent template tables (partman.template_*) at
-        -- runtime; default privileges extend DML/sequence access to them.
-        ALTER DEFAULT PRIVILEGES IN SCHEMA partman
-            GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO so_telegraf;
-        ALTER DEFAULT PRIVILEGES IN SCHEMA partman
-            GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO so_telegraf;
-        -- Hourly partman maintenance. cron.schedule is idempotent by jobname.
-        SELECT cron.schedule(
-          'telegraf-partman-maintenance',
-          '17 * * * *',
-          'CALL partman.run_maintenance_proc()'
-        );
-        EOSQL
+    - name: /usr/sbin/so-telegraf-postgres group_role
     - require:
       - cmd: postgres_create_telegraf_db
+      - file: postgres_sbin
 
 {%   set creds = salt['pillar.get']('telegraf:postgres_creds', {}) %}
 {%   for mid, entry in creds.items() %}
 {%     if entry.get('user') and entry.get('pass') %}
 {%       set u = entry.user %}
-{%       set p = entry.pass | replace("'", "''") %}
+{%       set p = entry.pass %}
 
 postgres_telegraf_role_{{ u }}:
   cmd.run:
-    - name: |
-        docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
-        DO $$
-        BEGIN
-            IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = '{{ u }}') THEN
-                EXECUTE format('CREATE ROLE %I WITH LOGIN PASSWORD %L', '{{ u }}', '{{ p }}');
-            ELSE
-                EXECUTE format('ALTER ROLE %I WITH PASSWORD %L', '{{ u }}', '{{ p }}');
-            END IF;
-        END
-        $$;
-        GRANT CONNECT ON DATABASE so_telegraf TO "{{ u }}";
-        GRANT so_telegraf TO "{{ u }}";
-        EOSQL
+    - name: /usr/sbin/so-telegraf-postgres user
+    - env:
+      - ROLE_USER: {{ u | tojson }}
+      - ROLE_PASS: {{ p | tojson }}
+    - hide_output: True
     - require:
+      - file: postgres_sbin
       - cmd: postgres_telegraf_group_role
 
 {%     endif %}
@@ -130,21 +86,12 @@ postgres_telegraf_role_{{ u }}:
 {%   set retention = salt['pillar.get']('postgres:telegraf:retention_days', 14) | int %}
 postgres_telegraf_retention_reconcile:
   cmd.run:
-    - name: |
-        docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
-        DO $$
-        BEGIN
-            IF EXISTS (SELECT 1 FROM pg_catalog.pg_extension WHERE extname = 'pg_partman') THEN
-                UPDATE partman.part_config
-                SET retention = '{{ retention }} days',
-                    retention_keep_table = false
-                WHERE parent_table LIKE 'telegraf.%';
-            END IF;
-        END
-        $$;
-        EOSQL
+    - name: /usr/sbin/so-telegraf-postgres retention
+    - env:
+      - RETENTION_DAYS: {{ retention }}
     - require:
       - cmd: postgres_telegraf_group_role
+      - file: postgres_sbin
 
 {% endif %}
 

salt/postgres/tools/sbin/so-postgres-backup

@@ -7,15 +7,29 @@
 
 . /usr/sbin/so-common
 
+# Without pipefail, a pipeline's exit status is gzip's. A failed pg_dumpall would
+# otherwise be masked by a successful gzip, silently producing a valid .gz that
+# holds a truncated dump.
+set -o pipefail
+
 # Backups contain role password hashes and full chat data; keep them 0600.
 umask 0077
 
 TODAY=$(date '+%Y_%m_%d')
 BACKUPDIR=/nsm/backup
 BACKUPFILE="$BACKUPDIR/so-postgres-backup-$TODAY.sql.gz"
+TMPFILE="$BACKUPFILE.tmp"
 MAXBACKUPS=7
+LOGFILE=/opt/so/log/postgres/backup.log
 
-mkdir -p $BACKUPDIR
+log() {
+  echo "$(date '+%Y-%m-%d %H:%M:%S') $*" >> "$LOGFILE"
+}
+
+mkdir -p "$BACKUPDIR"
+
+# Remove any temp files left behind by a previously crashed run
+rm -f "$BACKUPDIR"/so-postgres-backup-*.sql.gz.tmp
 
 # Skip if already backed up today
 if [ -f "$BACKUPFILE" ]; then
@@ -27,13 +41,33 @@ if ! docker ps --format '{{.Names}}' | grep -q '^so-postgres$'; then
   exit 0
 fi
 
-# Dump all databases and roles, compress
-docker exec so-postgres pg_dumpall -U postgres | gzip > "$BACKUPFILE"
+# Always clean up the temp file on exit; the success path clears this trap
+# after the atomic rename so the finished backup is not deleted.
+trap 'rm -f "$TMPFILE"' EXIT
 
-# Retention cleanup
-NUMBACKUPS=$(find $BACKUPDIR -type f -name "so-postgres-backup*" | wc -l)
+# Dump all databases and roles, compress. Write to a temp file so the final
+# filename only ever appears for a complete, verified backup.
+if ! docker exec so-postgres pg_dumpall -U postgres | gzip > "$TMPFILE"; then
+  log "ERROR: pg_dumpall/gzip failed; backup aborted"
+  exit 1
+fi
+
+# Verify the compressed stream is intact before publishing it
+if ! gzip -t "$TMPFILE"; then
+  log "ERROR: backup failed gzip integrity check; backup aborted"
+  exit 1
+fi
+
+# Atomically publish the verified backup
+mv "$TMPFILE" "$BACKUPFILE"
+trap - EXIT
+log "OK: wrote $BACKUPFILE"
+
+# Retention cleanup (only reached after a successful backup). The glob is
+# restricted to finished backups so an in-progress .tmp can never be counted.
+NUMBACKUPS=$(find "$BACKUPDIR" -type f -name "so-postgres-backup-*.sql.gz" | wc -l)
 while [ "$NUMBACKUPS" -gt "$MAXBACKUPS" ]; do
-  OLDEST=$(find $BACKUPDIR -type f -name "so-postgres-backup*" -printf '%T+ %p\n' | sort | head -n 1 | awk -F" " '{print $2}')
+  OLDEST=$(find "$BACKUPDIR" -type f -name "so-postgres-backup-*.sql.gz" -printf '%T+ %p\n' | sort | head -n 1 | awk -F" " '{print $2}')
   rm -f "$OLDEST"
-  NUMBACKUPS=$(find $BACKUPDIR -type f -name "so-postgres-backup*" | wc -l)
+  NUMBACKUPS=$(find "$BACKUPDIR" -type f -name "so-postgres-backup-*.sql.gz" | wc -l)
 done

salt/postgres/tools/sbin/so-telegraf-postgres

@@ -0,0 +1,110 @@
+#!/bin/bash
+set -e
+
+# Provision Telegraf state inside the so-postgres container.
+# Usage: so-telegraf-postgres <subcommand>
+#   create_db    Ensure the so_telegraf database exists.
+#   group_role   Provision the so_telegraf group role, telegraf/partman schemas,
+#                pg_partman, pg_cron, and the hourly partman maintenance job.
+#   user         Create or update a per-minion login role granted to so_telegraf.
+#                Env: ROLE_USER, ROLE_PASS.
+#   retention    Reconcile partman retention on telegraf parents.
+#                Env: RETENTION_DAYS.
+
+cmd="${1:?subcommand required}"
+
+case "$cmd" in
+  create_db)
+    if ! docker exec so-postgres psql -U postgres -tAc \
+        "SELECT 1 FROM pg_database WHERE datname='so_telegraf'" | grep -q 1; then
+      docker exec so-postgres psql -v ON_ERROR_STOP=1 -U postgres \
+        -c "CREATE DATABASE so_telegraf"
+    fi
+    ;;
+
+  group_role)
+    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'so_telegraf') THEN
+        CREATE ROLE so_telegraf NOLOGIN;
+    END IF;
+END
+$$;
+GRANT CONNECT ON DATABASE so_telegraf TO so_telegraf;
+CREATE SCHEMA IF NOT EXISTS telegraf AUTHORIZATION so_telegraf;
+GRANT USAGE, CREATE ON SCHEMA telegraf TO so_telegraf;
+CREATE SCHEMA IF NOT EXISTS partman;
+CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman;
+CREATE EXTENSION IF NOT EXISTS pg_cron;
+-- Telegraf (running as so_telegraf) calls partman.create_parent()
+-- on first write of each metric, which needs USAGE on the partman
+-- schema, EXECUTE on its functions/procedures, and write access to
+-- partman.part_config so it can register new partitioned parents.
+GRANT USAGE, CREATE ON SCHEMA partman TO so_telegraf;
+GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA partman TO so_telegraf;
+GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA partman TO so_telegraf;
+GRANT EXECUTE ON ALL PROCEDURES IN SCHEMA partman TO so_telegraf;
+-- partman creates per-parent template tables (partman.template_*) at
+-- runtime; default privileges extend DML/sequence access to them.
+ALTER DEFAULT PRIVILEGES IN SCHEMA partman
+    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO so_telegraf;
+ALTER DEFAULT PRIVILEGES IN SCHEMA partman
+    GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO so_telegraf;
+-- Hourly partman maintenance. cron.schedule is idempotent by jobname.
+SELECT cron.schedule(
+  'telegraf-partman-maintenance',
+  '17 * * * *',
+  'CALL partman.run_maintenance_proc()'
+);
+EOSQL
+    ;;
+
+  user)
+    : "${ROLE_USER:?ROLE_USER is required}"
+    : "${ROLE_PASS:?ROLE_PASS is required}"
+    # psql does not substitute :vars inside dollar-quoted strings, so the
+    # conditional CREATE/ALTER is built outside any DO block and dispatched
+    # with \gexec. format() handles identifier/literal quoting.
+    docker exec -i so-postgres psql \
+      -v ON_ERROR_STOP=1 \
+      -v role_user="$ROLE_USER" \
+      -v role_pass="$ROLE_PASS" \
+      -U postgres -d so_telegraf <<'EOSQL'
+SELECT format(
+  CASE WHEN EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = :'role_user')
+       THEN 'ALTER ROLE %I WITH LOGIN PASSWORD %L'
+       ELSE 'CREATE ROLE %I WITH LOGIN PASSWORD %L'
+  END,
+  :'role_user',
+  :'role_pass'
+) \gexec
+GRANT CONNECT ON DATABASE so_telegraf TO :"role_user";
+GRANT so_telegraf TO :"role_user";
+EOSQL
+    ;;
+
+  retention)
+    : "${RETENTION_DAYS:?RETENTION_DAYS is required}"
+    # \gset + \if guards against a missing pg_partman without using a DO
+    # block (psql :var substitution doesn't reach into dollar-quoted code).
+    docker exec -i so-postgres psql \
+      -v ON_ERROR_STOP=1 \
+      -v retention_days="$RETENTION_DAYS" \
+      -U postgres -d so_telegraf <<'EOSQL'
+SELECT CASE WHEN EXISTS (SELECT 1 FROM pg_catalog.pg_extension WHERE extname = 'pg_partman')
+            THEN 'true' ELSE 'false' END AS has_partman \gset
+\if :has_partman
+UPDATE partman.part_config
+SET retention = :'retention_days' || ' days',
+    retention_keep_table = false
+WHERE parent_table LIKE 'telegraf.%';
+\endif
+EOSQL
+    ;;
+
+  *)
+    echo "Unknown subcommand: $cmd" >&2
+    exit 1
+    ;;
+esac

salt/reactor/check_hypervisor.sls

@@ -3,12 +3,15 @@
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.
 
-{% if data['id'].endswith('_hypervisor') and data['result'] == True %}
+{% set hid = data['id'] %}
+{% if hid|regex_match('^([A-Za-z0-9._-]{1,253})$')
+   and hid.endswith('_hypervisor')
+   and data['result'] == True %}
 
 {%   if data['act'] == 'accept' %}
 check_and_trigger:
   runner.setup_hypervisor.setup_environment:
-    - minion_id: {{ data['id'] }}
+    - minion_id: {{ hid }}
 {%   endif %}
 
 {%   if data['act'] == 'delete' %}
@@ -17,8 +20,7 @@ delete_hypervisor:
     - args:
       - mods: orch.delete_hypervisor
       - pillar:
-          minion_id: {{ data['id'] }}
+          minion_id: {{ hid }}
 {%   endif %}
 
 {% endif %}
-

salt/reactor/createEmptyPillar.sls

@@ -1,7 +1,7 @@
 #!py
 
 # Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
-# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.
 
@@ -9,30 +9,42 @@ import logging
 import os
 import pwd
 import grp
+import re
+
+log = logging.getLogger(__name__)
+
+PILLAR_ROOT = '/opt/so/saltstack/local/pillar/minions/'
+_VMNAME_RE = re.compile(r'^[A-Za-z0-9._-]{1,253}$')
+
 
 def run():
-  vm_name = data['kwargs']['name']
-  logging.error("createEmptyPillar reactor: vm_name: %s" % vm_name)
-  pillar_root = '/opt/so/saltstack/local/pillar/minions/'
+  vm_name = data.get('kwargs', {}).get('name', '')
+  if not _VMNAME_RE.match(str(vm_name)):
+    log.error("createEmptyPillar reactor: refusing unsafe vm_name=%r", vm_name)
+    return {}
+
+  log.info("createEmptyPillar reactor: vm_name: %s", vm_name)
   pillar_files = ['adv_' + vm_name + '.sls', vm_name + '.sls']
 
   try:
-    # Get socore user and group IDs
     socore_uid = pwd.getpwnam('socore').pw_uid
     socore_gid = grp.getgrnam('socore').gr_gid
+    pillar_root_real = os.path.realpath(PILLAR_ROOT)
 
     for f in pillar_files:
-      full_path = pillar_root + f
-      if not os.path.exists(full_path):
-        # Create empty file
-        os.mknod(full_path)
-        # Set ownership to socore:socore
-        os.chown(full_path, socore_uid, socore_gid)
-        # Set mode to 644 (rw-r--r--)
-        os.chmod(full_path, 0o640)
-        logging.error("createEmptyPillar reactor: created %s with socore:socore ownership and mode 644" % f)
+      full_path = os.path.join(PILLAR_ROOT, f)
+      resolved = os.path.realpath(full_path)
+      if os.path.dirname(resolved) != pillar_root_real:
+        log.error("createEmptyPillar reactor: refusing path outside pillar root: %s", resolved)
+        continue
+      if os.path.exists(resolved):
+        continue
+      os.mknod(resolved)
+      os.chown(resolved, socore_uid, socore_gid)
+      os.chmod(resolved, 0o640)
+      log.info("createEmptyPillar reactor: created %s with socore:socore ownership and mode 0640", f)
 
   except (KeyError, OSError) as e:
-    logging.error("createEmptyPillar reactor: Error setting ownership/permissions: %s" % str(e))
+    log.error("createEmptyPillar reactor: Error setting ownership/permissions: %s", e)
 
   return {}

salt/reactor/deleteKey.sls

@@ -1,18 +1,40 @@
+#!py
+
 # 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.
 
-remove_key:
-  wheel.key.delete:
-    - args:
-      - match: {{ data['name'] }}
+import logging
+import re
 
-{{ data['name'] }}_pillar_clean:
-  runner.state.orchestrate:
-    - args:
-      - mods: orch.vm_pillar_clean
-      - pillar:
-          vm_name: {{ data['name'] }}
+log = logging.getLogger(__name__)
 
-{% do salt.log.info('deleteKey reactor: deleted minion key: %s' % data['name']) %}
+_VMNAME_RE = re.compile(r'^[A-Za-z0-9._-]{1,253}$')
+
+
+def run():
+  name = data.get('name', '')
+  if not _VMNAME_RE.match(str(name)):
+    log.error("deleteKey reactor: refusing unsafe name=%r", name)
+    return {}
+
+  log.info("deleteKey reactor: deleted minion key: %s", name)
+
+  return {
+    'remove_key': {
+      'wheel.key.delete': [
+        {'args': [
+          {'match': name},
+        ]},
+      ],
+    },
+    '%s_pillar_clean' % name: {
+      'runner.state.orchestrate': [
+        {'args': [
+          {'mods': 'orch.vm_pillar_clean'},
+          {'pillar': {'vm_name': name}},
+        ]},
+      ],
+    },
+  }

salt/soc/files/soc/sigma_so_pipeline.yaml

@@ -117,6 +117,121 @@ transformations:
       - type: logsource
         product: linux
         service: auth
+    # Maps M365 audit rules to Elastic Agent O365 integration logs
+    - id: m365_audit_field_mappings
+      type: field_name_mapping
+      mapping:
+        Operation: event.action
+        ResultStatus: event.outcome
+        ApplicationId: o365.audit.ApplicationId
+        ObjectId: o365.audit.ObjectId
+        RequestType: o365.audit.RequestType
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: audit
+    - id: m365_audit_add-fields
+      type: add_condition
+      conditions:
+        event.dataset: 'o365.audit'
+        event.module: 'o365'
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: audit
+    # Maps M365 exchange rules to Elastic Agent O365 integration logs
+    - id: m365_exchange_field_mappings
+      type: field_name_mapping
+      mapping:
+        eventSource: event.provider
+        eventName: event.action
+        status: event.outcome
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: exchange
+    - id: m365_exchange_add-fields
+      type: add_condition
+      conditions:
+        event.dataset: 'o365.audit'
+        event.module: 'o365'
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: exchange
+    # Maps M365 threat_management rules to Elastic Agent O365 integration logs
+    - id: m365_threat_management_field_mappings
+      type: field_name_mapping
+      mapping:
+        eventSource: event.provider
+        eventName: event.action
+        status: event.outcome
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: threat_management
+    - id: m365_threat_management_add-fields
+      type: add_condition
+      conditions:
+        event.dataset: 'o365.audit'
+        event.module: 'o365'
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: threat_management
+    # Maps M365 threat_detection rules to Elastic Agent O365 integration logs
+    - id: m365_threat_detection_field_mappings
+      type: field_name_mapping
+      mapping:
+        eventSource: event.provider
+        eventName: event.action
+        status: event.outcome
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: threat_detection
+    - id: m365_threat_detection_add-fields
+      type: add_condition
+      conditions:
+        event.dataset: 'o365.audit'
+        event.module: 'o365'
+      rule_conditions:
+      - type: logsource
+        product: m365
+        service: threat_detection
+    # Maps FortiGate event rules to Elastic Agent Fortinet integration logs
+    - id: fortigate_event_field_mappings
+      type: field_name_mapping
+      mapping:
+        action: fortinet.firewall.action
+        cfgpath: fortinet.firewall.cfgpath
+        cfgobj: fortinet.firewall.cfgobj
+        cfgattr: fortinet.firewall.cfgattr
+        devname: observer.name
+        devid: observer.serial_number
+        logid: event.code
+        type: fortinet.firewall.type
+        subtype: fortinet.firewall.subtype
+        level: log.level
+        vd: fortinet.firewall.vd
+        logdesc: fortinet.firewall.desc
+        user: user.name
+        ui: fortinet.firewall.ui
+        cfgtid: fortinet.firewall.cfgtid
+        msg: message
+      rule_conditions:
+      - type: logsource
+        product: fortigate
+        service: event
+    - id: fortigate_event_add-fields
+      type: add_condition
+      conditions:
+        event.dataset: 'fortinet_fortigate.log'
+        event.module: 'fortinet_fortigate'
+      rule_conditions:
+      - type: logsource
+        product: fortigate
+        service: event
     # event.code should always be a string
     - id: convert_event_code_to_string
       type: convert_type
@@ -126,15 +241,36 @@ transformations:
           fields:
           - event.code
     # Maps process_creation rules to endpoint process creation logs
-    # This is an OS-agnostic mapping, to account for logs that don't specify source OS
     - id: endpoint_process_create_windows_add-fields
       type: add_condition
       conditions:
         event.category: 'process'
         event.type: 'start'
+        host.os.type: 'windows'
       rule_conditions:
       - type: logsource
         category: process_creation
+        product: windows
+    - id: endpoint_process_create_macos_add-fields
+      type: add_condition
+      conditions:
+        event.category: 'process'
+        event.type: 'start'
+        host.os.type: 'macos'
+      rule_conditions:
+      - type: logsource
+        category: process_creation
+        product: macos
+    - id: endpoint_process_create_linux_add-fields
+      type: add_condition
+      conditions:
+        event.category: 'process'
+        event.type: 'start'
+        host.os.type: 'linux'
+      rule_conditions:
+      - type: logsource
+        category: process_creation
+        product: linux
     # Maps file_event rules to endpoint file creation logs
     # This is an OS-agnostic mapping, to account for logs that don't specify source OS
     - id: endpoint_file_create_add-fields

salt/strelka/defaults.yaml

@@ -261,7 +261,7 @@ strelka:
               priority: 5
               options:
                 limit: 1000
-          'ScanLNK':
+          'ScanLnk':
             - positive:
                 flavors:
                   - 'lnk_file'

salt/strelka/soc_strelka.yaml

@@ -99,7 +99,7 @@ strelka:
           'ScanJpeg': *scannerOptions
           'ScanJson': *scannerOptions
           'ScanLibarchive': *scannerOptions
-          'ScanLNK': *scannerOptions
+          'ScanLnk': *scannerOptions
           'ScanLsb': *scannerOptions
           'ScanLzma': *scannerOptions
           'ScanMacho': *scannerOptions

salt/telegraf/defaults.yaml

@@ -1,6 +1,6 @@
 telegraf:
   enabled: False
-  output: BOTH
+  output: INFLUXDB
   config:
     interval: '30s'
     metric_batch_size: 1000

salt/top.sls

@@ -119,7 +119,7 @@ base:
     - kafka
     - pcap.cleanup
 
-  '*_manager or *_managerhype and G@saltversion:{{saltversion}} and not I@node_data:False':
+  '*_manager and G@saltversion:{{saltversion}} and not I@node_data:False':
     - match: compound
     - salt.master
     - registry
@@ -146,6 +146,32 @@ base:
     - stig
     - kafka
 
+  '*_managerhype and G@saltversion:{{saltversion}} and not I@node_data:False':
+    - match: compound
+    - salt.master
+    - registry
+    - nginx
+    - influxdb
+    - postgres
+    - strelka.manager
+    - soc
+    - kratos
+    - hydra
+    - firewall
+    - manager
+    - sensoroni
+    - telegraf
+    - backup.config_backup
+    - elasticsearch
+    - logstash
+    - redis
+    - elastic-fleet-package-registry
+    - kibana
+    - elastalert
+    - utility
+    - elasticfleet
+    - kafka
+
   '*_managerhype and I@features:vrt and G@saltversion:{{saltversion}}':
     - match: compound
     - manager.hypervisor
@@ -286,7 +312,6 @@ base:
     - libvirt
     - libvirt.images
     - elasticfleet.install_agent_grid
-    - stig
   
   '*_desktop and G@saltversion:{{saltversion}}':
     - sensoroni

setup/so-functions

@@ -745,6 +745,56 @@ configure_network_sensor() {
 	return $err
 }
 
+configure_management_bond() {
+	local bond_name="bond1"
+	local bond_mode=${MBOND_MODE:-active-backup}
+
+	info "Setting up $bond_name management interface with mode $bond_mode"
+
+	if [[ ${#MBNICS[@]} -eq 0 ]]; then
+		error "[ERROR] No management bond NICs were selected."
+		fail_setup
+	fi
+
+	nmcli -t -f NAME con show | grep -Fxq "$bond_name"
+	local found_int=$?
+
+	if [[ $found_int != 0 ]]; then
+		nmcli con add type bond ifname "$bond_name" con-name "$bond_name" mode "$bond_mode" -- \
+			ipv6.method ignore \
+			connection.autoconnect yes >> "$setup_log" 2>&1
+	else
+		nmcli con mod "$bond_name" \
+			bond.options "mode=$bond_mode" \
+			ipv6.method ignore \
+			connection.autoconnect yes >> "$setup_log" 2>&1
+	fi
+
+	local err=0
+	for MBNIC in "${MBNICS[@]}"; do
+		local slave_name="$bond_name-slave-$MBNIC"
+
+		nmcli -t -f NAME con show | grep -Fxq "$slave_name"
+		found_int=$?
+
+		if [[ $found_int != 0 ]]; then
+			nmcli con add type ethernet ifname "$MBNIC" con-name "$slave_name" master "$bond_name" -- \
+				connection.autoconnect yes >> "$setup_log" 2>&1
+		else
+			nmcli con mod "$slave_name" \
+				connection.master "$bond_name" \
+				connection.slave-type bond \
+				connection.autoconnect yes >> "$setup_log" 2>&1
+		fi
+
+		nmcli con up "$slave_name" >> "$setup_log" 2>&1
+		local ret=$?
+		[[ $ret -eq 0 ]] || err=$ret
+	done
+
+	return $err
+}
+
 configure_hyper_bridge() {
 	info "Setting up hypervisor bridge"
 	info "Checking $MNIC ipv4.method is auto or manual"
@@ -999,6 +1049,11 @@ filter_unused_nics() {
 			grep_string="$grep_string\|$BONDNIC"
 		done
 	fi
+	if [[ $MBNICS ]]; then
+		for BONDNIC in "${MBNICS[@]}"; do
+			grep_string="$grep_string\|$BONDNIC"
+		done
+	fi
 
 	# Finally, set filtered_nics to any NICs we aren't using (and ignore interfaces that aren't of use)
 	filtered_nics=$(ip link | awk -F: '$0 !~ "lo|vir|veth|br|docker|wl|^[^0-9]"{print $2}' | grep -vwe "$grep_string"  | sed 's/ //g' | sed -r 's/(.*)(\.[0-9]+)@\1/\1\2/g')
@@ -1388,7 +1443,7 @@ network_init() {
 	title "Initializing Network"
 	disable_ipv6
 	set_hostname
-	if [[ ( $is_iso || $is_desktop_iso ) ]]; then
+	if [[ $is_iso || $is_desktop_iso ]]; then
 		set_management_interface
 	fi
 }
@@ -1701,6 +1756,24 @@ remove_package() {
 	fi
 }
 
+ensure_pyyaml() {
+	title "Ensuring python3-pyyaml is installed"
+	if rpm -q python3-pyyaml >/dev/null 2>&1; then
+		info "python3-pyyaml already installed"
+		return 0
+	fi
+	info "python3-pyyaml not found, attempting to install"
+	set -o pipefail
+	dnf -y install python3-pyyaml 2>&1 | tee -a "$setup_log"
+	local result=$?
+	set +o pipefail
+	if [[ $result -ne 0 ]] || ! rpm -q python3-pyyaml >/dev/null 2>&1; then
+		error "Failed to install python3-pyyaml (exit=$result)"
+		fail_setup
+	fi
+	info "python3-pyyaml installed successfully"
+}
+
 # When updating the salt version, also update the version in securityonion-builds/images/iso-task/Dockerfile and salt/salt/master.defaults.yaml and salt/salt/minion.defaults.yaml
 # CAUTION! SALT VERSION UDDATES - READ BELOW
 # When updating the salt version, also update the version in:
@@ -2084,8 +2157,12 @@ set_initial_firewall_access() {
 # Set up the management interface on the ISO
 set_management_interface() {
 	title "Setting up the main interface"
+	if [[ $MNIC == "bond1" ]]; then
+		configure_management_bond || fail_setup
+	fi
+
 	if [ "$address_type" = 'DHCP' ]; then
-		logCmd "nmcli con mod $MNIC connection.autoconnect yes"
+		logCmd "nmcli con mod $MNIC connection.autoconnect yes ipv4.method auto"
 		logCmd "nmcli con up $MNIC"
 		logCmd "nmcli -p connection show $MNIC"
 	else

setup/so-setup

@@ -66,6 +66,9 @@ set_timezone
 # Let's see what OS we are dealing with here
 detect_os
 
+# Ensure python3-pyyaml is available before any code that may need so-yaml/PyYAML
+ensure_pyyaml
+
 
 # Check to see if this is the setup type of "desktop".
 is_desktop=

setup/so-whiptail

@@ -845,18 +845,99 @@ whiptail_management_nic() {
 	[ -n "$TESTING" ] && return
 
 	filter_unused_nics
+	local management_nic_options=( "${nic_list_management[@]}" )
+	if [[ $is_iso || $is_desktop_iso ]]; then
+		management_nic_options+=( "BOND" "Configure a bonded management interface" )
+	fi
 
-	MNIC=$(whiptail --title "$whiptail_title" --menu "Please select the NIC you would like to use for management.\n\nUse the arrow keys to move around and the Enter key to select." 20 75 12 "${nic_list_management[@]}" 3>&1 1>&2 2>&3 )	
+	MNIC=$(whiptail --title "$whiptail_title" --menu "Please select the NIC you would like to use for management.\n\nUse the arrow keys to move around and the Enter key to select." 20 75 12 "${management_nic_options[@]}" 3>&1 1>&2 2>&3 )
 	local exitstatus=$?
 	whiptail_check_exitstatus $exitstatus
 
 	while [ -z "$MNIC" ]
 	do
-		MNIC=$(whiptail --title "$whiptail_title" --menu "Please select the NIC you would like to use for management.\n\nUse the arrow keys to move around and the Enter key to select." 22 75 12 "${nic_list_management[@]}" 3>&1 1>&2 2>&3 )	
+		MNIC=$(whiptail --title "$whiptail_title" --menu "Please select the NIC you would like to use for management.\n\nUse the arrow keys to move around and the Enter key to select." 22 75 12 "${management_nic_options[@]}" 3>&1 1>&2 2>&3 )
 		local exitstatus=$?
 		whiptail_check_exitstatus $exitstatus
 	done
 
+	if [[ $MNIC == "BOND" ]]; then
+		whiptail_management_bond
+	fi
+}
+
+whiptail_management_bond() {
+
+	[ -n "$TESTING" ] && return
+
+	MBOND_MODE=$(whiptail --title "$whiptail_title" --menu \
+	"Choose the bond mode for the management interface.\n\nThe management bond will be created as bond1." 20 75 7 \
+	"active-backup" "One active NIC with failover (recommended)" \
+	"balance-rr" "Round-robin transmit policy" \
+	"balance-xor" "Transmit based on selected hash policy" \
+	"broadcast" "Transmit everything on all slave interfaces" \
+	"802.3ad" "Dynamic link aggregation (requires switch support)" \
+	"balance-tlb" "Adaptive transmit load balancing" \
+	"balance-alb" "Adaptive load balancing" 3>&1 1>&2 2>&3)
+	local exitstatus=$?
+	whiptail_check_exitstatus $exitstatus
+
+	while [ -z "$MBOND_MODE" ]
+	do
+		MBOND_MODE=$(whiptail --title "$whiptail_title" --menu \
+		"Choose the bond mode for the management interface.\n\nThe management bond will be created as bond1." 20 75 7 \
+		"active-backup" "One active NIC with failover (recommended)" \
+		"balance-rr" "Round-robin transmit policy" \
+		"balance-xor" "Transmit based on selected hash policy" \
+		"broadcast" "Transmit everything on all slave interfaces" \
+		"802.3ad" "Dynamic link aggregation (requires switch support)" \
+		"balance-tlb" "Adaptive transmit load balancing" \
+		"balance-alb" "Adaptive load balancing" 3>&1 1>&2 2>&3)
+		local exitstatus=$?
+		whiptail_check_exitstatus $exitstatus
+	done
+
+	whiptail_management_bond_nics
+	MNIC="bond1"
+
+	export MBOND_MODE MNIC
+}
+
+whiptail_management_bond_nics() {
+
+	[ -n "$TESTING" ] && return
+
+	MBNICS=()
+	filter_unused_nics
+
+	MBNICS=$(whiptail --title "$whiptail_title" --checklist "Please add NICs to the Management Interface:" 20 75 12 "${nic_list[@]}" 3>&1 1>&2 2>&3)
+	local exitstatus=$?
+	whiptail_check_exitstatus $exitstatus
+
+	while [ -z "$MBNICS" ]
+	do
+		MBNICS=$(whiptail --title "$whiptail_title" --checklist "Please add NICs to the Management Interface:" 20 75 12 "${nic_list[@]}" 3>&1 1>&2 2>&3)
+		local exitstatus=$?
+		whiptail_check_exitstatus $exitstatus
+	done
+
+	MBNICS=$(echo "$MBNICS" | tr -d '"')
+
+	IFS=' ' read -ra MBNICS <<< "$MBNICS"
+
+	for bond_nic in "${MBNICS[@]}"; do
+		for dev_status in "${nmcli_dev_status_list[@]}"; do
+			if [[ $dev_status == "${bond_nic}:unmanaged" ]]; then
+				whiptail \
+					--title "$whiptail_title" \
+					--msgbox "$bond_nic is unmanaged by Network Manager. Please remove it from other network management tools then re-run setup." \
+					8 75
+				exit
+			fi
+		done
+	done
+
+	export MBNICS
 }
 
 whiptail_net_method() {