mirror of
https://github.com/Security-Onion-Solutions/securityonion.git
synced 2026-07-18 23:03:39 +02:00
Josh Patterson
9e7e6edae0
Add 100%-coverage unit tests for the three custom salt beacons (postgres_pillar_beacon, rules_beacon, zeek) and add salt/_beacons to the python-test workflow's paths trigger and matrix. To pass the workflow's flake8 lint over the whole directory: - zeek.py: reindent to 4 spaces, drop trailing blank line, noqa the Salt-injected __salt__ references (F821); no logic change. - postgres_pillar_beacon.py: noqa C901 on beacon() (complexity 13 > 12).
173 lines
7.5 KiB
Python
173 lines
7.5 KiB
Python
# 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()
|