Merge pull request #16071 from Security-Onion-Solutions/saltthangs

auto state apply
2026-07-23 01:05:40 +02:00 · 2026-07-20 08:56:08 -04:00
parent c2075ddafb 4886034fef
commit c4295b4e0a
70 changed files with 2148 additions and 115 deletions
@@ -0,0 +1,142 @@
+# 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.
+
+# Custom salt beacon that watches the SOC audit_settings table in postgres for
+# new settings changes and emits a beacon event per new row. This replaces the
+# inotify watch on /opt/so/saltstack/local/pillar -- instead of monitoring pillar
+# files on disk, we monitor the securityonion.audit_settings table that SOC writes to.
+#
+# Detection is poll-based with a monotonic `id` watermark persisted to
+# WATERMARK_FILE: each pass selects rows with id greater than the last id seen,
+# which makes it self-healing (a missed poll simply catches up on the next one).
+#
+# Each emitted event carries setting_id and node_id; the push_pillar reactor maps
+# setting_id -> app via pillar_push_map.yaml and writes a push intent, after which
+# the existing so-push-drainer / orch.push_batch pipeline takes over unchanged.
+
+import logging
+import os
+import subprocess
+
+log = logging.getLogger(__name__)
+
+WATERMARK_FILE = '/opt/so/state/postgres_pillar_beacon_watch.id'
+CONTAINER = 'so-postgres'
+DATABASE = 'securityonion'
+
+# Unaligned, tuples-only psql output with a field separator that cannot appear in
+# an id/setting_id/node_id, so we can split each row reliably.
+FIELD_SEP = '\x1f'
+
+
+def __virtual__():
+    return True
+
+
+def validate(config):
+    return True, 'valid'
+
+
+def _read_watermark():
+    # Returns the last processed id, or None if the watermark has not been seeded.
+    try:
+        with open(WATERMARK_FILE, 'r') as f:
+            return int((f.read() or '').strip())
+    except (IOError, ValueError):
+        return None
+
+
+def _write_watermark(value):
+    try:
+        os.makedirs(os.path.dirname(WATERMARK_FILE), exist_ok=True)
+        tmp = WATERMARK_FILE + '.tmp'
+        with open(tmp, 'w') as f:
+            f.write(str(int(value)))
+        os.rename(tmp, WATERMARK_FILE)
+    except OSError:
+        log.exception('postgres_pillar_beacon: failed to persist watermark to %s', WATERMARK_FILE)
+
+
+def _query(sql):
+    # Run a query against securityonion inside the so-postgres container over the unix
+    # socket (trust auth, no password). Returns stdout on success, or None on any
+    # failure so the caller can no-op and retry on the next interval.
+    cmd = [
+        'docker', 'exec', CONTAINER,
+        'psql', '-U', 'postgres', '-d', DATABASE,
+        '-tA', '-F', FIELD_SEP, '-c', sql,
+    ]
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+    except subprocess.TimeoutExpired:
+        log.warning('postgres_pillar_beacon: psql timed out')
+        return None
+    except Exception:
+        log.exception('postgres_pillar_beacon: failed to exec psql')
+        return None
+    if result.returncode != 0:
+        log.warning('postgres_pillar_beacon: psql failed (rc=%s): %s',
+                    result.returncode, (result.stderr or '').strip())
+        return None
+    return result.stdout
+
+
+def beacon(config):  # noqa: C901
+    retval = []
+
+    watermark = _read_watermark()
+
+    # First run / missing watermark: seed to the current MAX(id) and emit nothing
+    # so we never replay the entire settings history into a fleetwide push.
+    if watermark is None:
+        seed = _query('SELECT COALESCE(MAX(id), 0) FROM audit_settings;')
+        if seed is None:
+            return retval  # postgres not ready yet; retry next interval
+        try:
+            _write_watermark(int((seed or '0').strip() or 0))
+        except ValueError:
+            log.warning('postgres_pillar_beacon: could not parse MAX(id) seed: %r', seed)
+        return retval
+
+    rows = _query(
+        "SELECT id, setting_id, COALESCE(node_id, '') FROM audit_settings "
+        "WHERE id > %d ORDER BY id;" % watermark
+    )
+    if rows is None:
+        return retval
+
+    max_id = watermark
+    for line in rows.splitlines():
+        # Do NOT str.strip() the whole line: Python treats the \x1f field
+        # separator (and \x1c-\x1e) as whitespace, so stripping would eat an
+        # empty trailing node_id field and make the row look malformed.
+        if not line.strip():
+            continue
+        parts = line.split(FIELD_SEP)
+        if len(parts) < 3:
+            log.warning('postgres_pillar_beacon: skipping malformed row: %r', line)
+            continue
+        try:
+            row_id = int(parts[0])
+        except ValueError:
+            log.warning('postgres_pillar_beacon: skipping row with non-int id: %r', line)
+            continue
+        setting_id = parts[1]
+        node_id = parts[2]
+        retval.append({
+            'tag': 'audit_settings',
+            'id': row_id,
+            'setting_id': setting_id,
+            'node_id': node_id,
+        })
+        if row_id > max_id:
+            max_id = row_id
+
+    if max_id > watermark:
+        _write_watermark(max_id)
+        log.info('postgres_pillar_beacon: emitted %d change(s), watermark %d -> %d',
+                 len(retval), watermark, max_id)
+
+    return retval
@@ -0,0 +1,165 @@
+# 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 os
+import shutil
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import postgres_pillar_beacon
+
+
+class TestPostgresPillarBeacon(unittest.TestCase):
+
+    def setUp(self):
+        # Point WATERMARK_FILE at a throwaway dir so the real read/write helpers
+        # (and their os.makedirs/os.rename) run against actual files, then clean
+        # it all up in tearDown.
+        self.tmpdir = tempfile.mkdtemp()
+        self.watermark = os.path.join(self.tmpdir, 'state', 'watch.id')
+        patcher = patch.object(postgres_pillar_beacon, 'WATERMARK_FILE', self.watermark)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    # -- trivial contract -------------------------------------------------
+
+    def test_virtual_returns_true(self):
+        self.assertTrue(postgres_pillar_beacon.__virtual__())
+
+    def test_validate_returns_valid(self):
+        self.assertEqual(postgres_pillar_beacon.validate({}), (True, 'valid'))
+
+    # -- _read_watermark --------------------------------------------------
+
+    def test_read_watermark_valid(self):
+        postgres_pillar_beacon._write_watermark(42)
+        self.assertEqual(postgres_pillar_beacon._read_watermark(), 42)
+
+    def test_read_watermark_missing_file_returns_none(self):
+        # tmp watermark file was never created
+        self.assertIsNone(postgres_pillar_beacon._read_watermark())
+
+    def test_read_watermark_garbage_returns_none(self):
+        os.makedirs(os.path.dirname(self.watermark), exist_ok=True)
+        with open(self.watermark, 'w') as f:
+            f.write('nope')
+        self.assertIsNone(postgres_pillar_beacon._read_watermark())
+
+    # -- _write_watermark -------------------------------------------------
+
+    def test_write_watermark_round_trip(self):
+        postgres_pillar_beacon._write_watermark(7)
+        with open(self.watermark) as f:
+            self.assertEqual(f.read(), '7')
+
+    def test_write_watermark_swallows_oserror(self):
+        with patch.object(postgres_pillar_beacon.os, 'makedirs', side_effect=OSError):
+            # Must not raise; failure is logged and the beacon retries next pass.
+            postgres_pillar_beacon._write_watermark(5)
+        self.assertFalse(os.path.exists(self.watermark))
+
+    # -- _query -----------------------------------------------------------
+
+    def test_query_success_returns_stdout_and_builds_argv(self):
+        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout='rows', stderr='')
+        with patch.object(postgres_pillar_beacon.subprocess, 'run', return_value=completed) as mock_run:
+            result = postgres_pillar_beacon._query('SELECT 1;')
+        self.assertEqual(result, 'rows')
+        argv = mock_run.call_args[0][0]
+        self.assertEqual(argv[:5], ['docker', 'exec', 'so-postgres', 'psql', '-U'])
+        self.assertIn('SELECT 1;', argv)
+        self.assertFalse(mock_run.call_args[1].get('shell', False))
+
+    def test_query_timeout_returns_none(self):
+        with patch.object(postgres_pillar_beacon.subprocess, 'run',
+                          side_effect=subprocess.TimeoutExpired(cmd='psql', timeout=30)):
+            self.assertIsNone(postgres_pillar_beacon._query('SELECT 1;'))
+
+    def test_query_generic_exception_returns_none(self):
+        with patch.object(postgres_pillar_beacon.subprocess, 'run', side_effect=Exception('boom')):
+            self.assertIsNone(postgres_pillar_beacon._query('SELECT 1;'))
+
+    def test_query_nonzero_returncode_returns_none(self):
+        completed = subprocess.CompletedProcess(args=[], returncode=1, stdout='', stderr='bad')
+        with patch.object(postgres_pillar_beacon.subprocess, 'run', return_value=completed):
+            self.assertIsNone(postgres_pillar_beacon._query('SELECT 1;'))
+
+    # -- beacon: first run / seeding --------------------------------------
+
+    def test_beacon_seeds_when_postgres_not_ready(self):
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=None), \
+             patch.object(postgres_pillar_beacon, '_query', return_value=None), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            self.assertEqual(postgres_pillar_beacon.beacon({}), [])
+        mock_write.assert_not_called()
+
+    def test_beacon_seeds_to_max_id_and_emits_nothing(self):
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=None), \
+             patch.object(postgres_pillar_beacon, '_query', return_value='7\n'), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            self.assertEqual(postgres_pillar_beacon.beacon({}), [])
+        mock_write.assert_called_once_with(7)
+
+    def test_beacon_seed_unparseable_is_swallowed(self):
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=None), \
+             patch.object(postgres_pillar_beacon, '_query', return_value='abc'), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            self.assertEqual(postgres_pillar_beacon.beacon({}), [])
+        mock_write.assert_not_called()
+
+    # -- beacon: steady state ---------------------------------------------
+
+    def test_beacon_query_failure_returns_empty(self):
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=10), \
+             patch.object(postgres_pillar_beacon, '_query', return_value=None), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            self.assertEqual(postgres_pillar_beacon.beacon({}), [])
+        mock_write.assert_not_called()
+
+    def test_beacon_emits_events_and_advances_watermark(self):
+        sep = postgres_pillar_beacon.FIELD_SEP
+        rows = '11%s5%snode1\n12%s6%s\n' % (sep, sep, sep, sep)
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=10), \
+             patch.object(postgres_pillar_beacon, '_query', return_value=rows), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            result = postgres_pillar_beacon.beacon({})
+        self.assertEqual(result, [
+            {'tag': 'audit_settings', 'id': 11, 'setting_id': '5', 'node_id': 'node1'},
+            {'tag': 'audit_settings', 'id': 12, 'setting_id': '6', 'node_id': ''},
+        ])
+        mock_write.assert_called_once_with(12)
+
+    def test_beacon_skips_malformed_blank_and_noninteger_rows(self):
+        sep = postgres_pillar_beacon.FIELD_SEP
+        rows = (
+            '\n'                       # blank line -> skipped
+            '13%s7\n'                  # too few fields -> skipped
+            'abc%s8%snodeX\n'          # non-integer id -> skipped
+            '14%s9%snodeY\n'           # the one good row
+        ) % (sep, sep, sep, sep, sep)
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=10), \
+             patch.object(postgres_pillar_beacon, '_query', return_value=rows), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            result = postgres_pillar_beacon.beacon({})
+        self.assertEqual(result, [
+            {'tag': 'audit_settings', 'id': 14, 'setting_id': '9', 'node_id': 'nodeY'},
+        ])
+        mock_write.assert_called_once_with(14)
+
+    def test_beacon_no_new_rows_does_not_advance_watermark(self):
+        with patch.object(postgres_pillar_beacon, '_read_watermark', return_value=10), \
+             patch.object(postgres_pillar_beacon, '_query', return_value=''), \
+             patch.object(postgres_pillar_beacon, '_write_watermark') as mock_write:
+            self.assertEqual(postgres_pillar_beacon.beacon({}), [])
+        mock_write.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,139 @@
+# 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.
+
+# Custom salt beacon that watches the suricata/strelka rule directories for changes
+# and emits a beacon event per changed directory. This replaces the stock salt
+# `inotify` beacon, which leaks a kernel inotify instance every time the minion
+# rebuilds the beacon loader's __context__ (orphaning the old pyinotify.Notifier
+# without closing it) until fs.inotify.max_user_instances is exhausted and the
+# beacon dies with EMFILE. Polling holds zero inotify instances, so the leak is
+# impossible, and it keeps firing during state runs (no blackout).
+#
+# Detection is poll-based with a per-directory fingerprint persisted to
+# WATERMARK_DIR: each pass walks the directory and hashes every file's
+# (relpath, st_mtime_ns, st_size), which catches content writes, additions,
+# moves, and deletions. A change in the digest emits one event; an unchanged
+# digest emits nothing. This makes it self-healing (a missed poll simply catches
+# up on the next one).
+#
+# Each emitted event carries the watched directory path under the configured tag
+# (e.g. salt/beacon/<minion>/rules_beacon/suricata); the push_suricata / push_strelka
+# reactors write a push intent, after which the existing so-push-drainer /
+# orch.push_batch pipeline takes over unchanged.
+
+import hashlib
+import logging
+import os
+import re
+
+log = logging.getLogger(__name__)
+
+WATERMARK_DIR = '/opt/so/state'
+
+# Temp/editor files that should not trigger a push. Mirrors the exclude regexes
+# the inotify beacon used. Matched against the full pathname.
+EXCLUDES = [
+    re.compile(r'\.sw[a-z]$'),
+    re.compile(r'~$'),
+    re.compile(r'/4913$'),
+    re.compile(r'/\.#'),
+]
+
+
+def __virtual__():
+    return True
+
+
+def validate(config):
+    return True, 'valid'
+
+
+def _paths_from_config(config):
+    # The beacon config arrives as a list of single-key dicts (salt beacon style).
+    # Merge it and return the {dir: tag} mapping under the 'paths' key.
+    merged = {}
+    if isinstance(config, list):
+        for item in config:
+            if isinstance(item, dict):
+                merged.update(item)
+    elif isinstance(config, dict):
+        merged = config
+    paths = merged.get('paths', {})
+    return paths if isinstance(paths, dict) else {}
+
+
+def _excluded(pathname):
+    for pattern in EXCLUDES:
+        if pattern.search(pathname):
+            return True
+    return False
+
+
+def _fingerprint(directory):
+    # Stat-only walk; hash each file's (relpath, mtime_ns, size). Returns a hex
+    # digest, or the digest of an empty tree if the directory does not exist.
+    h = hashlib.sha1()
+    if os.path.isdir(directory):
+        entries = []
+        for root, _dirs, files in os.walk(directory):
+            for name in files:
+                full = os.path.join(root, name)
+                if _excluded(full):
+                    continue
+                try:
+                    st = os.stat(full)
+                except OSError:
+                    continue
+                rel = os.path.relpath(full, directory)
+                entries.append('%s\0%d\0%d' % (rel, st.st_mtime_ns, st.st_size))
+        for line in sorted(entries):
+            h.update(line.encode('utf-8', 'surrogateescape'))
+            h.update(b'\n')
+    return h.hexdigest()
+
+
+def _watermark_file(tag):
+    return os.path.join(WATERMARK_DIR, 'rules_beacon_%s.hash' % tag)
+
+
+def _read_watermark(tag):
+    try:
+        with open(_watermark_file(tag), 'r') as f:
+            return (f.read() or '').strip() or None
+    except IOError:
+        return None
+
+
+def _write_watermark(tag, digest):
+    path = _watermark_file(tag)
+    try:
+        os.makedirs(WATERMARK_DIR, exist_ok=True)
+        tmp = path + '.tmp'
+        with open(tmp, 'w') as f:
+            f.write(digest)
+        os.rename(tmp, path)
+    except OSError:
+        log.exception('rules_beacon: failed to persist watermark to %s', path)
+
+
+def beacon(config):
+    retval = []
+
+    for directory, tag in _paths_from_config(config).items():
+        digest = _fingerprint(directory)
+        previous = _read_watermark(tag)
+
+        # First run / missing watermark: seed the digest and emit nothing so a
+        # fresh host does not fire a spurious fleetwide push.
+        if previous is None:
+            _write_watermark(tag, digest)
+            continue
+
+        if digest != previous:
+            _write_watermark(tag, digest)
+            retval.append({'tag': tag, 'path': directory})
+            log.info('rules_beacon: change detected in %s, emitting %s', directory, tag)
+
+    return retval
@@ -0,0 +1,172 @@
+# 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 hashlib
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import rules_beacon
+
+
+class TestRulesBeacon(unittest.TestCase):
+
+    def setUp(self):
+        # Isolate all on-disk state (watermarks and the dirs we fingerprint) in a
+        # throwaway tree, and point WATERMARK_DIR at it so the real read/write
+        # helpers run against actual files.
+        self.tmpdir = tempfile.mkdtemp()
+        self.state = os.path.join(self.tmpdir, 'state')
+        patcher = patch.object(rules_beacon, 'WATERMARK_DIR', self.state)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def _make_dir(self, name, files=None):
+        path = os.path.join(self.tmpdir, name)
+        os.makedirs(path, exist_ok=True)
+        for fname, content in (files or {}).items():
+            with open(os.path.join(path, fname), 'w') as f:
+                f.write(content)
+        return path
+
+    # -- trivial contract -------------------------------------------------
+
+    def test_virtual_returns_true(self):
+        self.assertTrue(rules_beacon.__virtual__())
+
+    def test_validate_returns_valid(self):
+        self.assertEqual(rules_beacon.validate({}), (True, 'valid'))
+
+    # -- _paths_from_config -----------------------------------------------
+
+    def test_paths_from_config_list_of_dicts(self):
+        config = [{'interval': 10}, {'paths': {'/a': 'suricata', '/b': 'strelka'}}]
+        self.assertEqual(
+            rules_beacon._paths_from_config(config),
+            {'/a': 'suricata', '/b': 'strelka'},
+        )
+
+    def test_paths_from_config_plain_dict(self):
+        self.assertEqual(
+            rules_beacon._paths_from_config({'paths': {'/a': 'suricata'}}),
+            {'/a': 'suricata'},
+        )
+
+    def test_paths_from_config_skips_non_dict_items(self):
+        self.assertEqual(rules_beacon._paths_from_config(['bogus', 42]), {})
+
+    def test_paths_from_config_paths_not_a_dict(self):
+        self.assertEqual(rules_beacon._paths_from_config({'paths': 'nope'}), {})
+
+    def test_paths_from_config_unexpected_type(self):
+        self.assertEqual(rules_beacon._paths_from_config('nonsense'), {})
+
+    # -- _excluded --------------------------------------------------------
+
+    def test_excluded_matches_temp_and_editor_files(self):
+        for pathname in ('/rules/foo.swp', '/rules/foo~', '/rules/4913', '/rules/.#foo'):
+            self.assertTrue(rules_beacon._excluded(pathname), pathname)
+
+    def test_excluded_allows_real_rule_files(self):
+        self.assertFalse(rules_beacon._excluded('/rules/suricata.rules'))
+
+    # -- _fingerprint -----------------------------------------------------
+
+    def test_fingerprint_missing_dir_is_empty_tree_digest(self):
+        missing = os.path.join(self.tmpdir, 'does-not-exist')
+        self.assertEqual(rules_beacon._fingerprint(missing), hashlib.sha1().hexdigest())
+
+    def test_fingerprint_changes_when_content_changes(self):
+        d = self._make_dir('rules', {'a.rules': 'alert'})
+        before = rules_beacon._fingerprint(d)
+        with open(os.path.join(d, 'a.rules'), 'w') as f:
+            f.write('alert tcp any any -> any any')  # different size
+        self.assertNotEqual(rules_beacon._fingerprint(d), before)
+
+    def test_fingerprint_ignores_excluded_files(self):
+        d = self._make_dir('rules', {'a.rules': 'alert'})
+        before = rules_beacon._fingerprint(d)
+        with open(os.path.join(d, 'a.rules.swp'), 'w') as f:
+            f.write('editor swap')
+        self.assertEqual(rules_beacon._fingerprint(d), before)
+
+    def test_fingerprint_skips_unstatable_entries(self):
+        # A dangling symlink appears in os.walk's file list but os.stat raises
+        # OSError, exercising the except-continue path.
+        d = self._make_dir('rules', {'a.rules': 'alert'})
+        good = rules_beacon._fingerprint(d)
+        os.symlink(os.path.join(d, 'missing-target'), os.path.join(d, 'broken.link'))
+        self.assertEqual(rules_beacon._fingerprint(d), good)
+
+    # -- _read_watermark / _write_watermark -------------------------------
+
+    def test_watermark_round_trip(self):
+        rules_beacon._write_watermark('suricata', 'deadbeef')
+        self.assertEqual(rules_beacon._read_watermark('suricata'), 'deadbeef')
+
+    def test_read_watermark_missing_returns_none(self):
+        self.assertIsNone(rules_beacon._read_watermark('suricata'))
+
+    def test_read_watermark_empty_file_returns_none(self):
+        os.makedirs(self.state, exist_ok=True)
+        with open(rules_beacon._watermark_file('suricata'), 'w') as f:
+            f.write('')
+        self.assertIsNone(rules_beacon._read_watermark('suricata'))
+
+    def test_write_watermark_swallows_oserror(self):
+        with patch.object(rules_beacon.os, 'makedirs', side_effect=OSError):
+            rules_beacon._write_watermark('suricata', 'deadbeef')
+        self.assertIsNone(rules_beacon._read_watermark('suricata'))
+
+    # -- beacon -----------------------------------------------------------
+
+    def _config(self, mapping):
+        return [{'paths': mapping}]
+
+    def test_beacon_seeds_first_run_and_emits_nothing(self):
+        with patch.object(rules_beacon, '_fingerprint', return_value='hash1'), \
+             patch.object(rules_beacon, '_read_watermark', return_value=None), \
+             patch.object(rules_beacon, '_write_watermark') as mock_write:
+            result = rules_beacon.beacon(self._config({'/rules/suricata': 'suricata'}))
+        self.assertEqual(result, [])
+        mock_write.assert_called_once_with('suricata', 'hash1')
+
+    def test_beacon_emits_on_change(self):
+        with patch.object(rules_beacon, '_fingerprint', return_value='newhash'), \
+             patch.object(rules_beacon, '_read_watermark', return_value='oldhash'), \
+             patch.object(rules_beacon, '_write_watermark') as mock_write:
+            result = rules_beacon.beacon(self._config({'/rules/suricata': 'suricata'}))
+        self.assertEqual(result, [{'tag': 'suricata', 'path': '/rules/suricata'}])
+        mock_write.assert_called_once_with('suricata', 'newhash')
+
+    def test_beacon_no_change_emits_nothing(self):
+        with patch.object(rules_beacon, '_fingerprint', return_value='samehash'), \
+             patch.object(rules_beacon, '_read_watermark', return_value='samehash'), \
+             patch.object(rules_beacon, '_write_watermark') as mock_write:
+            result = rules_beacon.beacon(self._config({'/rules/suricata': 'suricata'}))
+        self.assertEqual(result, [])
+        mock_write.assert_not_called()
+
+    def test_beacon_end_to_end_with_real_files(self):
+        # Exercise the full stack (real fingerprint + real watermark files) across
+        # two poll passes: first seeds silently, second fires after a write.
+        d = self._make_dir('rules', {'a.rules': 'alert'})
+        config = self._config({d: 'suricata'})
+
+        self.assertEqual(rules_beacon.beacon(config), [])  # seed pass
+        self.assertEqual(rules_beacon.beacon(config), [])  # unchanged pass
+
+        with open(os.path.join(d, 'b.rules'), 'w') as f:
+            f.write('alert tcp any any -> any any')
+        self.assertEqual(rules_beacon.beacon(config), [{'tag': 'suricata', 'path': d}])
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -3,31 +3,30 @@ import logging

 def status():

