# SPDX-License-Identifier: GPL-3.0-or-later """Tests for baseline reconciliation: fingerprints, store I/O, alert filtering.""" import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from enodia_sentinel import reconcile from enodia_sentinel.alert import Alert, Severity from enodia_sentinel.config import Config def _cfg(tmp: Path) -> Config: cfg = Config() cfg.log_dir = tmp return cfg def _alert(key: str, sig: str = "x") -> Alert: return Alert(Severity.HIGH, sig, key, "detail") class TestAlertIdentity(unittest.TestCase): def test_fim_modified_added_removed_map_to_fim(self): for prefix in ("fim:mod:", "fim:add:", "fim:del:"): kind, target = reconcile.alert_identity(prefix + "/etc/hosts") self.assertEqual((kind, target), ("fim", "/etc/hosts")) def test_pkgfile_key(self): self.assertEqual(reconcile.alert_identity("fim:pkg:/usr/bin/ssh"), ("pkgfile", "/usr/bin/ssh")) def test_listener_key(self): self.assertEqual(reconcile.alert_identity("lis:8080/nginx"), ("listener", "8080/nginx")) def test_suid_key(self): self.assertEqual(reconcile.alert_identity("suid:/usr/bin/sudo"), ("suid", "/usr/bin/sudo")) def test_unknown_key_is_passthrough(self): self.assertEqual(reconcile.alert_identity("rsh:1234"), (None, None)) class TestParseDuration(unittest.TestCase): def test_days_hours_minutes(self): self.assertEqual(reconcile.parse_duration("7d"), timedelta(days=7)) self.assertEqual(reconcile.parse_duration("12h"), timedelta(hours=12)) self.assertEqual(reconcile.parse_duration("30m"), timedelta(minutes=30)) def test_invalid_raises(self): for bad in ("", "7", "d", "7x", "-1d", "1.5h"): with self.assertRaises(ValueError): reconcile.parse_duration(bad) class TestBuildFingerprint(unittest.TestCase): def test_fim_extracts_content_fields_only(self): live = {"mode": 33188, "uid": 0, "gid": 0, "size": 12, "sha256": "abc123"} fp = reconcile.build_fingerprint("fim", "/etc/hosts", live) self.assertEqual(fp, {"sha256": "abc123", "mode": 33188, "uid": 0, "gid": 0}) def test_fim_absent_target_is_all_none(self): fp = reconcile.build_fingerprint("fim", "/gone", None) self.assertEqual(fp, {"sha256": None, "mode": None, "uid": None, "gid": None}) def test_pkgfile_sorts_reasons(self): fp = reconcile.build_fingerprint( "pkgfile", "/usr/bin/ssh", ["size mismatch", "checksum mismatch"]) self.assertEqual(fp, {"reasons": ["checksum mismatch", "size mismatch"]}) def test_listener_and_suid_are_identity(self): self.assertEqual( reconcile.build_fingerprint("listener", "8080/nginx", None), {"key": "8080/nginx"}) self.assertEqual( reconcile.build_fingerprint("suid", "/usr/bin/sudo", None), {"path": "/usr/bin/sudo"}) class TestStoreRoundTrip(unittest.TestCase): def setUp(self): self.d = tempfile.TemporaryDirectory() self.cfg = _cfg(Path(self.d.name)) def tearDown(self): self.d.cleanup() reconcile.ReconcileStore.invalidate_cache() def test_missing_file_is_empty_store(self): store = reconcile.ReconcileStore.load(self.cfg) self.assertEqual(store.list(), []) def test_corrupt_json_fails_closed_to_empty(self): reconcile.ReconcileStore.store_path(self.cfg).write_text("{not json") reconcile.ReconcileStore.invalidate_cache() store = reconcile.ReconcileStore.load(self.cfg) self.assertEqual(store.list(), []) def test_accept_persists_and_reloads(self): store = reconcile.ReconcileStore.load(self.cfg) fp = reconcile.build_fingerprint("fim", "/etc/hosts", {"sha256": "a", "mode": 1, "uid": 0, "gid": 0}) store.accept("fim", "/etc/hosts", fp, "added host entry", "luna") reconcile.ReconcileStore.invalidate_cache() again = reconcile.ReconcileStore.load(self.cfg) recs = again.list() self.assertEqual(len(recs), 1) self.assertEqual(recs[0].target, "/etc/hosts") self.assertEqual(recs[0].reason, "added host entry") self.assertEqual(recs[0].actor, "luna") self.assertEqual(recs[0].status, "ok") def test_accept_existing_updates_returns_true(self): store = reconcile.ReconcileStore.load(self.cfg) fp = {"key": "8080/nginx"} self.assertFalse(store.accept("listener", "8080/nginx", fp, "r1", "luna")) self.assertTrue(store.accept("listener", "8080/nginx", fp, "r2", "luna")) self.assertEqual(len(store.list()), 1) def test_revoke_removes_and_reports_missing(self): store = reconcile.ReconcileStore.load(self.cfg) store.accept("suid", "/usr/bin/sudo", {"path": "/usr/bin/sudo"}, "r", "luna") self.assertTrue(store.revoke("suid", "/usr/bin/sudo")) self.assertFalse(store.revoke("suid", "/usr/bin/sudo")) self.assertEqual(store.list(), []) class TestFilterAlerts(unittest.TestCase): def setUp(self): self.d = tempfile.TemporaryDirectory() self.cfg = _cfg(Path(self.d.name)) self.store = reconcile.ReconcileStore.load(self.cfg) def tearDown(self): self.d.cleanup() reconcile.ReconcileStore.invalidate_cache() def test_unacked_alert_passes_through(self): out = self.store.filter_alerts([_alert("fim:mod:/etc/hosts")], {}) self.assertEqual(len(out), 1) def test_matching_fim_ack_suppresses(self): live = {"sha256": "a", "mode": 33188, "uid": 0, "gid": 0, "size": 9} fp = reconcile.build_fingerprint("fim", "/etc/hosts", live) self.store.accept("fim", "/etc/hosts", fp, "edit", "luna") out = self.store.filter_alerts( [_alert("fim:mod:/etc/hosts")], {"fim": {"/etc/hosts": live}}) self.assertEqual(out, []) def test_changed_fim_passes_and_marks_stale(self): accepted = {"sha256": "a", "mode": 33188, "uid": 0, "gid": 0} self.store.accept("fim", "/etc/hosts", accepted, "edit", "luna") changed = {"sha256": "DIFFERENT", "mode": 33188, "uid": 0, "gid": 0} out = self.store.filter_alerts( [_alert("fim:mod:/etc/hosts")], {"fim": {"/etc/hosts": changed}}) self.assertEqual(len(out), 1) self.assertEqual(self.store.list()[0].status, "stale") def test_accepted_removal_stays_suppressed_when_absent(self): fp = reconcile.build_fingerprint("fim", "/etc/gone", None) self.store.accept("fim", "/etc/gone", fp, "decommissioned", "luna") out = self.store.filter_alerts( [_alert("fim:del:/etc/gone")], {"fim": {}}) self.assertEqual(out, []) def test_expired_ttl_passes_and_marks_stale(self): past = datetime.now(timezone.utc) - timedelta(hours=1) self.store.accept("listener", "8080/nginx", {"key": "8080/nginx"}, "temp", "luna", expires_at=past) out = self.store.filter_alerts([_alert("lis:8080/nginx")], {}) self.assertEqual(len(out), 1) self.assertEqual(self.store.list()[0].status, "stale") def test_listener_ack_suppresses_within_ttl(self): future = datetime.now(timezone.utc) + timedelta(days=1) self.store.accept("listener", "8080/nginx", {"key": "8080/nginx"}, "temp", "luna", expires_at=future) out = self.store.filter_alerts([_alert("lis:8080/nginx")], {}) self.assertEqual(out, []) class TestListAndStale(unittest.TestCase): def setUp(self): self.d = tempfile.TemporaryDirectory() self.cfg = _cfg(Path(self.d.name)) def tearDown(self): self.d.cleanup() reconcile.ReconcileStore.invalidate_cache() def test_list_stale_only_filters(self): store = reconcile.ReconcileStore.load(self.cfg) future = datetime.now(timezone.utc) + timedelta(days=1) past = datetime.now(timezone.utc) - timedelta(days=1) store.accept("listener", "1/a", {"key": "1/a"}, "ok", "luna", expires_at=future) store.accept("listener", "2/b", {"key": "2/b"}, "exp", "luna", expires_at=past) stale = store.list(stale_only=True) self.assertEqual([r.target for r in stale], ["2/b"]) def test_list_persists_stale_transition(self): store = reconcile.ReconcileStore.load(self.cfg) past = datetime.now(timezone.utc) - timedelta(days=1) store.accept("listener", "2/b", {"key": "2/b"}, "exp", "luna", expires_at=past) store.list() # triggers TTL re-evaluation + lazy write-back reconcile.ReconcileStore.invalidate_cache() reloaded = reconcile.ReconcileStore.load(self.cfg) self.assertEqual(reloaded.list()[0].status, "stale") if __name__ == "__main__": unittest.main()