# SPDX-License-Identifier: GPL-3.0-or-later """Incident-grouping tests: pure lineage/assign cores, the record() index updater, and an end-to-end capture→group→CLI path — all over a temp log_dir, no /proc or daemon needed.""" import io import json import tempfile import unittest from contextlib import redirect_stdout from pathlib import Path from enodia_sentinel import incident from enodia_sentinel.alert import Alert, Severity from enodia_sentinel.cli import main from enodia_sentinel.config import Config def _alert(sig, key, pids=(), sev=Severity.HIGH, sid=1): return Alert(severity=sev, signature=sig, key=key, detail=f"{sig} detail", pids=tuple(pids), sid=sid, classtype="test") class TestLineage(unittest.TestCase): def test_walks_ancestry_excluding_init(self): # 4242 -> 999 (web) -> 1 (init, excluded) parents = {4242: 999, 999: 1} line = incident.lineage_of([4242], parents.get, depth=8) self.assertEqual(line, {4242, 999}) # 1 is excluded def test_depth_cap(self): parents = {5: 4, 4: 3, 3: 2, 2: 1} self.assertEqual(incident.lineage_of([5], parents.get, depth=1), {5, 4}) def test_unknown_parent_stops(self): self.assertEqual(incident.lineage_of([7], lambda p: 0), {7}) def test_empty_pids(self): self.assertEqual(incident.lineage_of([], lambda p: 0), set()) class TestAssign(unittest.TestCase): def _idx(self, **inc): base = {"id": "inc-A", "host": "h", "last_ts": 1000.0, "lineage": []} base.update(inc) return {base["id"]: base} def test_shared_lineage_joins(self): idx = self._idx(lineage=[999, 4242]) self.assertEqual( incident.assign(idx, {5000, 999}, 1100.0, "h", window=1800), "inc-A") def test_disjoint_lineage_starts_new(self): idx = self._idx(lineage=[999, 4242]) self.assertIsNone( incident.assign(idx, {7000, 8000}, 1100.0, "h", window=1800)) def test_closed_window_not_matched(self): idx = self._idx(lineage=[999]) self.assertIsNone( incident.assign(idx, {999}, 5000.0, "h", window=1800)) def test_pidless_joins_most_recent_open(self): idx = { "inc-A": {"id": "inc-A", "host": "h", "last_ts": 1000.0, "lineage": [1]}, "inc-B": {"id": "inc-B", "host": "h", "last_ts": 1500.0, "lineage": [2]}, } # empty lineage (FIM-style) -> most recently active open incident self.assertEqual(incident.assign(idx, set(), 1600.0, "h", 1800), "inc-B") def test_other_host_not_matched(self): idx = self._idx(lineage=[999]) self.assertIsNone(incident.assign(idx, {999}, 1100.0, "other", 1800)) class TestRecord(unittest.TestCase): def setUp(self): self.dir = tempfile.TemporaryDirectory() self.cfg = Config() self.cfg.log_dir = Path(self.dir.name) def tearDown(self): self.dir.cleanup() def test_related_alerts_share_one_incident(self): # batch 1: a shell under web server 999 a1 = incident.record(self.cfg, "alert-1.log", [_alert("reverse_shell", "r:1", [4242])], lineage={4242, 999}, when=1000.0, host="h") # batch 2: a listener opened by the same shell (lineage shares 999) a2 = incident.record(self.cfg, "alert-2.log", [_alert("new_listener", "l:1", [4243])], lineage={4243, 999}, when=1050.0, host="h") self.assertEqual(a1, a2) idx = incident.load_index(self.cfg) self.assertEqual(len(idx), 1) inc = idx[a1] self.assertEqual(inc["alert_count"], 2) self.assertEqual(inc["signatures"], ["reverse_shell", "new_listener"]) self.assertEqual(inc["snapshots"], ["alert-1.log", "alert-2.log"]) def test_unrelated_alerts_make_two_incidents(self): a1 = incident.record(self.cfg, "a1.log", [_alert("reverse_shell", "r:1", [10])], lineage={10, 11}, when=1000.0, host="h") a2 = incident.record(self.cfg, "a2.log", [_alert("reverse_shell", "r:2", [20])], lineage={20, 21}, when=1001.0, host="h") self.assertNotEqual(a1, a2) self.assertEqual(len(incident.load_index(self.cfg)), 2) def test_severity_escalates_to_max(self): iid = incident.record(self.cfg, "a1.log", [_alert("new_listener", "l", [10], Severity.MEDIUM)], lineage={10}, when=1000.0, host="h") incident.record(self.cfg, "a2.log", [_alert("reverse_shell", "r", [10], Severity.CRITICAL)], lineage={10}, when=1010.0, host="h") self.assertEqual(incident.load_index(self.cfg)[iid]["severity"], "CRITICAL") def test_tracking_disabled_returns_none(self): self.cfg.incident_tracking = False self.assertIsNone(incident.record(self.cfg, "a.log", [_alert("x", "k", [1])], lineage={1}, when=1.0, host="h")) class TestCLI(unittest.TestCase): def setUp(self): self.dir = tempfile.TemporaryDirectory() self.tmp = Path(self.dir.name) self.cfg = Config() self.cfg.log_dir = self.tmp # one incident with two member snapshots on disk for name, sig, t in (("alert-1", "reverse_shell", "2026-06-10T10:00:00-07:00"), ("alert-2", "new_listener", "2026-06-10T10:01:00-07:00")): (self.tmp / f"{name}.json").write_text(json.dumps({ "time": t, "host": "h", "severity": "HIGH", "alerts": [{"signature": sig, "sid": 100010}], "processes": [{"pid": 4242}]})) self.iid = incident.record(self.cfg, "alert-1.log", [_alert("reverse_shell", "r", [4242])], lineage={4242, 999}, when=1000.0, host="h") incident.record(self.cfg, "alert-2.log", [_alert("new_listener", "l", [4243])], lineage={4243, 999}, when=1050.0, host="h") def tearDown(self): self.dir.cleanup() def _run(self, *args): import os os.environ["ENODIA_LOG_DIR"] = str(self.tmp) try: buf = io.StringIO() with redirect_stdout(buf): code = main(list(args)) return code, buf.getvalue() finally: os.environ.pop("ENODIA_LOG_DIR", None) def test_list(self): code, out = self._run("incident", "list") self.assertEqual(code, 0) self.assertIn(self.iid, out) self.assertIn("reverse_shell", out) def test_show_builds_timeline(self): code, out = self._run("incident", "show", self.iid) self.assertEqual(code, 0) self.assertIn("timeline", out) self.assertIn("alert-1.log", out) self.assertIn("alert-2.log", out) # ordered by snapshot time self.assertLess(out.index("10:00:00"), out.index("10:01:00")) def test_show_unknown_id(self): code, _ = self._run("incident", "show", "inc-nope") self.assertEqual(code, 1) def test_export_bundles_snapshots(self): code, out = self._run("incident", "export", self.iid) self.assertEqual(code, 0) bundle = json.loads(out) self.assertEqual(bundle["incident"]["id"], self.iid) self.assertEqual(len(bundle["snapshots"]), 2) if __name__ == "__main__": unittest.main()