- cmd = "runuser -l zeek -c '/opt/zeek/bin/zeekctl status'"
- retval = __salt__['docker.run']('so-zeek', cmd)
- logging.info('zeekctl_module: zeekctl.status retval: %s' % retval)
+    cmd = "runuser -l zeek -c '/opt/zeek/bin/zeekctl status'"
+    retval = __salt__['docker.run']('so-zeek', cmd)  # noqa: F821
+    logging.info('zeekctl_module: zeekctl.status retval: %s' % retval)

- return retval
+    return retval


 def beacon(config):

-  retval = []
+    retval = []

-  is_enabled = __salt__['healthcheck.is_enabled']()
-  logging.info('zeek_beacon: healthcheck_is_enabled: %s' % is_enabled)
+    is_enabled = __salt__['healthcheck.is_enabled']()  # noqa: F821
+    logging.info('zeek_beacon: healthcheck_is_enabled: %s' % is_enabled)

-  if is_enabled:
-    zeekstatus = status().lower().split(' ')
-    logging.info('zeek_beacon: zeekctl.status: %s' % str(zeekstatus))
-    if 'stopped' in zeekstatus or 'crashed' in zeekstatus or 'error' in zeekstatus or 'error:' in zeekstatus:
-     zeek_restart = True
-    else:
-     zeek_restart = False
+    if is_enabled:
+        zeekstatus = status().lower().split(' ')
+        logging.info('zeek_beacon: zeekctl.status: %s' % str(zeekstatus))
+        if 'stopped' in zeekstatus or 'crashed' in zeekstatus or 'error' in zeekstatus or 'error:' in zeekstatus:
+            zeek_restart = True
+        else:
+            zeek_restart = False

-    __salt__['telegraf.send']('healthcheck zeek_restart=%s' % str(zeek_restart))
-    retval.append({'zeek_restart': zeek_restart})
-    logging.info('zeek_beacon: retval: %s' % str(retval))
-
-  return retval
+        __salt__['telegraf.send']('healthcheck zeek_restart=%s' % str(zeek_restart))  # noqa: F821
+        retval.append({'zeek_restart': zeek_restart})
+        logging.info('zeek_beacon: retval: %s' % str(retval))

