mirror of
https://github.com/Security-Onion-Solutions/securityonion.git
synced 2026-08-30 19:29:19 +02:00
Merge pull request #16194 from Security-Onion-Solutions/fix/auto-state-apply-local-salt-files
Detect hand-placed local/salt files in Auto State Apply
This commit is contained in:
Josh Patterson
@@ -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/<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.
|
||||
# (e.g. salt/beacon/<minion>/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
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
"""
|
||||
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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 {}
|
||||
@@ -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 {}
|
||||
@@ -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 {}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user