diff --git a/salt/_beacons/rules_beacon.py b/salt/_beacons/local_files_beacon.py similarity index 65% rename from salt/_beacons/rules_beacon.py rename to salt/_beacons/local_files_beacon.py index 3bde11c30..7d619d01e 100644 --- a/salt/_beacons/rules_beacon.py +++ b/salt/_beacons/local_files_beacon.py @@ -3,13 +3,14 @@ # 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). +# Custom salt beacon that watches hand-edited directories under +# /opt/so/saltstack/local/salt/ 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 @@ -19,9 +20,10 @@ # up on the next one). # # Each emitted event carries the watched directory path under the configured tag -# (e.g. salt/beacon//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. +# (e.g. salt/beacon//local_files_beacon/zeek); the push_files reactor +# looks the tag up in salt/reactor/pillar_push_map.yaml and writes a push intent, +# after which the existing so-push-drainer / orch.push_batch pipeline takes over +# unchanged. import hashlib import logging @@ -77,7 +79,9 @@ def _fingerprint(directory): h = hashlib.sha1() if os.path.isdir(directory): entries = [] - for root, _dirs, files in os.walk(directory): + for root, dirs, files in os.walk(directory): + # zkg packages are git clones; .git churn would fire a state apply on its own. + dirs[:] = [d for d in dirs if d != '.git'] for name in files: full = os.path.join(root, name) if _excluded(full): @@ -94,20 +98,22 @@ def _fingerprint(directory): return h.hexdigest() -def _watermark_file(tag): - return os.path.join(WATERMARK_DIR, 'rules_beacon_%s.hash' % tag) +def _watermark_file(tag, directory): + # Keyed by directory: zeek/policy and zeek/zkg share the tag `zeek`. + scope = hashlib.sha1(directory.encode('utf-8', 'surrogateescape')).hexdigest()[:12] + return os.path.join(WATERMARK_DIR, 'local_files_beacon_%s_%s.hash' % (tag, scope)) -def _read_watermark(tag): +def _read_watermark(tag, directory): try: - with open(_watermark_file(tag), 'r') as f: + with open(_watermark_file(tag, directory), 'r') as f: return (f.read() or '').strip() or None except IOError: return None -def _write_watermark(tag, digest): - path = _watermark_file(tag) +def _write_watermark(tag, directory, digest): + path = _watermark_file(tag, directory) try: os.makedirs(WATERMARK_DIR, exist_ok=True) tmp = path + '.tmp' @@ -115,7 +121,7 @@ def _write_watermark(tag, digest): f.write(digest) os.rename(tmp, path) except OSError: - log.exception('rules_beacon: failed to persist watermark to %s', path) + log.exception('local_files_beacon: failed to persist watermark to %s', path) def beacon(config): @@ -123,17 +129,17 @@ def beacon(config): for directory, tag in _paths_from_config(config).items(): digest = _fingerprint(directory) - previous = _read_watermark(tag) + previous = _read_watermark(tag, directory) # 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) + _write_watermark(tag, directory, digest) continue if digest != previous: - _write_watermark(tag, digest) + _write_watermark(tag, directory, digest) retval.append({'tag': tag, 'path': directory}) - log.info('rules_beacon: change detected in %s, emitting %s', directory, tag) + log.info('local_files_beacon: change detected in %s, emitting %s', directory, tag) return retval diff --git a/salt/_beacons/local_files_beacon_test.py b/salt/_beacons/local_files_beacon_test.py new file mode 100644 index 000000000..89fffd487 --- /dev/null +++ b/salt/_beacons/local_files_beacon_test.py @@ -0,0 +1,221 @@ +# 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 local_files_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(local_files_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(local_files_beacon.__virtual__()) + + def test_validate_returns_valid(self): + self.assertEqual(local_files_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( + local_files_beacon._paths_from_config(config), + {'/a': 'suricata', '/b': 'strelka'}, + ) + + def test_paths_from_config_plain_dict(self): + self.assertEqual( + local_files_beacon._paths_from_config({'paths': {'/a': 'suricata'}}), + {'/a': 'suricata'}, + ) + + def test_paths_from_config_skips_non_dict_items(self): + self.assertEqual(local_files_beacon._paths_from_config(['bogus', 42]), {}) + + def test_paths_from_config_paths_not_a_dict(self): + self.assertEqual(local_files_beacon._paths_from_config({'paths': 'nope'}), {}) + + def test_paths_from_config_unexpected_type(self): + self.assertEqual(local_files_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(local_files_beacon._excluded(pathname), pathname) + + def test_excluded_allows_real_rule_files(self): + self.assertFalse(local_files_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(local_files_beacon._fingerprint(missing), hashlib.sha1().hexdigest()) + + def test_fingerprint_changes_when_content_changes(self): + d = self._make_dir('rules', {'a.rules': 'alert'}) + before = local_files_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(local_files_beacon._fingerprint(d), before) + + def test_fingerprint_ignores_excluded_files(self): + d = self._make_dir('rules', {'a.rules': 'alert'}) + before = local_files_beacon._fingerprint(d) + with open(os.path.join(d, 'a.rules.swp'), 'w') as f: + f.write('editor swap') + self.assertEqual(local_files_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 = local_files_beacon._fingerprint(d) + os.symlink(os.path.join(d, 'missing-target'), os.path.join(d, 'broken.link')) + self.assertEqual(local_files_beacon._fingerprint(d), good) + + def test_fingerprint_prunes_git_metadata(self): + # zkg packages are git clones, so the watched tree carries .git. + d = self._make_dir('zkg', {'pkg.zeek': 'print 1;'}) + before = local_files_beacon._fingerprint(d) + git_dir = os.path.join(d, 'pkg', '.git', 'refs', 'heads') + os.makedirs(git_dir) + with open(os.path.join(git_dir, 'main'), 'w') as f: + f.write('0' * 40) + self.assertEqual(local_files_beacon._fingerprint(d), before) + + def test_fingerprint_still_sees_worktree_next_to_git(self): + d = self._make_dir('zkg', {'pkg.zeek': 'print 1;'}) + os.makedirs(os.path.join(d, 'pkg', '.git')) + before = local_files_beacon._fingerprint(d) + with open(os.path.join(d, 'pkg', 'scripts.zeek'), 'w') as f: + f.write('print 2;') + self.assertNotEqual(local_files_beacon._fingerprint(d), before) + + # -- _read_watermark / _write_watermark ------------------------------- + + def test_watermark_round_trip(self): + local_files_beacon._write_watermark('suricata', '/rules/suricata', 'deadbeef') + self.assertEqual( + local_files_beacon._read_watermark('suricata', '/rules/suricata'), 'deadbeef') + + def test_read_watermark_missing_returns_none(self): + self.assertIsNone(local_files_beacon._read_watermark('suricata', '/rules/suricata')) + + def test_read_watermark_empty_file_returns_none(self): + os.makedirs(self.state, exist_ok=True) + with open(local_files_beacon._watermark_file('suricata', '/rules/suricata'), 'w') as f: + f.write('') + self.assertIsNone(local_files_beacon._read_watermark('suricata', '/rules/suricata')) + + def test_write_watermark_swallows_oserror(self): + with patch.object(local_files_beacon.os, 'makedirs', side_effect=OSError): + local_files_beacon._write_watermark('suricata', '/rules/suricata', 'deadbeef') + self.assertIsNone(local_files_beacon._read_watermark('suricata', '/rules/suricata')) + + def test_watermark_file_differs_per_directory_within_one_tag(self): + # zeek/policy and zeek/zkg share the tag 'zeek'. + self.assertNotEqual( + local_files_beacon._watermark_file('zeek', '/local/zeek/policy'), + local_files_beacon._watermark_file('zeek', '/local/zeek/zkg'), + ) + + def test_watermarks_are_independent_within_one_tag(self): + local_files_beacon._write_watermark('zeek', '/local/zeek/policy', 'policyhash') + local_files_beacon._write_watermark('zeek', '/local/zeek/zkg', 'zkghash') + self.assertEqual( + local_files_beacon._read_watermark('zeek', '/local/zeek/policy'), 'policyhash') + self.assertEqual( + local_files_beacon._read_watermark('zeek', '/local/zeek/zkg'), 'zkghash') + + # -- beacon ----------------------------------------------------------- + + def _config(self, mapping): + return [{'paths': mapping}] + + def test_beacon_seeds_first_run_and_emits_nothing(self): + with patch.object(local_files_beacon, '_fingerprint', return_value='hash1'), \ + patch.object(local_files_beacon, '_read_watermark', return_value=None), \ + patch.object(local_files_beacon, '_write_watermark') as mock_write: + result = local_files_beacon.beacon(self._config({'/rules/suricata': 'suricata'})) + self.assertEqual(result, []) + mock_write.assert_called_once_with('suricata', '/rules/suricata', 'hash1') + + def test_beacon_emits_on_change(self): + with patch.object(local_files_beacon, '_fingerprint', return_value='newhash'), \ + patch.object(local_files_beacon, '_read_watermark', return_value='oldhash'), \ + patch.object(local_files_beacon, '_write_watermark') as mock_write: + result = local_files_beacon.beacon(self._config({'/rules/suricata': 'suricata'})) + self.assertEqual(result, [{'tag': 'suricata', 'path': '/rules/suricata'}]) + mock_write.assert_called_once_with('suricata', '/rules/suricata', 'newhash') + + def test_beacon_no_change_emits_nothing(self): + with patch.object(local_files_beacon, '_fingerprint', return_value='samehash'), \ + patch.object(local_files_beacon, '_read_watermark', return_value='samehash'), \ + patch.object(local_files_beacon, '_write_watermark') as mock_write: + result = local_files_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(local_files_beacon.beacon(config), []) # seed pass + self.assertEqual(local_files_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(local_files_beacon.beacon(config), [{'tag': 'suricata', 'path': d}]) + + def test_beacon_two_dirs_one_tag_do_not_flap(self): + # Tag-keyed watermarks would clobber each other and emit on every pass. + policy = self._make_dir('zeek/policy', {'intel.dat': '#fields\tindicator'}) + zkg = self._make_dir('zeek/zkg', {'README': 'place packages here'}) + config = self._config({policy: 'zeek', zkg: 'zeek'}) + + self.assertEqual(local_files_beacon.beacon(config), []) # seed pass + self.assertEqual(local_files_beacon.beacon(config), []) # idle + self.assertEqual(local_files_beacon.beacon(config), []) # still idle + + with open(os.path.join(policy, 'intel.dat'), 'a') as f: + f.write('\nevil.com\tIntel::DOMAIN\tsource\n') + self.assertEqual(local_files_beacon.beacon(config), [{'tag': 'zeek', 'path': policy}]) + self.assertEqual(local_files_beacon.beacon(config), []) # quiet again + + +if __name__ == '__main__': + unittest.main() diff --git a/salt/_beacons/rules_beacon_test.py b/salt/_beacons/rules_beacon_test.py deleted file mode 100644 index adcd19181..000000000 --- a/salt/_beacons/rules_beacon_test.py +++ /dev/null @@ -1,172 +0,0 @@ -# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one -# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at -# https://securityonion.net/license; you may not use this file except in compliance with the -# Elastic License 2.0. - -import 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() diff --git a/salt/manager/files/beacons_pushstate.conf.jinja b/salt/manager/files/beacons_pushstate.conf.jinja index fba53b759..8e68b51e0 100644 --- a/salt/manager/files/beacons_pushstate.conf.jinja +++ b/salt/manager/files/beacons_pushstate.conf.jinja @@ -3,9 +3,16 @@ beacons: postgres_pillar_beacon: - interval: {{ AUTOAPPLY.drain_interval }} - disable_during_state_run: False - rules_beacon: + local_files_beacon: - interval: {{ AUTOAPPLY.drain_interval }} - disable_during_state_run: False + # Tags are app names in salt/reactor/pillar_push_map.yaml. + # Allowlist on purpose: salt writes elsewhere under local/salt/ and would self-retrigger. - paths: /opt/so/saltstack/local/salt/suricata/rules: suricata /opt/so/saltstack/local/salt/strelka/rules/compiled: strelka + /opt/so/saltstack/local/salt/zeek/policy: zeek + /opt/so/saltstack/local/salt/zeek/zkg: zeek + /opt/so/saltstack/local/salt/elasticsearch/files/ingest: elasticsearch + /opt/so/saltstack/local/salt/elasticsearch/roles: elasticsearch + /opt/so/saltstack/local/salt/logstash/pipelines/config/custom: logstash diff --git a/salt/manager/tools/sbin/so-push-drainer b/salt/manager/tools/sbin/so-push-drainer index 8cc97a8d5..66fa67e2c 100644 --- a/salt/manager/tools/sbin/so-push-drainer +++ b/salt/manager/tools/sbin/so-push-drainer @@ -20,7 +20,7 @@ is older than debounce_seconds, this script: 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 +Reactor sls files (push_files, push_pillar) write intents but never dispatch directly """ diff --git a/salt/reactor/pillar_push_map.yaml b/salt/reactor/pillar_push_map.yaml index 993444bd2..8c78d2bb7 100644 --- a/salt/reactor/pillar_push_map.yaml +++ b/salt/reactor/pillar_push_map.yaml @@ -1,3 +1,6 @@ +# Read by push_pillar.sls (SOC config saves) and push_files.sls (local/salt file edits); +# both key on the app name. An app missing here waits for the next scheduled highstate. +# # 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). diff --git a/salt/reactor/push_files.sls b/salt/reactor/push_files.sls new file mode 100644 index 000000000..8f93258b6 --- /dev/null +++ b/salt/reactor/push_files.sls @@ -0,0 +1,131 @@ +#!py + +# Reactor invoked by local_files_beacon when a watched directory under +# /opt/so/saltstack/local/salt/ changes. The beacon tag is an app name in +# pillar_push_map.yaml, so file changes and pillar changes route through the same +# table -- see salt/reactor/push_pillar.sls. +# +# The app comes from the event tag, not the payload: salt's beacon loop pops the +# beacon's 'tag' key off the data and appends it to the event tag instead (see +# salt/beacons/__init__.py). The reactor renderer sets both `tag` and `data` as +# module globals. +# +# 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_files: %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_files: 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_files: 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_files: 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_files: 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_files: push disabled, skipping') + return {} + + event = data.get('data', data) # noqa: F821 -- data provided by reactor + path = event.get('path', '') + app = tag.rsplit('/', 1)[-1].strip() # noqa: F821 -- tag provided by reactor + + if not app: + LOG.debug('push_files: ignoring event with no app segment: tag=%s', tag) # noqa: F821 + return {} + + entry = _load_push_map().get(app) + if not entry: + LOG.warning( + 'push_files: app "%s" is not in pillar_push_map.yaml; change will be ' + 'picked up at the next scheduled highstate (path=%s)', + app, path, + ) + return {} + + _write_intent('files_{}'.format(app), list(entry), path) + LOG.info('push_files: intent updated for %s (path=%s)', app, path) + return {} diff --git a/salt/reactor/push_strelka.sls b/salt/reactor/push_strelka.sls deleted file mode 100644 index 31f66e0a6..000000000 --- a/salt/reactor/push_strelka.sls +++ /dev/null @@ -1,96 +0,0 @@ -#!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 {} diff --git a/salt/reactor/push_suricata.sls b/salt/reactor/push_suricata.sls deleted file mode 100644 index 11fbd1286..000000000 --- a/salt/reactor/push_suricata.sls +++ /dev/null @@ -1,95 +0,0 @@ -#!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 {} diff --git a/salt/salt/files/reactor_pushstate.conf b/salt/salt/files/reactor_pushstate.conf index dc23e68e4..1dc6edf1d 100644 --- a/salt/salt/files/reactor_pushstate.conf +++ b/salt/salt/files/reactor_pushstate.conf @@ -1,7 +1,5 @@ reactor: - - 'salt/beacon/*/rules_beacon/suricata': - - salt://reactor/push_suricata.sls - - 'salt/beacon/*/rules_beacon/strelka': - - salt://reactor/push_strelka.sls + - 'salt/beacon/*/local_files_beacon/*': + - salt://reactor/push_files.sls - 'salt/beacon/*/postgres_pillar_beacon/audit_settings': - salt://reactor/push_pillar.sls