+    return retval
@@ -0,0 +1,59 @@
+# 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 unittest
+from unittest.mock import MagicMock
+
+import zeek
+
+ZEEKCTL_CMD = "runuser -l zeek -c '/opt/zeek/bin/zeekctl status'"
+
+
+class TestZeekBeacon(unittest.TestCase):
+
+    def setUp(self):
+        # zeek.py relies on the __salt__ dunder that Salt injects at load time.
+        # Nothing defines it under test, so we attach a dict of mock loader
+        # functions to the module and remove it again afterwards.
+        self.salt = {
+            'docker.run': MagicMock(return_value='Zeek is running'),
+            'healthcheck.is_enabled': MagicMock(return_value=True),
+            'telegraf.send': MagicMock(),
+        }
+        zeek.__salt__ = self.salt
+        self.addCleanup(lambda: delattr(zeek, '__salt__'))
+
+    # -- status -----------------------------------------------------------
+
+    def test_status_runs_zeekctl_and_returns_output(self):
+        self.salt['docker.run'].return_value = 'Zeek is running'
+        result = zeek.status()
+        self.assertEqual(result, 'Zeek is running')
+        self.salt['docker.run'].assert_called_once_with('so-zeek', ZEEKCTL_CMD)
+
+    # -- beacon -----------------------------------------------------------
+
+    def test_beacon_disabled_returns_empty_and_skips_telegraf(self):
+        self.salt['healthcheck.is_enabled'].return_value = False
+        self.assertEqual(zeek.beacon({}), [])
+        self.salt['telegraf.send'].assert_not_called()
+
+    def test_beacon_running_reports_no_restart(self):
+        self.salt['docker.run'].return_value = 'Zeek is running'
+        self.assertEqual(zeek.beacon({}), [{'zeek_restart': False}])
+        self.salt['telegraf.send'].assert_called_once_with('healthcheck zeek_restart=False')
+
+    def test_beacon_unhealthy_status_triggers_restart(self):
+        # Each of these status tokens should flag a restart (the or-chain in beacon).
+        for status_text in ('Zeek is stopped', 'Zeek crashed', 'Zeek error state', 'Zeek error:'):
+            with self.subTest(status=status_text):
+                self.salt['docker.run'].return_value = status_text
+                self.salt['telegraf.send'].reset_mock()
+                self.assertEqual(zeek.beacon({}), [{'zeek_restart': True}])
+                self.salt['telegraf.send'].assert_called_once_with('healthcheck zeek_restart=True')
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -602,42 +602,6 @@ run_check_net_err() {
 	fi
 }

-wait_for_salt_minion() {
-    local minion="$1"
-    local max_wait="${2:-30}"
-    local interval="${3:-2}"
-    local logfile="${4:-'/dev/stdout'}"
-    local elapsed=0
-
-	echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - Waiting for salt-minion '$minion' to be ready..."
-
-    while [ $elapsed -lt $max_wait ]; do
-        # Check if service is running
-		echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - Check if salt-minion service is running"
-        if ! systemctl is-active --quiet salt-minion; then
-            echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - salt-minion service not running (elapsed: ${elapsed}s)"
-            sleep $interval
-            elapsed=$((elapsed + interval))
-            continue
-        fi
-		echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - salt-minion service is running"
-
-        # Check if minion responds to ping
-		echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - Check if $minion responds to ping"
-        if salt "$minion" test.ping --timeout=3 --out=json 2>> "$logfile" | grep -q "true"; then
-            echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - salt-minion '$minion' is connected and ready!"
-            return 0
-        fi
-        
-        echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - Waiting... (${elapsed}s / ${max_wait}s)"
-        sleep $interval
-        elapsed=$((elapsed + interval))
-    done
-
-    echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - ERROR: salt-minion '$minion' not ready after $max_wait seconds"
-    return 1
-}
-
 salt_minion_count() {
 	local MINIONDIR="/opt/so/saltstack/local/pillar/minions"
 	MINIONCOUNT=$(ls -la $MINIONDIR/*.sls | grep -v adv_ | wc -l)
@@ -702,7 +666,7 @@ systemctl_func() {

 	echo ""
 	echo "${echo_action^}ing $service_name service at $(date +"%T.%6N")"
-    systemctl $action $service_name && echo "Successfully ${echo_action}ed $service_name." || echo "Failed to $action $service_name."
+    systemctl $action $service_name && echo "Successfully ${echo_action}ed $service_name at $(date +"%T.%6N")." || echo "Failed to $action $service_name at $(date +"%T.%6N")."
 	echo ""
 }

@@ -74,13 +74,13 @@ def output(options, console, code, data):
    summary = { "status_code": code, "containers": data }
    print(json.dumps(summary))
  elif "-q" not in options:
-    if code == 2:
-      console.print(" [bold yellow]:hourglass: [bold white]System appears to be starting. No highstate has completed since the system was restarted.")
-    elif code == 99:
+    if code == 99:
      console.print(" [bold red]:exclamation: [bold white]Installation does not appear to be complete. A highstate has not fully completed.")
    elif code == 100:
      console.print(" [bold red]:exclamation: [bold white]Installation encountered errors.")
    else:
+      if code == 2:
+        console.print(" [bold yellow]:hourglass: [bold white]System appears to be starting. No highstate has completed since the system was restarted. Container status is shown below.")
      table = Table(title = "Security Onion Status", show_edge = False, safe_box = True, box = box.MINIMAL)
      table.add_column("Container", justify="right", style="white", no_wrap=True)
      table.add_column("Status", justify="left", style="green", no_wrap=True)
@@ -154,8 +154,14 @@ def check_status(options, console):
  code = check_installation_status(options, console)
  if code == 0:
    code = check_system_status(options, console)
-  if code == 0:
-    code, container_list = check_container_status(options, console)
+  # Containers now start on boot without a highstate, so gather/display their
+  # status even when the system is still "starting" (code 2). Keep the starting
+  # code as the exit/status_code so SOC keeps showing the "restarting" message
+  # on the Grid until a highstate completes.
+  if code == 0 or code == 2:
+    container_code, container_list = check_container_status(options, console)
+    if code == 0:
+      code = container_code
  output(options, console, code, container_list)
  return code

@@ -180,4 +186,3 @@ def main():

 if __name__ == "__main__":
  main()
-
@@ -1,5 +1,3 @@
-{% import_yaml 'salt/minion.defaults.yaml' as SALT_MINION_DEFAULTS -%}
-
 #!/bin/bash
 #
 # Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
@@ -7,7 +5,7 @@
 # https://securityonion.net/license; you may not use this file except in compliance with the
 # Elastic License 2.0.

-
+{% from 'salt/schedule.map.jinja' import SCHEDULEMERGED %}

 # this script checks the time the file /opt/so/log/salt/state-apply-test was last modified and restarts the salt-minion service if it is outside a threshold date/time
 # the file is modified via file.touch using a scheduled job healthcheck.salt-minion.state-apply-test that runs a state.apply. 
@@ -20,12 +18,14 @@

 QUIET=false
 UPTIME_REQ=1800 #in seconds, how long the box has to be up before considering restarting salt-minion due to /opt/so/log/salt/state-apply-test not being touched
+HIGHSTATE_UPTIME_REQ=900 #in seconds; if the box has been up this long and no highstate has completed since boot, force one
 CURRENT_TIME=$(date +%s)
 SYSTEM_START_TIME=$(date -d "$(</proc/uptime awk '{print $1}') seconds ago" +%s)
 LAST_HIGHSTATE_END=$([ -e "/opt/so/log/salt/lasthighstate" ] && date -r /opt/so/log/salt/lasthighstate +%s || echo 0)
 LAST_HEALTHCHECK_STATE_APPLY=$([ -e "/opt/so/log/salt/state-apply-test" ] && date -r /opt/so/log/salt/state-apply-test +%s || echo 0)
 # SETTING THRESHOLD TO ANYTHING UNDER 600 seconds may cause a lot of salt-minion restarts since the job to touch the file occurs every 5-8 minutes by default
-THRESHOLD={{SALT_MINION_DEFAULTS.salt.minion.check_threshold}} #within how many seconds the file /opt/so/log/salt/state-apply-test must have been touched/modified before the salt minion is restarted
+# THRESHOLD is derived from the salt schedule highstate interval + 1 hour, so the minion-check grace period tracks the schedule automatically.
+THRESHOLD=$(( ({{ SCHEDULEMERGED.highstate_interval_hours }} + 1) * 3600 )) #within how many seconds the file /opt/so/log/salt/state-apply-test must have been touched/modified before the salt minion is restarted
 THRESHOLD_DATE=$((LAST_HEALTHCHECK_STATE_APPLY+THRESHOLD))

 logCmd() {
@@ -77,24 +77,50 @@ done

 log "running so-salt-minion-check"

+RESTARTED=false
+
+# Check 1 (minion-restart-check): if the minion has stopped applying states (the
+# state-apply-test healthcheck file has gone stale), restart the salt-minion service.
 if [ $CURRENT_TIME -ge $((SYSTEM_START_TIME+$UPTIME_REQ))  ]; then
  if [ $THRESHOLD_DATE -le $CURRENT_TIME ]; then
-    log "salt-minion is unable to apply states" E
-    log "/opt/so/log/salt/healthcheck-state-apply not touched by required date: `date -d @$THRESHOLD_DATE`, last touched: `date -d @$LAST_HEALTHCHECK_STATE_APPLY`" I
-    log "last highstate completed at `date -d @$LAST_HIGHSTATE_END`" I
-    log "checking if any jobs are running" I
+    log "[minion-restart-check] salt-minion is unable to apply states; restarting salt-minion" E
+    log "[minion-restart-check] state-apply-test not touched by required date `date -d @$THRESHOLD_DATE`, last touched `date -d @$LAST_HEALTHCHECK_STATE_APPLY`" I
+    log "[minion-restart-check] last highstate completed at `date -d @$LAST_HIGHSTATE_END`" I
+    log "[minion-restart-check] checking if any jobs are running" I
    logCmd "salt-call --local saltutil.running" I
-    log "ensure salt.minion-state-apply-test is enabled" I
+    log "[minion-restart-check] ensure salt.minion-state-apply-test is enabled" I
    logCmd "salt-call state.enable salt.minion-state-apply-test" I
-    log "ensure highstate is enabled" I
+    log "[minion-restart-check] ensure highstate is enabled" I
    logCmd "salt-call state.enable highstate" I
-    log "killing all salt-minion processes" I
+    log "[minion-restart-check] killing all salt-minion processes" I
    logCmd "pkill -9 -ef /usr/bin/salt-minion" I
-    log "starting salt-minion service" I
+    log "[minion-restart-check] starting salt-minion service" I
    logCmd "systemctl start salt-minion" I
+    log "[minion-restart-check] waiting for salt-minion to become ready, then applying highstate in the background (queued)" I
+    nohup bash -c '/usr/sbin/so-salt-minion-wait; salt-call state.highstate queue=True' >> "/opt/so/log/salt/so-salt-minion-check" 2>&1 &
+    RESTARTED=true
  else
-    log "/opt/so/log/salt/healthcheck-state-apply last touched: `date -d @$LAST_HEALTHCHECK_STATE_APPLY` must be touched by `date -d @$THRESHOLD_DATE` to avoid salt-minion restart" I
+    log "[minion-restart-check] healthy: state-apply-test last touched `date -d @$LAST_HEALTHCHECK_STATE_APPLY`, must go stale past `date -d @$THRESHOLD_DATE` to trigger a salt-minion restart" I
  fi
 else
-  log "system uptime only $((CURRENT_TIME-SYSTEM_START_TIME)) seconds does not meet $UPTIME_REQ second requirement." I
+  log "[minion-restart-check] skipped: system uptime $((CURRENT_TIME-SYSTEM_START_TIME))s is below the ${UPTIME_REQ}s minimum required before a salt-minion restart" I
+fi
+
+# Check 2 (boot-highstate-check): if the host has been up long enough but no highstate
+# has completed since this boot, force one. This recovers a host whose boot highstate
+# (so-boot-highstate.service) failed or was skipped, even while the minion is otherwise
+# healthy (touching state-apply-test). We deliberately do NOT enable highstate here: if
+# soup has disabled it during an upgrade, Salt will refuse the highstate and we avoid
+# forcing one mid-upgrade.
+if $RESTARTED; then
+  log "[boot-highstate-check] skipped: minion-restart-check already queued a highstate this run" I
+elif [ $CURRENT_TIME -lt $((SYSTEM_START_TIME+HIGHSTATE_UPTIME_REQ)) ]; then
+  log "[boot-highstate-check] skipped: system uptime $((CURRENT_TIME-SYSTEM_START_TIME))s is below the ${HIGHSTATE_UPTIME_REQ}s minimum required before forcing a highstate" I
+elif [ $LAST_HIGHSTATE_END -ge $SYSTEM_START_TIME ]; then
+  log "[boot-highstate-check] healthy: a highstate completed at `date -d @$LAST_HIGHSTATE_END`, after this boot at `date -d @$SYSTEM_START_TIME`" I
+elif salt-call --local saltutil.running 2>/dev/null | grep -q 'state.highstate'; then
+  log "[boot-highstate-check] no highstate has completed since boot, but one is already running; skipping" I
+else
+  log "[boot-highstate-check] no highstate has completed since boot after $((CURRENT_TIME-SYSTEM_START_TIME))s uptime; applying highstate" E
+  nohup bash -c 'salt-call state.highstate -l info queue=True' >> "/opt/so/log/salt/so-salt-minion-check" 2>&1 &
 fi
@@ -9,7 +9,8 @@
 prune_images:
  cmd.run:
    - name: so-docker-prune
-    - order: last
+    - onlyif: command -v /usr/sbin/so-docker-prune >/dev/null 2>&1
+    - order: 9000

 {% else %}

@@ -19,6 +19,7 @@ wait_for_elasticsearch:
 so-elastalert:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastalert:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: elastalert
    - name: so-elastalert
    - user: so-elastalert
@@ -15,6 +15,7 @@ include:
 so-elastic-fleet-package-registry:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-fleet-package-registry:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-elastic-fleet-package-registry
    - hostname: Fleet-package-reg-{{ GLOBALS.hostname }}
    - detach: True
@@ -16,6 +16,7 @@ include:
 so-elastic-agent:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-elastic-agent
    - hostname: {{ GLOBALS.hostname }}
    - detach: True
@@ -46,6 +46,7 @@ elasticagent_syncartifacts:
 so-elastic-fleet:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elastic-agent:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-elastic-fleet
    - hostname: FleetServer-{{ GLOBALS.hostname }}
    - detach: True
@@ -24,6 +24,7 @@ include:
 so-elasticsearch:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-elasticsearch:{{ ELASTICSEARCHMERGED.version }}
+    - restart_policy: unless-stopped
    - hostname: elasticsearch
    - name: so-elasticsearch
    - user: elasticsearch
@@ -1,3 +1,3 @@
 global:
  pcapengine: SURICATA
-  pipeline: REDIS
+  pipeline: REDIS
@@ -22,6 +22,7 @@ include:
 so-hydra:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-hydra:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: hydra
    - name: so-hydra
    - networks:
@@ -58,7 +59,6 @@ so-hydra:
      - {{ ULIMIT.name }}={{ ULIMIT.soft }}:{{ ULIMIT.hard }}
    {%   endfor %}
    {% endif %}
-    - restart_policy: unless-stopped
    - watch:
      - file: hydraconfig
    - require:
@@ -15,6 +15,7 @@ include:
 so-idh:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-idh:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-idh
    - detach: True
    - network_mode: host
@@ -18,6 +18,7 @@ include:
 so-influxdb:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-influxdb:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: influxdb
    - networks:
      - sobridge:
@@ -27,6 +27,7 @@ include:
 so-kafka:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-kafka:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: so-kafka
    - name: so-kafka
    - networks:
@@ -17,6 +17,7 @@ include:
 so-kibana:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-kibana:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: kibana
    - user: "932:0"
    - networks:
@@ -15,6 +15,7 @@ include:
 so-kratos:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-kratos:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: kratos
    - name: so-kratos
    - networks:
@@ -51,7 +52,6 @@ so-kratos:
      - {{ ULIMIT.name }}={{ ULIMIT.soft }}:{{ ULIMIT.hard }}
    {%   endfor %}
    {% endif %}
-    - restart_policy: unless-stopped
    - watch:
      - file: kratosschema
      - file: kratosconfig
@@ -0,0 +1,67 @@
+# This state is designed to run on a development manager running in a libvirt VM. It will map the default pillar and salt directories
+# from /opt/so/saltstack/default to your local development machine as the source path.
+# The VM requires a filesystem to be added. Only the source path should be changed to your development codebase
+#  Driver: virtio-9p
+#  Source path: ~/project/securityonion
+#  Target path: saltDev
+
+# If you want a directory to be RW, then kvm must have group privileges.
+# ll /home/user/projects/securityonion/salt/hypervisor
+# total 48
+# drwxrwxr-x  3 user kvm 4096 Feb 13 11:18 ./
+# drwxrwxr-x 64 user user 4096 Feb 13 10:32 ../
+# -rw-rw-r--  1 user kvm 2238 Feb 12 15:06 defaults.yaml
+# -rw-rw-r--  1 user kvm 1467 Feb 12 15:06 init.sls
+# -rw-rw-r--  1 user kvm   70 Feb 13 09:37 soc_hypervisor.yaml
+# drwxrwxr-x  3 user kvm 4096 Feb 12 15:06 tools/
+
+# Ensure required kernel modules are configured for loading
+/etc/modules-load.d/virtio-9p.conf:
+  file.managed:
+    - contents: |
+        9pnet_virtio
+        9pnet
+        9p
+    - mode: 644
+    - user: root
+    - group: root
+
+# Load the kernel modules immediately (in the correct order)
+load_9p_modules:
+  cmd.run:
+    - names:
+      - modprobe 9pnet_virtio
+      - modprobe 9pnet
+      - modprobe 9p
+    - unless: lsmod | grep -E '9pnet_virtio|9pnet|9p'
+
+# Ensure mount point exists
+/opt/so/saltstack/default:
+  file.directory:
+    - user: root
+    - group: root
+    - mode: 755
+    - makedirs: True
+
+# Configure fstab entry using mount.fstab_present
+# Configure fstab entry using mount.fstab_present
+saltdev_fstab:
+  mount.fstab_present:
+    - name: saltDev
+    - fs_file: /opt/so/saltstack/default
+    - fs_vfstype: 9p
+    - fs_mntops: _netdev,trans=virtio,version=9p2000.L
+    - fs_freq: 0
+    - fs_passno: 0
+
+# Mount the filesystem if not already mounted
+mount_saltdev:
+  mount.mounted:
+    - name: /opt/so/saltstack/default
+    - device: saltDev
+    - fstype: 9p
+    - opts: _netdev,trans=virtio,version=9p2000.L
+    - require:
+      - file: /opt/so/saltstack/default
+      - mount: saltdev_fstab
+      - cmd: load_9p_modules
@@ -28,6 +28,7 @@ include:
 so-logstash:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-logstash:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: so-logstash
    - name: so-logstash
    - networks:
@@ -0,0 +1,21 @@
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+{% from 'salt/auto_apply.map.jinja' import AUTOAPPLY %}
+
+include:
+  - salt.minion
+
+{% if GLOBALS.is_manager and AUTOAPPLY.enabled %}
+salt_beacons_pushstate:
+  file.managed:
+    - name: /etc/salt/minion.d/beacons_pushstate.conf
+    - source: salt://manager/files/beacons_pushstate.conf.jinja
+    - template: jinja
+    - watch_in:
+      - service: salt_minion_service
+{% else %}
+salt_beacons_pushstate:
+  file.absent:
+    - name: /etc/salt/minion.d/beacons_pushstate.conf
+    - watch_in:
+      - service: salt_minion_service
+{% endif %}
@@ -0,0 +1,11 @@
+{% from 'salt/auto_apply.map.jinja' import AUTOAPPLY %}
+beacons:
+  postgres_pillar_beacon:
+    - interval: {{ AUTOAPPLY.drain_interval }}
+    - disable_during_state_run: False
+  rules_beacon:
+    - interval: {{ AUTOAPPLY.drain_interval }}
+    - disable_during_state_run: False
+    - paths:
+        /opt/so/saltstack/local/salt/suricata/rules: suricata
+        /opt/so/saltstack/local/salt/strelka/rules/compiled: strelka
@@ -15,6 +15,7 @@ include:
  - manager.elasticsearch
  - manager.kibana
  - manager.managed_soc_annotations
+  - manager.beacons

 repo_log_dir:
  file.directory:
@@ -260,6 +261,7 @@ surifiltersrules:
    - user: 939
    - group: 939

+
 {% else %}

 {{sls}}_state_not_allowed:
@@ -0,0 +1,231 @@
+#!/opt/saltstack/salt/bin/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.
+
+"""
+so-push-drainer
+===============
+
+Scheduled drainer for the active-push feature. Runs on the manager every
+drain_interval seconds (default 15) via a salt schedule in salt/salt/push_drain_schedule.sls.
+
+For each intent file under /opt/so/state/push_pending/*.json whose last_touch
+is older than debounce_seconds, this script:
+  * concatenates the actions lists from every ready intent
+  * dedupes by (state or __highstate__, tgt, tgt_type)
+  * dispatches a single `salt-run state.orchestrate orch.push_batch --async`
+    with the deduped actions list passed as pillar kwargs
+  * deletes the contributed intent files on successful dispatch
+
+Reactor sls files (push_suricata, push_strelka, push_pillar) write intents
+but never dispatch directly
+"""
+
+import fcntl
+import glob
+import json
+import logging
+import logging.handlers
+import os
+import subprocess
+import sys
+import time
+
+import salt.client
+
+PENDING_DIR = '/opt/so/state/push_pending'
+LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
+LOG_FILE = '/opt/so/log/salt/so-push-drainer.log'
+
+HIGHSTATE_SENTINEL = '__highstate__'
+
+
+def _make_logger():
+    logger = logging.getLogger('so-push-drainer')
+    logger.setLevel(logging.INFO)
+    if not logger.handlers:
+        os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
+        handler = logging.handlers.RotatingFileHandler(
+            LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=3,
+        )
+        handler.setFormatter(logging.Formatter(
+            '%(asctime)s | %(levelname)s | %(message)s',
+        ))
+        logger.addHandler(handler)
+    return logger
+
+
+def _load_push_cfg():
+    """Read the salt:auto_apply pillar subtree via salt-call. Returns a dict."""
+    caller = salt.client.Caller()
+    cfg = caller.cmd('pillar.get', 'salt:auto_apply', {})
+    return cfg if isinstance(cfg, dict) else {}
+
+
+def _read_intent(path, log):
+    try:
+        with open(path, 'r') as f:
+            return json.load(f)
+    except (IOError, ValueError) as exc:
+        log.warning('cannot read intent %s: %s', path, exc)
+        return None
+    except Exception:
+        log.exception('unexpected error reading %s', path)
+        return None
+
+
+def _dedupe_actions(actions):
+    seen = set()
+    deduped = []
+    for action in actions:
+        if not isinstance(action, dict):
+            continue
+        state_key = HIGHSTATE_SENTINEL if action.get('highstate') else action.get('state')
+        tgt = action.get('tgt')
+        tgt_type = action.get('tgt_type', 'compound')
+        if not state_key or not tgt:
+            continue
+        key = (state_key, tgt, tgt_type)
+        if key in seen:
+            continue
+        seen.add(key)
+        deduped.append(action)
+    return deduped
+
+
+def _dispatch(actions, log):
+    pillar_arg = json.dumps({'actions': actions})
+    cmd = [
+        'salt-run',
+        'state.orchestrate',
+        'orch.push_batch',
+        'pillar={}'.format(pillar_arg),
+        '--async',
+    ]
+    log.info('dispatching: %s', ' '.join(cmd[:3]) + ' pillar=<{} actions>'.format(len(actions)))
+    try:
+        result = subprocess.run(
+            cmd, check=True, capture_output=True, text=True, timeout=60,
+        )
+    except subprocess.CalledProcessError as exc:
+        log.error('dispatch failed (rc=%s): stdout=%s stderr=%s',
+                  exc.returncode, exc.stdout, exc.stderr)
+        return False
+    except subprocess.TimeoutExpired:
+        log.error('dispatch timed out after 60s')
+        return False
+    except Exception:
+        log.exception('dispatch raised')
+        return False
+    log.info('dispatch accepted: %s', (result.stdout or '').strip())
+    return True
+
+
+def main():
+    log = _make_logger()
+
+    if not os.path.isdir(PENDING_DIR):
+        # Nothing to do; reactors create the dir on first use.
+        return 0
+
+    try:
+        push = _load_push_cfg()
+    except Exception:
+        log.exception('failed to read salt:auto_apply pillar; aborting drain pass')
+        return 1
+
+    if not push.get('enabled', True):
+        log.debug('push disabled; exiting')
+        return 0
+
+    debounce_seconds = int(push.get('debounce_seconds', 30))
+
+    os.makedirs(PENDING_DIR, exist_ok=True)
+    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
+    try:
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+
+        intent_files = [
+            p for p in sorted(glob.glob(os.path.join(PENDING_DIR, '*.json')))
+            if os.path.basename(p) != '.lock'
+        ]
+        if not intent_files:
+            return 0
+
+        now = time.time()
+        ready = []
+        skipped = 0
+        broken = []
+        for path in intent_files:
+            intent = _read_intent(path, log)
+            if not isinstance(intent, dict):
+                broken.append(path)
+                continue
+            last_touch = intent.get('last_touch', 0)
+            if now - last_touch < debounce_seconds:
+                skipped += 1
+                continue
+            ready.append((path, intent))
+
+        for path in broken:
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+        if not ready:
+            if skipped:
+                log.debug('no ready intents (%d still in debounce window)', skipped)
+            return 0
+
+        combined_actions = []
+        oldest_first_touch = now
+        all_paths = []
+        for path, intent in ready:
+            combined_actions.extend(intent.get('actions', []) or [])
+            first = intent.get('first_touch', now)
+            if first < oldest_first_touch:
+                oldest_first_touch = first
+            all_paths.extend(intent.get('paths', []) or [])
+
+        deduped = _dedupe_actions(combined_actions)
+        if not deduped:
+            log.warning('%d intent(s) had no usable actions; clearing', len(ready))
+            for path, _ in ready:
+                try:
+                    os.unlink(path)
+                except OSError:
+                    pass
+            return 0
+
+        debounce_duration = now - oldest_first_touch
+        log.info(
+            'draining %d intent(s): %d action(s) after dedupe (raw=%d), '
+            'debounce_duration=%.1fs, paths=%s',
+            len(ready), len(deduped), len(combined_actions),
+            debounce_duration, all_paths[:20],
+        )
+
+        if not _dispatch(deduped, log):
+            log.warning('dispatch failed; leaving intent files in place for retry')
+            return 1
+
+        for path, _ in ready:
+            try:
+                os.unlink(path)
+            except OSError:
+                log.exception('failed to remove drained intent %s', path)
+
+        return 0
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+
+if __name__ == '__main__':
+    sys.exit(main())
@@ -762,6 +762,21 @@ ensure_postgres_local_pillar() {
  chown -R socore:socore "$dir"
 }

+ensure_salt_local_pillar() {
+  # The salt.auto_apply settings are a new SOC settings
+  # module, so the new pillar/top.sls references salt.soc_salt / salt.adv_salt
+  # unconditionally. Managers upgrading from before this change have no
+  # /opt/so/saltstack/local/pillar/salt/ (make_some_dirs only runs at install
+  # time), so the stubs must be created here before salt-master restarts against
+  # the new top.sls.
+  echo "Ensuring salt local pillar stubs exist."
+  local dir=/opt/so/saltstack/local/pillar/salt
+  mkdir -p "$dir"
+  [[ -f "$dir/soc_salt.sls" ]] || touch "$dir/soc_salt.sls"
+  [[ -f "$dir/adv_salt.sls" ]] || touch "$dir/adv_salt.sls"
+  chown -R socore:socore "$dir"
+}
+
 ensure_postgres_secret() {
  # On a fresh install, generate_passwords + secrets_pillar seed
  # secrets:postgres_pass in /opt/so/saltstack/local/pillar/secrets.sls. That
@@ -958,6 +973,8 @@ update_kafka_metadata() {
 }

 up_to_3.2.0() {
+  ensure_salt_local_pillar
+
  # download 9.3.7 elastic agent packages
  determine_elastic_agent_upgrade

@@ -1546,18 +1563,13 @@ verify_es_version_compatibility() {
 }

 wait_for_salt_minion_with_restart() {
-    local minion="$1"
-    local max_wait="${2:-60}"
-    local interval="${3:-3}"
-    local logfile="$4"
-    
-    wait_for_salt_minion "$minion" "$max_wait" "$interval" "$logfile"
+    /usr/sbin/so-salt-minion-wait
    local result=$?
    
    if [[ $result -ne 0 ]]; then
        echo "$(date '+%a %d %b %Y %H:%M:%S.%6N') - salt-minion not ready, attempting restart..."
        systemctl_func "restart" "salt-minion"
-        wait_for_salt_minion "$minion" "$max_wait" "$interval" "$logfile"
+        /usr/sbin/so-salt-minion-wait
        result=$?
    fi
    
@@ -2011,9 +2023,9 @@ main() {
    enable_highstate

    echo ""
-    echo "Running a highstate. This could take several minutes."
+    echo "Running a highstate at $(date +"%T.%6N"). This could take several minutes."
    set +e
-    wait_for_salt_minion_with_restart "$MINIONID" "60" "3" "$SOUP_LOG" || fail "Salt minion was not running or ready."
+    wait_for_salt_minion_with_restart || fail "Salt minion was not running or ready."
    highstate
    set -e

@@ -2025,8 +2037,8 @@ main() {

    check_saltmaster_status

-    echo "Running a highstate to complete the Security Onion upgrade on this manager. This could take several minutes."
-    wait_for_salt_minion_with_restart "$MINIONID" "60" "3" "$SOUP_LOG" || fail "Salt minion was not running or ready."
+    echo "Running a highstate at $(date +"%T.%6N") to complete the Security Onion upgrade on this manager. This could take several minutes."
+    wait_for_salt_minion_with_restart || fail "Salt minion was not running or ready."

    # Stop long-running scripts to allow potentially updated scripts to load on the next execution.
    if pgrep salt-relay.sh > /dev/null 2>&1; then
@@ -34,6 +34,7 @@ make-rule-dir-nginx:
 so-nginx:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-nginx:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: so-nginx
    - networks:
      - sobridge:
@@ -0,0 +1,37 @@
+{% from 'salt/auto_apply.map.jinja' import AUTOAPPLY %}
+{% set actions = salt['pillar.get']('actions', []) %}
+{% set BATCH = AUTOAPPLY.batch %}
+{% set BATCH_WAIT = AUTOAPPLY.batch_wait %}
+
+{% for action in actions %}
+{%   if action.get('highstate') %}
+apply_highstate_{{ loop.index }}:
+  salt.state:
+    - tgt: '{{ action.tgt }}'
+    - tgt_type: {{ action.get('tgt_type', 'compound') }}
+    - highstate: True
+    - batch: {{ action.get('batch', BATCH) }}
+    - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
+    - kwarg:
+        queue: 2
+{%   else %}
+refresh_pillar_{{ loop.index }}:
+  salt.function:
+    - name: saltutil.refresh_pillar
+    - tgt: '{{ action.tgt }}'
+    - tgt_type: {{ action.get('tgt_type', 'compound') }}
+
+apply_{{ action.state | replace('.', '_') }}_{{ loop.index }}:
+  salt.state:
+    - tgt: '{{ action.tgt }}'
+    - tgt_type: {{ action.get('tgt_type', 'compound') }}
+    - sls:
+      - {{ action.state }}
+    - batch: {{ action.get('batch', BATCH) }}
+    - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
+    - kwarg:
+        queue: 2
+    - require:
+      - salt: refresh_pillar_{{ loop.index }}
+{%   endif %}
+{% endfor %}
@@ -19,6 +19,7 @@ include:
 so-postgres:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-postgres:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: so-postgres
    - networks:
      - sobridge:
@@ -0,0 +1,251 @@
+# One pillar directory can map to multiple (state, tgt) actions.
+# tgt is a raw salt compound expression. tgt_type is always "compound".
+# Per-action `batch` / `batch_wait` override the orch defaults (25% / 15s).
+# An action with `highstate: True` triggers state.highstate instead of
+# state.apply -- see salt/orch/push_batch.sls.
+#
+# Notes:
+#   - `bpf` is a pillar-only dir (no state of its own) consumed by both
+#     zeek and suricata via macros, so a bpf pillar change re-applies both.
+#   - suricata/strelka/zeek/elasticsearch/redis/kafka/logstash etc. have
+#     their own pillar dirs AND their own state, so they map 1:1 (or 1:2
+#     in strelka's case, because of the split init.sls / manager.sls).
+#
+# Intentional omissions (these will log a "not in pillar_push_map.yaml"
+# warning in push_pillar.sls and wait for the next scheduled highstate):
+#   - `data` and `node_data`: pillar-only data consumed by many states;
+#     handling them generically would amount to a fleetwide highstate.
+#   - `host`: soc_host describes mainint/mainip; a change is a re-IP and
+#     needs a coordinated procedure, not an immediate state push.
+#   - `hypervisor`: state changes touch libvirt and are disruptive; leave
+#     to the next scheduled highstate.
+#   - `sensor`: every field in soc_sensor.yaml is `readonly: True` or
+#     per-minion (`node: True`). Per-minion edits are persisted under
+#     pillar/minions/<id>.sls and are handled by Branch A of push_pillar.sls
+#     (per-minion highstate intent), not by this app-pillar map.
+#
+# The role sets here were verified line-by-line against salt/top.sls. If
+# salt/top.sls changes how an app is targeted, update the corresponding
+# compound here.
+
+# firewall: the one pillar everyone touches. Applied everywhere intentionally
+# because every host's iptables needs to know about every other host in the
+# grid. Salt's firewall state is idempotent (file.managed + iptables-restore
+# onchanges in salt/firewall/init.sls), so hosts whose rendered firewall is
+# unchanged do a file comparison and no-op without touching iptables -- actual
+# reload happens only on the hosts whose rules actually changed. Fleetwide
+# blast radius is intentional and matches the pre-plan behavior via highstate.
+# Adding N sensors in a burst coalesces into one dispatch via the drainer.
+firewall:
+  - state: firewall
+    tgt: '*'
+
+# backup: backup.config_backup runs on eval, standalone, manager, managerhype,
+# managersearch (NOT import -- the backup pillar is included on import per
+# pillar/top.sls but the backup state is not run there per salt/top.sls).
+backup:
+  - state: backup.config_backup
+    tgt: 'G@role:so-eval or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# bpf is pillar-only (no state); consumed by both zeek and suricata as macros.
+# Both states run on sensor_roles + so-import per salt/top.sls.
+bpf:
+  - state: zeek
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-import or G@role:so-sensor or G@role:so-standalone'
+  - state: suricata
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-import or G@role:so-sensor or G@role:so-standalone'
+
+# ca is applied universally.
+ca:
+  - state: ca
+    tgt: '*'
+
+# docker: universal. The docker state is in both the all-non-managers and
+# all-managers branches of salt/top.sls.
+docker:
+  - state: docker
+    tgt: '*'
+
+# elastalert: eval, standalone, manager, managerhype, managersearch (NOT import).
+elastalert:
+  - state: elastalert
+    tgt: 'G@role:so-eval or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# elastic-fleet-package-registry: manager_roles exactly.
+elastic-fleet-package-registry:
+  - state: elastic-fleet-package-registry
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# elasticsearch: 8 roles.
+elasticsearch:
+  - state: elasticsearch
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-searchnode or G@role:so-standalone'
+
+# elasticagent: so-heavynode only.
+elasticagent:
+  - state: elasticagent
+    tgt: 'G@role:so-heavynode'
+
+# elasticfleet: base state only on pillar change. elasticfleet.install_agent_grid
+# is a deploy/enrollment step, not a config reload; leave it to the next highstate.
+elasticfleet:
+  - state: elasticfleet
+    tgt: 'G@role:so-eval or G@role:so-fleet or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# global: fanout to a fleetwide highstate. The global pillar (soc_global.sls)
+# carries cross-cutting settings (pipeline, url_base, imagerepo, mdengine, ...)
+# that are consumed by virtually every state, so a targeted re-apply isn't
+# meaningful. The drainer's batch/batch_wait throttling controls blast radius.
+global:
+  - highstate: True
+    tgt: '*'
+
+# healthcheck: eval, sensor, standalone only.
+healthcheck:
+  - state: healthcheck
+    tgt: 'G@role:so-eval or G@role:so-sensor or G@role:so-standalone'
+
+# hydra: manager_roles exactly.
+hydra:
+  - state: hydra
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# idh: so-idh only.
+idh:
+  - state: idh
+    tgt: 'G@role:so-idh'
+
+# influxdb: manager_roles exactly.
+influxdb:
+  - state: influxdb
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# kafka: standalone, manager, managerhype, managersearch, searchnode, receiver.
+kafka:
+  - state: kafka
+    tgt: 'G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-receiver or G@role:so-searchnode or G@role:so-standalone'
+
+# kibana: manager_roles exactly.
+kibana:
+  - state: kibana
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# kratos: manager_roles exactly.
+kratos:
+  - state: kratos
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# logrotate: universal (top-of-file '*' branch in salt/top.sls).
+logrotate:
+  - state: logrotate
+    tgt: '*'
+
+# logstash: 8 roles, no eval/import.
+logstash:
+  - state: logstash
+    tgt: 'G@role:so-fleet or G@role:so-heavynode or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-receiver or G@role:so-searchnode or G@role:so-standalone'
+
+# manager: manager_roles exactly. The manager state is also referenced under
+# *_sensor / *_heavynode top.sls blocks via `sensor`, but the standalone
+# `manager` state itself runs only on manager_roles.
+manager:
+  - state: manager
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# nginx: 10 specific roles. NOT receiver, idh, hypervisor, desktop.
+nginx:
+  - state: nginx
+    tgt: 'G@role:so-eval or G@role:so-fleet or G@role:so-heavynode or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-searchnode or G@role:so-sensor or G@role:so-standalone'
+
+# ntp: universal (top-of-file '*' branch in salt/top.sls).
+ntp:
+  - state: ntp
+    tgt: '*'
+
+# patch: universal. soc_patch carries the OS update schedule, applied via
+# patch.os.schedule on every node (it's in both the all-non-managers and
+# all-managers branches of salt/top.sls).
+patch:
+  - state: patch.os.schedule
+    tgt: '*'
+
+# postgres: manager_roles exactly.
+postgres:
+  - state: postgres
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# redis: 6 roles. standalone, manager, managerhype, managersearch, heavynode, receiver.
+# (NOT eval, NOT import, NOT searchnode.)
+redis:
+  - state: redis
+    tgt: 'G@role:so-heavynode or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-receiver or G@role:so-standalone'
+
+# registry: manager_roles exactly.
+registry:
+  - state: registry
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# salt: fanout to a fleetwide highstate. The salt.auto_apply settings tune the
+# push pipeline itself (enabled, debounce/drain intervals, batch sizing) and
+# salt.schedule sets the per-minion highstate interval; they are consumed by the
+# manager's schedule, beacons, and master reactor config as well as every
+# minion's highstate schedule, so a targeted re-apply isn't meaningful. A salt
+# audit row only fires for SOC-driven salt.auto_apply / salt.schedule edits --
+# salt version bumps go through soup, not SOC, so they never reach this map.
+salt:
+  - highstate: True
+    tgt: '*'
+
+# sensoroni: universal.
+sensoroni:
+  - state: sensoroni
+    tgt: '*'
+
+# soc: manager_roles exactly.
+soc:
+  - state: soc
+    tgt: 'G@role:so-eval or G@role:so-import or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-standalone'
+
+# stig: broad. Runs on standalone, manager, managerhype, managersearch,
+# searchnode, sensor, receiver, fleet, hypervisor, desktop.
+# NOT eval, NOT import, NOT heavynode, NOT idh (the *_idh block in
+# salt/top.sls intentionally omits stig).
+stig:
+  - state: stig
+    tgt: 'G@role:so-desktop or G@role:so-fleet or G@role:so-hypervisor or G@role:so-manager or G@role:so-managerhype or G@role:so-managersearch or G@role:so-receiver or G@role:so-searchnode or G@role:so-sensor or G@role:so-standalone'
+
+# strelka: sensor-side only on pillar change (sensor_roles). strelka.manager is
+# intentionally NOT fired on pillar changes -- YARA rule and strelka config
+# pillar changes are consumed by the sensor-side strelka backend, and re-running
+# strelka.manager on managers is both unnecessary and disruptive. strelka.manager
+# is left to the 2-hour highstate.
+strelka:
+  - state: strelka
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-sensor or G@role:so-standalone'
+
+# suricata: sensor_roles + so-import (5 roles).
+suricata:
+  - state: suricata
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-import or G@role:so-sensor or G@role:so-standalone'
+
+# telegraf: universal.
+telegraf:
+  - state: telegraf
+    tgt: '*'
+
+# versionlock: universal (top-of-file '*' branch in salt/top.sls).
+versionlock:
+  - state: versionlock
+    tgt: '*'
+
+# vm: libvirt-driver hypervisors only. Matched by the salt-cloud:driver:libvirt
+# grain (compound supports nested grain matching via G@<key>:<subkey>:<value>).
+# pillar/vm/soc_vm.sls write path is referenced at salt/_runners/setup_hypervisor.py:856.
+vm:
+  - state: vm
+    tgt: 'G@salt-cloud:driver:libvirt'
+
+# zeek: sensor_roles + so-import (5 roles).
+zeek:
+  - state: zeek
+    tgt: 'G@role:so-eval or G@role:so-heavynode or G@role:so-import or G@role:so-sensor or G@role:so-standalone'
@@ -0,0 +1,176 @@
+#!py
+
+# Reactor invoked by the postgres_pillar_beacon when SOC records settings changes in
+# the securityonion.audit_settings table (see salt/_beacons/postgres_pillar_beacon.py). The beacon
+# emits one event per new row carrying setting_id and node_id.
+#
+# Two branches, keyed on node_id:
+#   A) node_id populated -> the change is scoped to that one minion. Look up the
+#      app in pillar_push_map.yaml and write an intent that runs the app's mapped
+#      state(s) targeted to just that node.
+#   B) node_id empty -> grid-wide app change. Look up the app in
+#      pillar_push_map.yaml and write an intent with the entry's actions as-is.
+#
+# The app name is the first dotted segment of setting_id (e.g. "telegraf.output"
+# -> "telegraf"), which matches the pillar_push_map.yaml keys 1:1.
+#
+# Reactors never dispatch directly. The so-push-drainer schedule picks up
+# ready intents, dedupes across pending files, and dispatches orch.push_batch.
+
+import fcntl
+import json
+import logging
+import os
+import time
+
+from salt.client import Caller
+import yaml
+
+LOG = logging.getLogger(__name__)
+
+PENDING_DIR = '/opt/so/state/push_pending'
+LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
+MAX_PATHS = 20
+
+# The pillar_push_map.yaml is shipped via salt:// but the reactor runs on the
+# master, which mounts the default saltstack tree at this path.
+PUSH_MAP_PATH = '/opt/so/saltstack/default/salt/reactor/pillar_push_map.yaml'
+
+_PUSH_MAP_CACHE = {'mtime': 0, 'data': None}
+
+
+def _load_push_map():
+    try:
+        st = os.stat(PUSH_MAP_PATH)
+    except OSError:
+        LOG.warning('push_pillar: %s not found', PUSH_MAP_PATH)
+        return {}
+    if _PUSH_MAP_CACHE['mtime'] != st.st_mtime:
+        try:
+            with open(PUSH_MAP_PATH, 'r') as f:
+                _PUSH_MAP_CACHE['data'] = yaml.safe_load(f) or {}
+        except Exception:
+            LOG.exception('push_pillar: failed to load %s', PUSH_MAP_PATH)
+            _PUSH_MAP_CACHE['data'] = {}
+        _PUSH_MAP_CACHE['mtime'] = st.st_mtime
+    return _PUSH_MAP_CACHE['data'] or {}
+
+
+def _push_enabled():
+    try:
+        caller = Caller()
+        return bool(caller.cmd('pillar.get', 'salt:auto_apply:enabled', True))
+    except Exception:
+        LOG.exception('push_pillar: pillar.get salt:auto_apply:enabled failed, assuming enabled')
+        return True
+
+
+def _write_intent(key, actions, path):
+    now = time.time()
+    try:
+        os.makedirs(PENDING_DIR, exist_ok=True)
+    except OSError:
+        LOG.exception('push_pillar: cannot create %s', PENDING_DIR)
+        return
+
+    intent_path = os.path.join(PENDING_DIR, '{}.json'.format(key))
+    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
+    try:
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+
+        intent = {}
+        if os.path.exists(intent_path):
+            try:
+                with open(intent_path, 'r') as f:
+                    intent = json.load(f)
+            except (IOError, ValueError):
+                intent = {}
+
+        intent.setdefault('first_touch', now)
+        intent['last_touch'] = now
+        intent['actions'] = actions
+        paths = intent.get('paths', [])
+        if path and path not in paths:
+            paths.append(path)
+            paths = paths[-MAX_PATHS:]
+        intent['paths'] = paths
+
+        tmp_path = intent_path + '.tmp'
+        with open(tmp_path, 'w') as f:
+            json.dump(intent, f)
+        os.rename(tmp_path, intent_path)
+    except Exception:
+        LOG.exception('push_pillar: failed to write intent %s', intent_path)
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+
+def _app_from_setting(setting_id):
+    # setting_id is e.g. 'telegraf.output' -> 'telegraf', 'ntp.config.servers' -> 'ntp'
+    if not setting_id:
+        return None
+    return setting_id.split('.', 1)[0] or None
+
+
+def _node_actions(entry, node_id):
+    # Copy the app's mapped actions but retarget each one to the single node.
+    # Preserves the state/highstate selection and any batch/batch_wait overrides.
+    actions = []
+    for action in entry:
+        if not isinstance(action, dict):
+            continue
+        node_action = dict(action)
+        node_action['tgt'] = node_id
+        node_action['tgt_type'] = 'glob'
+        actions.append(node_action)
+    return actions
+
+
+def run():
+    if not _push_enabled():
+        LOG.info('push_pillar: push disabled, skipping')
+        return {}
+
+    # The postgres_pillar_beacon nests its payload under data['data']; fall back to the
+    # top level so the reactor is robust to either shape.
+    event = data.get('data', data)  # noqa: F821 -- data provided by reactor
+    setting_id = event.get('setting_id', '')
+    node_id = (event.get('node_id') or '').strip()
+
+    app = _app_from_setting(setting_id)
+    if not app:
+        LOG.debug('push_pillar: ignoring event with no app segment: setting_id=%s', setting_id)
+        return {}
+
+    push_map = _load_push_map()
+    entry = push_map.get(app)
+    if not entry:
+        LOG.warning(
+            'push_pillar: app "%s" is not in pillar_push_map.yaml; change will be '
+            'picked up at the next scheduled highstate (setting_id=%s)',
+            app, setting_id,
+        )
+        return {}
+
+    # Branch A: per-node change -> retarget the app's states to just that node.
+    if node_id:
+        actions = _node_actions(entry, node_id)
+        if not actions:
+            LOG.warning('push_pillar: no usable actions for app "%s" (setting_id=%s)', app, setting_id)
+            return {}
+        _write_intent(
+            'node_{}_{}'.format(node_id, app), actions,
+            'audit:{}@{}'.format(setting_id, node_id),
+        )
+        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s)',
+                 app, node_id, setting_id)
+        return {}
+
+    # Branch B: grid-wide app change -> use the map entry's actions as-is.
+    actions = list(entry)  # copy to avoid mutating the cache
+    _write_intent('pillar_{}'.format(app), actions, 'audit:{}'.format(setting_id))
+    LOG.info('push_pillar: app intent updated for %s (setting_id=%s)', app, setting_id)
+    return {}
@@ -0,0 +1,96 @@
+#!py
+
+# Reactor invoked by the rules_beacon poll beacon (salt/_beacons/rules_beacon.py) on rule
+# file changes under /opt/so/saltstack/local/salt/strelka/rules/compiled/.
+#
+# Writes (or updates) a push intent at /opt/so/state/push_pending/rules_strelka.json
+# and returns {}. The so-push-drainer schedule picks up ready intents, dedupes
+# across pending files, and dispatches orch.push_batch. Reactors never dispatch
+# directly
+
+import fcntl
+import json
+import logging
+import os
+import time
+
+from salt.client import Caller
+
+LOG = logging.getLogger(__name__)
+
+PENDING_DIR = '/opt/so/state/push_pending'
+LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
+MAX_PATHS = 20
+
+# Mirrors GLOBALS.sensor_roles in salt/vars/globals.map.jinja. Sensor-side
+# strelka runs on exactly these four roles; so-import gets strelka.manager
+# instead, which is not fired on pillar changes.
+SENSOR_ROLES = ['so-eval', 'so-heavynode', 'so-sensor', 'so-standalone']
+
+
+def _sensor_compound():
+    return ' or '.join('G@role:{}'.format(r) for r in SENSOR_ROLES)
+
+
+def _push_enabled():
+    try:
+        caller = Caller()
+        return bool(caller.cmd('pillar.get', 'salt:auto_apply:enabled', True))
+    except Exception:
+        LOG.exception('push_strelka: pillar.get salt:auto_apply:enabled failed, assuming enabled')
+        return True
+
+
+def _write_intent(key, actions, path):
+    now = time.time()
+    try:
+        os.makedirs(PENDING_DIR, exist_ok=True)
+    except OSError:
+        LOG.exception('push_strelka: cannot create %s', PENDING_DIR)
+        return
+
+    intent_path = os.path.join(PENDING_DIR, '{}.json'.format(key))
+    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
+    try:
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+
+        intent = {}
+        if os.path.exists(intent_path):
+            try:
+                with open(intent_path, 'r') as f:
+                    intent = json.load(f)
+            except (IOError, ValueError):
+                intent = {}
+
+        intent.setdefault('first_touch', now)
+        intent['last_touch'] = now
+        intent['actions'] = actions
+        paths = intent.get('paths', [])
+        if path and path not in paths:
+            paths.append(path)
+            paths = paths[-MAX_PATHS:]
+        intent['paths'] = paths
+
+        tmp_path = intent_path + '.tmp'
+        with open(tmp_path, 'w') as f:
+            json.dump(intent, f)
+        os.rename(tmp_path, intent_path)
+    except Exception:
+        LOG.exception('push_strelka: failed to write intent %s', intent_path)
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+
+def run():
+    if not _push_enabled():
+        LOG.info('push_strelka: push disabled, skipping')
+        return {}
+
+    path = data.get('path', '')  # noqa: F821 -- data provided by reactor
+    actions = [{'state': 'strelka', 'tgt': _sensor_compound()}]
+    _write_intent('rules_strelka', actions, path)
+    LOG.info('push_strelka: intent updated for path=%s', path)
+    return {}
@@ -0,0 +1,95 @@
+#!py
+
+# Reactor invoked by the rules_beacon poll beacon (salt/_beacons/rules_beacon.py) on rule
+# file changes under /opt/so/saltstack/local/salt/suricata/rules/.
+#
+# Writes (or updates) a push intent at /opt/so/state/push_pending/rules_suricata.json
+# and returns {}. The so-push-drainer schedule picks up ready intents, dedupes
+# across pending files, and dispatches orch.push_batch. Reactors never dispatch
+# directly
+
+import fcntl
+import json
+import logging
+import os
+import time
+
+from salt.client import Caller
+
+LOG = logging.getLogger(__name__)
+
+PENDING_DIR = '/opt/so/state/push_pending'
+LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
+MAX_PATHS = 20
+
+# Mirrors GLOBALS.sensor_roles in salt/vars/globals.map.jinja. Suricata also
+# runs on so-import per salt/top.sls, so that role is appended below.
+SENSOR_ROLES = ['so-eval', 'so-heavynode', 'so-sensor', 'so-standalone']
+
+
+def _sensor_compound_plus_import():
+    return ' or '.join('G@role:{}'.format(r) for r in SENSOR_ROLES) + ' or G@role:so-import'
+
+
+def _push_enabled():
+    try:
+        caller = Caller()
+        return bool(caller.cmd('pillar.get', 'salt:auto_apply:enabled', True))
+    except Exception:
+        LOG.exception('push_suricata: pillar.get salt:auto_apply:enabled failed, assuming enabled')
+        return True
+
+
+def _write_intent(key, actions, path):
+    now = time.time()
+    try:
+        os.makedirs(PENDING_DIR, exist_ok=True)
+    except OSError:
+        LOG.exception('push_suricata: cannot create %s', PENDING_DIR)
+        return
+
+    intent_path = os.path.join(PENDING_DIR, '{}.json'.format(key))
+    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
+    try:
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+
+        intent = {}
+        if os.path.exists(intent_path):
+            try:
+                with open(intent_path, 'r') as f:
+                    intent = json.load(f)
+            except (IOError, ValueError):
+                intent = {}
+
+        intent.setdefault('first_touch', now)
+        intent['last_touch'] = now
+        intent['actions'] = actions
+        paths = intent.get('paths', [])
+        if path and path not in paths:
+            paths.append(path)
+            paths = paths[-MAX_PATHS:]
+        intent['paths'] = paths
+
+        tmp_path = intent_path + '.tmp'
+        with open(tmp_path, 'w') as f:
+            json.dump(intent, f)
+        os.rename(tmp_path, intent_path)
+    except Exception:
+        LOG.exception('push_suricata: failed to write intent %s', intent_path)
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+
+def run():
+    if not _push_enabled():
+        LOG.info('push_suricata: push disabled, skipping')
+        return {}
+
+    path = data.get('path', '')  # noqa: F821 -- data provided by reactor
+    actions = [{'state': 'suricata', 'tgt': _sensor_compound_plus_import()}]
+    _write_intent('rules_suricata', actions, path)
+    LOG.info('push_suricata: intent updated for path=%s', path)
+    return {}
@@ -17,6 +17,7 @@ include:
 so-redis:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-redis:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: so-redis
    - user: socore
    - networks:
@@ -17,11 +17,13 @@ include:
 so-dockerregistry:
  docker_container.running:
    - image: ghcr.io/security-onion-solutions/registry:3.1.1
+    # Intentionally `always`-- registry is critical and must
+    # come back up even if it was manually stopped.
+    - restart_policy: always
    - hostname: so-registry
    - networks:
      - sobridge:
        - ipv4_address: {{ DOCKERMERGED.containers['so-dockerregistry'].ip }}
-    - restart_policy: always
    - port_bindings:
      {% for BINDING in DOCKERMERGED.containers['so-dockerregistry'].port_bindings %}
      - {{ BINDING }}
@@ -0,0 +1,2 @@
+{% import_yaml 'salt/defaults.yaml' as SALT_DEFAULTS %}
+{% set AUTOAPPLY = salt['pillar.get']('salt:auto_apply', SALT_DEFAULTS.salt.auto_apply, merge=True) %}
@@ -3,7 +3,7 @@
 {% set SCHEDULE = salt['pillar.get']('healthcheck:schedule', 30) %}

 include:
-  - salt
+  - salt.minion

 {% if CHECKS and ENABLED %}
 salt_beacons:
@@ -14,12 +14,13 @@ salt_beacons:
    - defaults:
        CHECKS: {{ CHECKS }}
        SCHEDULE: {{ SCHEDULE }}
-    - watch_in: 
+    - watch_in:
      - service: salt_minion_service
 {% else %}
 salt_beacons:
  file.absent:
    - name: /etc/salt/minion.d/beacons.conf
-    - watch_in: 
+    - watch_in:
      - service: salt_minion_service
 {% endif %}
+
@@ -0,0 +1,9 @@
+salt:
+  auto_apply:
+    enabled: true
+    debounce_seconds: 30
+    drain_interval: 15
+    batch: '25%'
+    batch_wait: 15
+  schedule:
+    highstate_interval_hours: 2
@@ -0,0 +1,7 @@
+reactor:
+  - 'salt/beacon/*/rules_beacon/suricata':
+    - salt://reactor/push_suricata.sls
+  - 'salt/beacon/*/rules_beacon/strelka':
+    - salt://reactor/push_strelka.sls
+  - 'salt/beacon/*/postgres_pillar_beacon/audit_settings':
+    - salt://reactor/push_pillar.sls
@@ -0,0 +1,11 @@
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+{% from 'salt/schedule.map.jinja' import SCHEDULEMERGED %}
+
+highstate_schedule:
+  schedule.present:
+    - function: state.highstate
+    - hours: {{ SCHEDULEMERGED.highstate_interval_hours }}
+    - maxrunning: 1
+{% if not GLOBALS.is_manager %}
+    - splay: 1800
+{% endif %}
@@ -5,3 +5,11 @@ salt_bootstrap:
    - source: salt://salt/scripts/bootstrap-salt.sh
    - mode: 755
    - show_changes: False
+
+salt_sbin:
+  file.recurse:
+    - name: /usr/sbin
+    - source: salt://salt/tools/sbin
+    - user: 939
+    - group: 939
+    - file_mode: 755
@@ -1,4 +1,4 @@
 lasthighstate:
  file.touch:
    - name: /opt/so/log/salt/lasthighstate
-    - order: last
+    - order: 9001
@@ -10,6 +10,7 @@
 #    software that is protected by the license key."

 {% from 'allowed_states.map.jinja' import allowed_states %}
+{% from 'salt/auto_apply.map.jinja' import AUTOAPPLY %}
 {% if sls in allowed_states %}

 include:
@@ -63,6 +64,21 @@ engines_config:
    - name: /etc/salt/master.d/engines.conf
    - source: salt://salt/files/engines.conf

+{% if AUTOAPPLY.enabled %}
+reactor_pushstate_config:
+  file.managed:
+    - name: /etc/salt/master.d/reactor_pushstate.conf
+    - source: salt://salt/files/reactor_pushstate.conf
+    - watch_in:
+      - service: salt_master_service
+{% else %}
+reactor_pushstate_config:
+  file.absent:
+    - name: /etc/salt/master.d/reactor_pushstate.conf
+    - watch_in:
+      - service: salt_master_service
+{% endif %}
+
 # update the bootstrap script when used for salt-cloud
 salt_bootstrap_cloud:
  file.managed:
@@ -78,7 +94,7 @@ salt_master_service:
      - file: checkmine_engine
      - file: pillarWatch_engine
      - file: engines_config
-    - order: last
+    - order: 9002

 {% else %}

@@ -2,4 +2,3 @@
 salt:
  minion:
    version: '3006.19'
-    check_threshold: 3600 # in seconds, threshold used for so-salt-minion-check. any value less than 600 seconds may cause a lot of salt-minion restarts since the job to touch the file occurs every 5-8 minutes by default
@@ -111,13 +111,17 @@ mark_setup_complete_for_upgrades:

 {% endif %}

-# this has to be outside the if statement above since there are <requisite>_in calls to this state
+# this has to be outside the if statement above since there are <requisite>_in calls to this state.
+# uses watch (not listen) so the restart fires in-state and its result lands on this state's
+# running entry; that is what lets wait_for_salt_minion_ready below detect any restart
+# uniformly via onchanges, regardless of whether the trigger came from these files or from
+# external watch_in's (e.g. beacons, master/pyinotify).
 salt_minion_service:
  service.running:
    - name: salt-minion
    - enable: True
    - onlyif: test "{{INSTALLEDSALTVERSION}}" == "{{SALTVERSION}}"
-    - listen:
+    - watch:
      - file: mine_functions
 {% if INSTALLEDSALTVERSION|string == SALTVERSION|string %}
      - file: set_log_levels
@@ -126,3 +130,24 @@ salt_minion_service:
      - file: signing_policy
 {% endif %}
    - order: last
+
+# block until the salt-minion daemon is ready for the current instance, so follow-on jobs and the
+# next highstate iteration do not race the restart. onchanges + require on salt_minion_service
+# catches every restart trigger uniformly because watch mod_watch results replace the service
+# state's running entry. wait logic lives in /usr/sbin/so-salt-minion-wait (deployed by salt_sbin
+# from salt/tools/sbin/); its steady-state authority is the master req/publish sockets for the
+# current daemon pid (resolved via systemd, not the pidfile), and it corroborates a just-restarted
+# instance with the pid-tagged "Minion is ready to receive requests!" log line only within a short
+# window of startup. Because that socket signal does not require a recent restart, the wait also
+# succeeds cleanly when salt_minion_service reports a non-restart change (e.g. an enable toggle)
+# rather than false-timing-out. set_log_levels above enforces the log_level_logfile: info that the
+# ready line depends on. salt restarts this unit with --no-block, so mod_watch returns while the old
+# daemon is still up; the script waits for systemd's restart job to drain before it reads MainPID.
+wait_for_salt_minion_ready:
+  cmd.run:
+    - name: /usr/sbin/so-salt-minion-wait
+    - onchanges:
+      - service: salt_minion_service
+    - require:
+      - service: salt_minion_service
+    - order: last
@@ -0,0 +1,17 @@
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+{% from 'salt/auto_apply.map.jinja' import AUTOAPPLY %}
+
+{% if GLOBALS.is_manager and AUTOAPPLY.enabled %}
+push_drain_schedule:
+  schedule.present:
+    - function: cmd.run
+    - job_args:
+      - /usr/sbin/so-push-drainer
+    - seconds: {{ AUTOAPPLY.drain_interval }}
+    - maxrunning: 1
+    - return_job: False
+{% elif GLOBALS.is_manager %}
+push_drain_schedule:
+  schedule.absent:
+    - name: push_drain_schedule
+{% endif %}
@@ -0,0 +1,2 @@
+{% import_yaml 'salt/defaults.yaml' as SALT_DEFAULTS %}
+{% set SCHEDULEMERGED = salt['pillar.get']('salt:schedule', SALT_DEFAULTS.salt.schedule, merge=True) %}
@@ -0,0 +1,39 @@
+salt:
+  auto_apply:
+    enabled:
+      description: Master kill-switch for the active push feature. When disabled, rule and pillar changes are picked up at the next scheduled highstate instead of being pushed immediately.
+      forcedType: bool
+      helpLink: push
+      global: True
+    debounce_seconds:
+      description: Trailing-edge debounce window in seconds. A push intent must be quiet for this long before the drainer dispatches. Rapid bursts of edits within this window coalesce into one dispatch.
+      forcedType: int
+      helpLink: push
+      global: True
+      advanced: True
+    drain_interval:
+      description: How often the push drainer checks for ready intents, in seconds. Small values lower dispatch latency at the cost of more background work on the manager.
+      forcedType: int
+      helpLink: push
+      global: True
+      advanced: True
+    batch:
+      description: "Host batch size for push orchestrations. A number (e.g. '10') or a percentage (e.g. '25%'). Limits how many minions run the push state at once so large fleets don't thundering-herd."
+      helpLink: push
+      global: True
+      advanced: True
+      regex: '^([0-9]+%?)$'
+      regexFailureMessage: Enter a whole number or a whole-number percentage (e.g. 10 or 25%).
+    batch_wait:
+      description: Seconds to wait between host batches in a push orchestration. Gives the fleet time to breathe between waves.
+      forcedType: int
+      helpLink: push
+      global: True
+      advanced: True
+  schedule:
+    highstate_interval_hours:
+      description: How often every minion in the grid runs a scheduled state.highstate, in hours. Lower values keep minions closer in sync at the cost of more load; higher values reduce load but increase worst-case latency for non-pushed changes. The salt-minion health check restarts a minion if its last highstate is older than this value plus one hour.
+      forcedType: int
+      helpLink: push
+      global: True
+      advanced: True
@@ -0,0 +1,207 @@
+#!/bin/bash
+#
+# 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.
+
+# Block until the current salt-minion daemon is ready to receive requests. Invoked from the
+# wait_for_salt_minion_ready state in salt/minion/init.sls after salt_minion_service fires its
+# watch-driven restart, so follow-on jobs and the next highstate iteration do not race it. It is
+# also correct on an already-running minion (no recent restart): the steady-state readiness signal
+# is the live master sockets, so it does not depend on a restart having just happened.
+#
+# Salt logs "Minion is ready to receive requests!" from Minion.tune_in() only after
+# sync_connect_master() returns, which means the pub channel authenticated, the long-running req
+# channel connected, and _post_master_init() finished loading modules and compiling pillar. Two
+# signals reproduce that:
+#
+#   1. Steady state  the pid holds an ESTABLISHED req connection to a master on 4506 plus a second
+#                    (publish) connection to that same master IP on another port. The publish port is
+#                    learned from the master's auth reply and is absent from minion config, so it is
+#                    derived from the connection rather than read from config. This is the always-on
+#                    authority: it reflects whatever daemon is running now, restart or not.
+#   2. Startup only  the pid-tagged ready line in the minion log. Salt's log_fmt_logfile embeds
+#                    [%(process)d] just before the message, so this is keyed to one daemon instance.
+#                    Salt logs it exactly once per start, so it exists only to close a ~2.8s window
+#                    after the sockets come up where they are established but _post_master_init() is
+#                    still finishing. It is therefore required only within READY_LINE_WINDOW seconds
+#                    of (re)start (by pid uptime); past that the line has scrolled out of the log and
+#                    the socket gate alone decides. See instance_ready().
+#
+# The daemon pid is resolved from systemd, never from /var/run/salt-minion.pid. salt_minion() runs
+# the real minion in a multiprocessing child; that child writes the pidfile, owns the sockets and
+# logs the ready line, while systemd's MainPID is the parent. During a restart the pidfile can still
+# name the OLD child, whose own ready line is already in the log -- matching it would report ready
+# instantly. Children of the current MainPID structurally exclude the old instance.
+#
+# That is only true once systemd has actually swapped MainPID. Salt restarts this unit with
+# --no-block (salt/modules/systemd_service.py:_no_block_default returns True for salt-minion), so
+# service.restart returns as soon as the job is enqueued and the state proceeds to run this script
+# while the OLD daemon is still up -- observed at ~7s before MainPID flips. Reading MainPID in that
+# window names the outgoing instance, which is still fully connected and has its own ready line, so
+# every gate below would pass on the daemon that is about to die. Wait for systemd's job queue for
+# the unit to drain first; that is the deterministic "the swap has happened" signal.
+
+. /usr/sbin/so-common
+
+set -u
+
+TIMEOUT=120
+MASTER_PORT=4506
+LOG_TAIL_LINES=10000
+# Seconds after a (re)start during which the pid-tagged ready line is still required. Past this the
+# daemon is clearly beyond the ~2.8s post-connect race and the socket gate is authoritative -- the
+# one-time ready line has scrolled out of the log tail on a long-running minion. Kept under TIMEOUT
+# so a fresh minion that connects but never logs the line still falls back to socket-only near the
+# end instead of false-timing-out.
+READY_LINE_WINDOW=90
+DEFAULT_LOG_FILE="/opt/so/log/salt/minion"
+LOG_FILE="$DEFAULT_LOG_FILE"
+
+# Decide whether the ready line can ever appear. salt-call --local sets file_client=local, so this
+# reads the merged config (honoring minion.d overrides) without contacting the master. salt defaults
+# log_level_logfile to None, meaning it inherits log_level, so resolve that before deciding.
+LOG_LEVEL_LOGFILE=$(salt-call --local --out=newline_values_only config.get log_level_logfile 2>/dev/null | head -n1)
+case "${LOG_LEVEL_LOGFILE,,}" in
+  ""|none) LOG_LEVEL_LOGFILE=$(salt-call --local --out=newline_values_only config.get log_level 2>/dev/null | head -n1) ;;
+esac
+
+case "${LOG_LEVEL_LOGFILE,,}" in
+  all|garbage|trace|debug|profile|info) USE_LOG_GATE=1 ;;
+  *)                                    USE_LOG_GATE=0 ;;
+esac
+
+if [ "$USE_LOG_GATE" -eq 1 ]; then
+  LOG_FILE=$(salt-call --local --out=newline_values_only config.get log_file 2>/dev/null | head -n1)
+  [ -z "$LOG_FILE" ] && LOG_FILE="$DEFAULT_LOG_FILE"
+  [ -d "$(dirname "$LOG_FILE")" ] || USE_LOG_GATE=0
+fi
+
+if command -v ss >/dev/null 2>&1; then
+  USE_SOCKET_GATE=1
+else
+  USE_SOCKET_GATE=0
+fi
+
+if [ "$USE_LOG_GATE" -eq 0 ] && [ "$USE_SOCKET_GATE" -eq 0 ]; then
+  echo "so-salt-minion-wait: no usable readiness signal (log_level_logfile='${LOG_LEVEL_LOGFILE:-unset}', ss not found)" >&2
+  exit 1
+fi
+
+if [ "$USE_LOG_GATE" -eq 1 ] && [ "$USE_SOCKET_GATE" -eq 1 ]; then
+  echo "so-salt-minion-wait: gating on pid-tagged ready line in ${LOG_FILE} plus master sockets"
+elif [ "$USE_LOG_GATE" -eq 1 ]; then
+  echo "so-salt-minion-wait: ss not found; gating on pid-tagged ready line in ${LOG_FILE} only"
+else
+  echo "so-salt-minion-wait: INFO file logging unavailable (log_level_logfile='${LOG_LEVEL_LOGFILE:-unset}'); gating on master sockets only"
+fi
+
+# True while systemd still has a queued or running job for the unit, i.e. an in-flight --no-block
+# restart. MainPID still names the outgoing daemon until this drains. The unit name is passed as a
+# filter and grepped as well, so this stays correct if an older systemctl ignores the filter.
+restart_pending() {
+  systemctl list-jobs --no-legend salt-minion.service 2>/dev/null | grep -q 'salt-minion\.service'
+}
+
+# Emit the pid(s) of the current daemon instance. systemd's MainPID is the parent keepalive process;
+# its child runs tune_in. Fall back to MainPID when there is no child (--disable-keepalive path).
+resolve_daemon_pids() {
+  local mainpid children
+  mainpid=$(systemctl show -p MainPID --value salt-minion 2>/dev/null)
+  if [ -z "$mainpid" ] || [ "$mainpid" = "0" ]; then
+    return 1
+  fi
+  children=$(pgrep -P "$mainpid" 2>/dev/null)
+  printf '%s\n' "${children:-$mainpid}"
+}
+
+# Elapsed seconds since this pid started (Linux procps etimes). Empty/non-numeric -> failure, so the
+# caller can fall back to the strict (log-gate-enforced) behavior when uptime cannot be read.
+pid_uptime() {
+  local pid=$1 secs
+  secs=$(ps -o etimes= -p "$pid" 2>/dev/null | tr -d ' ')
+  case "$secs" in ''|*[!0-9]*) return 1 ;; esac
+  printf '%s\n' "$secs"
+}
+
+# True iff the ready line tagged with this pid is in the current or most recently rotated log.
+ready_logged() {
+  local pid=$1 f
+  for f in "$LOG_FILE" "$LOG_FILE.1"; do
+    [ -r "$f" ] || continue
+    if tail -n "$LOG_TAIL_LINES" "$f" 2>/dev/null | grep -Fq "[$pid] Minion is ready to receive requests!"; then
+      return 0
+    fi
+  done
+  return 1
+}
+
+# True iff this pid holds an ESTABLISHED req connection to a master on MASTER_PORT and a second
+# ESTABLISHED connection to that same master IP on another port. The trailing comma in "pid=N,"
+# keeps pid=123 from matching pid=1234. Grid comms are IPv4 (the unit's ExecStartPre gates on ip -4).
+socket_ready() {
+  local pid=$1 mip master_ips
+  master_ips=$(ss -tnp state established "dport = :${MASTER_PORT}" 2>/dev/null \
+    | grep -F "pid=${pid}," \
+    | grep -oE "[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:${MASTER_PORT}" \
+    | sed "s/:${MASTER_PORT}\$//" \
+    | sort -u)
+  [ -z "$master_ips" ] && return 1
+  for mip in $master_ips; do
+    if ss -tnp state established "dst ${mip} and dport != :${MASTER_PORT}" 2>/dev/null | grep -qF "pid=${pid},"; then
+      return 0
+    fi
+  done
+  return 1
+}
+
+instance_ready() {
+  local pid=$1 uptime
+  # The log gate only closes the ~2.8s window right after the master sockets come up where they are
+  # established but _post_master_init() is still loading modules/compiling pillar. Salt logs the
+  # pid-tagged ready line exactly once at startup, so on a daemon that started long ago the line has
+  # scrolled out of the log tail and the gate could never pass -- making the wait require a recent
+  # restart. Enforce it only while the daemon is young enough that the race could still be open; past
+  # READY_LINE_WINDOW the socket gate is authoritative. If uptime can't be read, keep the strict
+  # behavior (uptime=0 -> gate enforced) so the fresh-restart path never regresses.
+  if [ "$USE_LOG_GATE" -eq 1 ]; then
+    uptime=$(pid_uptime "$pid") || uptime=0
+    if [ "$uptime" -lt "$READY_LINE_WINDOW" ] && ! ready_logged "$pid"; then
+      return 1
+    fi
+  fi
+  if [ "$USE_SOCKET_GATE" -eq 1 ] && ! socket_ready "$pid"; then
+    return 1
+  fi
+  return 0
+}
+
+elapsed=0
+pids=""
+announced_pending=0
+while [ "$elapsed" -lt "$TIMEOUT" ]; do
+  if restart_pending; then
+    # An in-flight --no-block restart: MainPID still names the outgoing daemon. Evaluating now
+    # would bless the instance that is about to be torn down.
+    if [ "$announced_pending" -eq 0 ]; then
+      echo "so-salt-minion-wait: systemd restart job in flight; waiting for it to drain"
+      announced_pending=1
+    fi
+  elif pids=$(resolve_daemon_pids); then
+    # shellcheck disable=SC2086
+    for pid in $pids; do
+      if instance_ready "$pid"; then
+        echo "salt-minion (pid ${pid}) ready after ${elapsed}s"
+        exit 0
+      fi
+    done
+  fi
+  sleep 1
+  elapsed=$((elapsed + 1))
+done
+
+mainpid=$(systemctl show -p MainPID --value salt-minion 2>/dev/null)
+restart_pending && pending=yes || pending=no
+echo "salt-minion did not become ready within ${TIMEOUT}s (MainPID=${mainpid:-unknown}, candidate pids='${pids:-none}', restart_job_pending=${pending}, log_gate=${USE_LOG_GATE}, socket_gate=${USE_SOCKET_GATE}, log_file=${LOG_FILE})" >&2
+exit 1
@@ -1,10 +0,0 @@
-{%   from 'vars/globals.map.jinja' import GLOBALS %}
-
-highstate_schedule:
-  schedule.present:
-    - function: state.highstate
-    - minutes: 15
-    - maxrunning: 1
-{% if not GLOBALS.is_manager %}
-    - splay: 120
-{% endif %}
@@ -14,6 +14,7 @@ include:
 so-sensoroni:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-soc:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - network_mode: host
    - binds:
      - /nsm/import:/nsm/import:rw
@@ -18,6 +18,7 @@ include:
 so-soc:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-soc:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - hostname: soc
    - name: so-soc
    - networks:
@@ -15,6 +15,7 @@ include:
 strelka_backend:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-strelka-backend:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - binds:
      - /opt/so/conf/strelka/backend/:/etc/strelka/:ro
      - /opt/so/conf/strelka/rules/compiled/:/etc/yara/:ro
@@ -47,7 +48,6 @@ strelka_backend:
      - {{ ULIMIT.name }}={{ ULIMIT.soft }}:{{ ULIMIT.hard }}
    {%   endfor %}
    {% endif %}
-    - restart_policy: on-failure
    - watch:
      - file: strelkasensorcompiledrules
      - file: backend_backend_config
@@ -15,6 +15,7 @@ include:
 strelka_coordinator:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-redis:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-strelka-coordinator
    - networks:
      - sobridge:
@@ -15,6 +15,7 @@ include:
 strelka_filestream:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-strelka-manager:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - binds:
      - /opt/so/conf/strelka/filestream/:/etc/strelka/:ro
      - /nsm/strelka:/nsm/strelka
@@ -15,6 +15,7 @@ include:
 strelka_frontend:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-strelka-manager:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - binds:
      - /opt/so/conf/strelka/frontend/:/etc/strelka/:ro
      - /nsm/strelka/log/:/var/log/strelka/:rw
@@ -15,6 +15,7 @@ include:
 strelka_gatekeeper:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-redis:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-strelka-gatekeeper
    - networks:
      - sobridge:
@@ -15,6 +15,7 @@ include:
 strelka_manager:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-strelka-manager:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - binds:
      - /opt/so/conf/strelka/manager/:/etc/strelka/:ro
      {% if DOCKERMERGED.containers['so-strelka-manager'].custom_bind_mounts %}
@@ -17,6 +17,7 @@ include:
 so-suricata:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-suricata:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - privileged: True
    - environment:
      - INTERFACE={{ GLOBALS.sensor.interface }}
@@ -7,6 +7,7 @@ so-tcpreplay:
  docker_container.running:
    - network_mode: "host"
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-tcpreplay:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - name: so-tcpreplay
    - user: root
    - interactive: True
@@ -18,6 +18,7 @@ include:
 so-telegraf:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-telegraf:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - user: 939
    - group_add: 939,920
    - environment:
@@ -19,7 +19,7 @@ base:
    - repo.client
    - versionlock
    - ntp
-    - schedule
+    - salt.highstate_schedule
    - logrotate

  # manager node on proper salt version with empty node_data pillar
@@ -55,6 +55,7 @@ base:
    - motd
    - salt.minion-check
    - salt.lasthighstate
+    - salt.push_drain_schedule
    - common
    - docker
    - docker_clean
@@ -294,7 +295,6 @@ base:
    - nginx
    - elasticfleet
    - elasticfleet.install_agent_grid
-    - schedule
    - stig

  '*_hypervisor and I@features:vrt and G@saltversion:{{saltversion}}':
@@ -16,6 +16,7 @@ include:
 so-zeek:
  docker_container.running:
    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-zeek:{{ GLOBALS.so_version }}
+    - restart_policy: unless-stopped
    - start: True
    - privileged: True
    {% if DOCKERMERGED.containers['so-zeek'].ulimits %}