From a56d72edd67e6e90673f52cabd35f8bf6125d047 Mon Sep 17 00:00:00 2001 From: Luna Date: Wed, 10 Jun 2026 17:52:42 -0700 Subject: [PATCH] Add incident grouping (roadmap v0.8: incident list/show/export) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapse related alerts into one incident, process-lineage first with a time-window fallback. Each alert batch's flagged PIDs are walked up the /proc PPid chain into a lineage set (excluding pid 0/1 so everything doesn't correlate through init); batches whose lineage sets intersect — sharing a process or a common ancestor like the web server or SSH session — join the same incident. PID-less batches (FIM drift, package tamper, hidden modules) fall back to the most recently active open incident within incident_window. snapshot.capture now computes lineage and records each snapshot into a JSON incident index (incidents.json), writing incident_id into both the report JSON (report level, so the per-alert schema is untouched) and the text header. New commands: incident list incidents newest-first, with signatures incident show summary + time-ordered snapshot timeline incident export JSON bundle: record + inlined snapshots The lineage/assign cores are pure functions; record() serializes the index under a lock (capture runs from sweep + eBPF threads) and is best-effort so an index problem never loses a snapshot. 17 new tests (lineage, assign, record, and an end-to-end capture→group→CLI path). Config: incident_tracking / incident_window / incident_lineage_depth. Docs + sample config updated; closes the last v0.8 roadmap item. Co-Authored-By: Claude Opus 4.8 --- config/enodia-sentinel.toml | 9 ++ docs/COMMAND_REFERENCE.md | 31 +++++- docs/ROADMAP.md | 13 +-- docs/SPECIFICATION.md | 9 +- enodia_sentinel/cli.py | 105 ++++++++++++++++++++ enodia_sentinel/config.py | 5 + enodia_sentinel/incident.py | 193 ++++++++++++++++++++++++++++++++++++ enodia_sentinel/snapshot.py | 13 ++- tests/test_incident.py | 180 +++++++++++++++++++++++++++++++++ 9 files changed, 545 insertions(+), 13 deletions(-) create mode 100644 enodia_sentinel/incident.py create mode 100644 tests/test_incident.py diff --git a/config/enodia-sentinel.toml b/config/enodia-sentinel.toml index ecad329..1314893 100644 --- a/config/enodia-sentinel.toml +++ b/config/enodia-sentinel.toml @@ -92,6 +92,15 @@ rootcheck_enabled = true rootcheck_interval = 300 # seconds between cross-view sweeps rootcheck_pid_cap = 65536 # upper PID to brute-force via kill(0) +# --- incident grouping --------------------------------------------------- +# Collapse related alerts into one incident: process-lineage first (shared PID +# or ancestor — the web server / SSH session the alerts descend from), then a +# time fallback for PID-less alerts (FIM drift, package tamper). Each snapshot +# gets an incident_id; browse with `enodia-sentinel incident list/show/export`. +incident_tracking = true +incident_window = 1800 # seconds an incident stays open for new alerts +incident_lineage_depth = 8 # ancestors walked when correlating + # --- host posture -------------------------------------------------------- # Config-hygiene audit run on demand (`enodia-sentinel posture check`), not in # the daemon loop: root SSH login, password auth, passwordless sudo, world- diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index eb97a8b..11bb913 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -155,6 +155,34 @@ Exit code: - `0`: sampled files match and signature policy is not downgraded. - `1`: a signed-package mismatch or insecure signature setting was found. +### `incident` + +```bash +enodia-sentinel incident list [--json] +enodia-sentinel incident show [--json] +enodia-sentinel incident export +``` + +Groups related alerts into incidents so one intrusion reads as one story instead +of several disconnected alerts. Grouping is **process-lineage first** (alerts +sharing a PID or a common ancestor — the web server or SSH session they descend +from), with a **time-window fallback** for alerts that carry no live PID (FIM +drift, package tamper, hidden modules). Each captured snapshot records an +`incident_id`; the index lives in `incidents.json` under `log_dir`. + +- `list` — incidents newest-first: id, last seen, severity, alert count, and the + signatures involved. +- `show ` — incident summary plus a time-ordered timeline of its member + snapshots (time, severity, signatures, PIDs, and the snapshot file to open). +- `export ` — a single JSON bundle of the incident record with every member + snapshot report inlined, suitable for handoff or offline analysis. + +Tuning lives in config: `incident_window` (how long an incident stays open for +new alerts) and `incident_lineage_depth` (ancestors walked when correlating). + +Exit code: `0` on success; `1` if the incident id is unknown; `2` if `show`/ +`export` is called without an id. + ### `posture check` ```bash @@ -255,9 +283,6 @@ does not require pip or a virtualenv. The roadmap reserves these command shapes: ```bash -enodia-sentinel incident list -enodia-sentinel incident show -enodia-sentinel incident export enodia-sentinel respond plan enodia-sentinel respond apply ``` diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index b2ab992..320ab1d 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -20,12 +20,13 @@ promised here; the sequence matters more than calendar precision. Purpose: make the current agent easier to operate and create the data model for work that is bigger than single alerts. -- Add `incident_id` grouping around related alerts while preserving existing - alert JSON. -- Add an incident timeline generator: - alert time, process ancestry, sockets, persistence writes, FIM diffs, package - transactions, and rootcheck findings. -- Add `enodia-sentinel incident list/show/export`. +- ✅ Add `incident_id` grouping around related alerts (process-lineage first, + time-window fallback) while preserving the existing alert JSON — `incident_id` + is added at snapshot-report level. +- ✅ Add an incident timeline generator: `incident show` builds a time-ordered + view across an incident's member snapshots (time, severity, signatures, PIDs). + Richer sources (package transactions, dedicated FIM-diff lines) can follow. +- ✅ Add `enodia-sentinel incident list/show/export`. - ✅ Add host posture checks (`enodia-sentinel posture check`): SSH root/password/empty-password login, passwordless and writable sudoers, world-writable PATH components, loose sensitive-file permissions, and disabled diff --git a/docs/SPECIFICATION.md b/docs/SPECIFICATION.md index 5577f9b..f815f24 100644 --- a/docs/SPECIFICATION.md +++ b/docs/SPECIFICATION.md @@ -92,7 +92,7 @@ When a fresh alert survives cooldown, Sentinel writes: | Interface | Role | |---|---| -| CLI | Run daemon, one-shot checks, baseline management, FIM checks, package DB checks, rootcheck, posture audit, triage, watchdog. | +| CLI | Run daemon, one-shot checks, baseline management, FIM checks, package DB checks, rootcheck, posture audit, incident review, triage, watchdog. | | Logs | Durable local evidence under `/var/log/enodia-sentinel`. | | Dashboard | Read-only web view for status, alert browsing, and snapshot inspection. | | Push | ntfy, Pushover, and generic webhook notifications. | @@ -164,8 +164,11 @@ The current `Alert` object is the public detection unit: | `detail` | Human-readable explanation. | | `pids` | Process IDs for snapshot deep dives. | -Future incident grouping should add an `incident_id` around one or more alerts, -without breaking the alert JSON schema. +Incident grouping adds an `incident_id` at the snapshot-report level (not inside +each `Alert`), so it groups one or more alerts without changing the alert JSON +schema. Grouping is process-lineage first, with a time-window fallback for +PID-less alerts; see the [command reference](COMMAND_REFERENCE.md) and +[roadmap](ROADMAP.md). ## Configuration Model diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index aa3e821..ee0833e 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -77,6 +77,13 @@ def main(argv: list[str] | None = None) -> int: help="anti-rootkit cross-view: hidden procs/modules/ports") st = sub.add_parser("status", help="print daemon health/alert summary") st.add_argument("--json", action="store_true", help="emit status as JSON") + inc = sub.add_parser("incident", + help="list/show/export incidents (grouped alerts)") + inc.add_argument("action", nargs="?", default="list", + choices=["list", "show", "export"], + help="incident action (default: list)") + inc.add_argument("id", nargs="?", help="incident id (for show/export)") + inc.add_argument("--json", action="store_true", help="emit as JSON") po = sub.add_parser("posture", help="audit host config hygiene (SSH, sudo, PATH, perms)") po.add_argument("action", nargs="?", default="check", choices=["check"], @@ -127,6 +134,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_rootcheck(cfg) if args.cmd == "status": return _cmd_status(cfg, args.json) + if args.cmd == "incident": + return _cmd_incident(cfg, args.action, args.id, args.json) if args.cmd == "posture": return _cmd_posture(cfg, args.json) if args.cmd == "watchdog": @@ -207,6 +216,102 @@ def _cmd_status(cfg: Config, as_json: bool) -> int: return 0 if (s["running"] and not s["heartbeat_stale"]) else 1 +def _load_snapshot_report(cfg: Config, log_name: str) -> dict | None: + import json + p = (cfg.log_dir / log_name).with_suffix(".json") + try: + return json.loads(p.read_text()) + except (OSError, ValueError): + return None + + +def _incident_timeline(cfg: Config, inc: dict) -> list[dict]: + """Ordered per-snapshot view: time, severity, signatures, pids.""" + rows = [] + for name in inc.get("snapshots", []): + rep = _load_snapshot_report(cfg, name) + if rep is None: + rows.append({"snapshot": name, "time": "?", "missing": True}) + continue + rows.append({ + "snapshot": name, + "time": rep.get("time", "?"), + "severity": rep.get("severity", "?"), + "signatures": list(dict.fromkeys( + a["signature"] for a in rep.get("alerts", []))), + "pids": rep.get("processes") and + [d["pid"] for d in rep["processes"]] or [], + }) + rows.sort(key=lambda r: r.get("time", "")) + return rows + + +def _cmd_incident(cfg: Config, action: str, iid: str | None, as_json: bool) -> int: + import json + + from . import incident + index = incident.load_index(cfg) + + if action == "list": + incs = sorted(index.values(), key=lambda i: i.get("last_ts", 0.0), + reverse=True) + if as_json: + print(json.dumps(incs, indent=2)) + return 0 + if not incs: + print("No incidents recorded.") + return 0 + print(f"{'INCIDENT':<28} {'LAST SEEN':<26} {'SEV':<9} {'#':>3} SIGNATURES") + for i in incs: + sigs = ", ".join(i.get("signatures", [])) + print(f"{i['id']:<28} {i.get('last_seen', '?'):<26} " + f"{i.get('severity', '?'):<9} {i.get('alert_count', 0):>3} {sigs}") + return 0 + + # show / export both need a resolved incident + if not iid: + print(f"error: 'incident {action}' needs an incident id " + f"(see 'incident list')", file=sys.stderr) + return 2 + inc = index.get(iid) + if inc is None: + print(f"error: no such incident: {iid}", file=sys.stderr) + return 1 + + if action == "export": + bundle = { + "incident": inc, + "snapshots": [r for name in inc.get("snapshots", []) + if (r := _load_snapshot_report(cfg, name)) is not None], + } + print(json.dumps(bundle, indent=2)) + return 0 + + # action == "show" + timeline = _incident_timeline(cfg, inc) + if as_json: + print(json.dumps({"incident": inc, "timeline": timeline}, indent=2)) + return 0 + print(f"Incident {inc['id']} [{inc.get('severity', '?')}] on {inc.get('host', '?')}") + print(f" first seen: {inc.get('first_seen', '?')}") + print(f" last seen: {inc.get('last_seen', '?')}") + print(f" signatures: {', '.join(inc.get('signatures', [])) or '—'}") + print(f" sids: {', '.join(map(str, inc.get('sids', []))) or '—'}") + print(f" pids: {', '.join(map(str, inc.get('pids', []))) or '—'}") + print(f" snapshots: {inc.get('alert_count', 0)} alert(s) over " + f"{len(inc.get('snapshots', []))} capture(s)") + print(" timeline:") + for r in timeline: + if r.get("missing"): + print(f" {r['time']:<26} {r['snapshot']} (snapshot pruned)") + continue + sigs = ", ".join(r["signatures"]) + pids = (" pids=" + ",".join(map(str, r["pids"]))) if r["pids"] else "" + print(f" {r['time']:<26} [{r['severity']}] {sigs}{pids}") + print(f" → {r['snapshot']}") + return 0 + + def _cmd_posture(cfg: Config, as_json: bool) -> int: from . import posture findings = sorted(posture.run(cfg), key=lambda x: -x.severity) diff --git a/enodia_sentinel/config.py b/enodia_sentinel/config.py index aa138ef..1e7d1ad 100644 --- a/enodia_sentinel/config.py +++ b/enodia_sentinel/config.py @@ -83,6 +83,11 @@ class Config: rootcheck_interval: int = 300 # seconds between cross-view sweeps rootcheck_pid_cap: int = 65536 # upper PID to brute-force via kill(0) + # incident grouping (collapse related alerts by process lineage, then time) + incident_tracking: bool = True + incident_window: int = 1800 # s an incident stays open for new alerts + incident_lineage_depth: int = 8 # ancestors walked when correlating + # host posture (config hygiene; advisory, command-driven, never blocks startup) posture_sshd_config: str = "/etc/ssh/sshd_config" posture_sudoers: str = "/etc/sudoers" diff --git a/enodia_sentinel/incident.py b/enodia_sentinel/incident.py new file mode 100644 index 0000000..e7c208b --- /dev/null +++ b/enodia_sentinel/incident.py @@ -0,0 +1,193 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Incident grouping — collapse related alerts into one incident. + +A single intrusion usually trips several signatures: a web service spawns a +shell (`reverse_shell`), the shell opens a port (`new_listener`) and writes +persistence (`persistence`). Reported as three unrelated alerts that is noise; +reported as one incident it is a story. + +Grouping is **process-lineage first, time second**. Each alert batch carries the +PIDs it flagged; we walk their ancestry (`/proc` PPid chain) into a *lineage +set*. Two batches join the same incident when their lineage sets intersect — +i.e. they share a process or a common ancestor (the web server, the SSH session). +Batches with no live PID (FIM drift, package tamper, hidden modules) can't be +placed by lineage, so they fall back to joining the most recently active open +incident within a time window. Distinct process trees stay distinct incidents. + +The incident index is a JSON file in ``log_dir`` updated as snapshots are +captured; ``incident_id`` is added to each snapshot report (the per-alert JSON +schema is untouched). The matching/lineage cores are pure functions so they test +without ``/proc`` or a daemon. +""" +from __future__ import annotations + +import json +import os +import threading +import time +from collections.abc import Callable, Iterable +from datetime import datetime + +from .alert import Alert, Severity +from .config import Config + +INDEX_NAME = "incidents.json" +MAX_INCIDENTS = 1000 # trim the index to the most recent N on save +_DEFAULT_EXCLUDE = frozenset({0, 1}) # never correlate via pid 0 or init + +# The index read-modify-write happens from capture threads (sweep + eBPF), so +# serialize it within this process. +_LOCK = threading.RLock() + + +# --- pure cores (tested without /proc or a daemon) ------------------------ + +def lineage_of( + pids: Iterable[int], + parent_of: Callable[[int], int], + depth: int = 8, + exclude: frozenset[int] = _DEFAULT_EXCLUDE, +) -> set[int]: + """Flagged PIDs plus their ancestors, walking PPid via ``parent_of``. + + ``parent_of(pid)`` returns the parent PID (0 when unknown/root). Stops at + ``exclude`` (pid 0/1 by default, so everything doesn't correlate through + init) and at ``depth`` ancestors. + """ + seen: set[int] = set() + for start in pids: + cur = start + steps = 0 + while cur and cur not in exclude and steps <= depth: + if cur in seen: + break + seen.add(cur) + cur = parent_of(cur) + steps += 1 + return seen + + +def _sev_rank(name: str) -> int: + try: + return int(Severity[name]) + except KeyError: + return 0 + + +def max_severity(a: str, b: str) -> str: + return a if _sev_rank(a) >= _sev_rank(b) else b + + +def assign(index: dict, lineage: set[int], when: float, host: str, + window: int) -> str | None: + """Return the id of an open incident this batch belongs to, or None. + + Lineage-bearing batches join an open incident whose lineage they intersect. + PID-less batches (empty lineage) join the most recently active open incident. + "Open" means last activity is within ``window`` seconds. Ties and PID-less + matches resolve to the most recently active incident. + """ + best: str | None = None + best_ts = -1.0 + for inc in index.values(): + if inc.get("host") != host: + continue + if (when - inc.get("last_ts", 0.0)) > window: + continue # closed + if lineage: + if not (set(inc.get("lineage", ())) & lineage): + continue + # else: PID-less batch — eligible to join any open incident on this host + if inc["last_ts"] > best_ts: + best, best_ts = inc["id"], inc["last_ts"] + return best + + +# --- index persistence ---------------------------------------------------- + +def index_path(cfg: Config): + return cfg.log_dir / INDEX_NAME + + +def load_index(cfg: Config) -> dict: + try: + data = json.loads(index_path(cfg).read_text()) + return data if isinstance(data, dict) else {} + except (OSError, ValueError): + return {} + + +def save_index(cfg: Config, index: dict) -> None: + # Keep only the most recent incidents so the file stays bounded. + if len(index) > MAX_INCIDENTS: + keep = sorted(index.values(), key=lambda i: i.get("last_ts", 0.0), + reverse=True)[:MAX_INCIDENTS] + index = {i["id"]: i for i in keep} + tmp = index_path(cfg).with_suffix(".json.tmp") + tmp.write_text(json.dumps(index, indent=2)) + os.replace(tmp, index_path(cfg)) + try: + index_path(cfg).chmod(0o640) + except OSError: + pass + + +def _new_id(when: float) -> str: + stamp = datetime.fromtimestamp(when).astimezone().strftime("%Y%m%d-%H%M%S") + return f"inc-{stamp}-{os.urandom(2).hex()}" + + +# --- the recorder (called from snapshot.capture) -------------------------- + +def record(cfg: Config, snapshot_name: str, alerts: list[Alert], + lineage: set[int], when: float, host: str) -> str | None: + """Attach this snapshot to an incident (new or existing); return its id. + + Best-effort: any failure returns None so a snapshot is never lost to an + incident-index problem. + """ + if not getattr(cfg, "incident_tracking", True): + return None + try: + with _LOCK: + index = load_index(cfg) + iid = assign(index, lineage, when, host, cfg.incident_window) + iso = datetime.fromtimestamp(when).astimezone().isoformat() + severity = str(max((a.severity for a in alerts), default=Severity.HIGH)) + sigs = list(dict.fromkeys(a.signature for a in alerts)) + sids = list(dict.fromkeys(a.sid for a in alerts if a.sid)) + pids = sorted({p for a in alerts for p in a.pids}) + + if iid is None: + iid = _new_id(when) + index[iid] = { + "id": iid, "host": host, + "first_ts": when, "last_ts": when, + "first_seen": iso, "last_seen": iso, + "severity": severity, + "signatures": [], "sids": [], "pids": [], + "lineage": [], "snapshots": [], "alert_count": 0, + } + inc = index[iid] + inc["last_ts"] = when + inc["last_seen"] = iso + inc["severity"] = max_severity(inc.get("severity", severity), severity) + inc["signatures"] = list(dict.fromkeys(inc["signatures"] + sigs)) + inc["sids"] = list(dict.fromkeys(inc["sids"] + sids)) + inc["pids"] = sorted(set(inc["pids"]) | set(pids)) + inc["lineage"] = sorted(set(inc["lineage"]) | lineage) + inc["snapshots"].append(snapshot_name) + inc["alert_count"] += len(alerts) + save_index(cfg, index) + return iid + except Exception: + return None + + +def lineage_from_state(pids: Iterable[int], state, cfg: Config) -> set[int]: + """Compute a lineage set for ``pids`` using a live SystemState for PPids.""" + def parent_of(pid: int) -> int: + proc = state.process(pid) + return proc.ppid if proc else 0 + return lineage_of(pids, parent_of, + depth=getattr(cfg, "incident_lineage_depth", 8)) diff --git a/enodia_sentinel/snapshot.py b/enodia_sentinel/snapshot.py index 92bd3ba..dab67fb 100644 --- a/enodia_sentinel/snapshot.py +++ b/enodia_sentinel/snapshot.py @@ -116,6 +116,8 @@ def _format_text(report: dict, extras: dict[str, str]) -> str: L.append(f"Time: {report['time']}") L.append(f"Host: {report['host']}") L.append(f"Severity: {report['severity']}") + if report.get("incident_id"): + L.append(f"Incident: {report['incident_id']}") L.append("") L.append("## Triggering detections") for a in report["alerts"]: @@ -161,11 +163,20 @@ def capture(alerts: list[Alert], state: SystemState, cfg: Config) -> Path: severity = max((a.severity for a in alerts), default=Severity.HIGH) pids: list[int] = sorted({p for a in alerts for p in a.pids}) + host = os.uname().nodename + + # Group this batch into an incident by process lineage (then time). Additive: + # the per-alert JSON schema is unchanged; incident_id sits at report level. + from . import incident + lineage = incident.lineage_from_state(pids, state, cfg) + incident_id = incident.record( + cfg, base.with_suffix(".log").name, alerts, lineage, now.timestamp(), host) report = { "time": now.isoformat(), - "host": os.uname().nodename, + "host": host, "severity": str(severity), + "incident_id": incident_id, "alerts": [a.to_dict() for a in alerts], "processes": [_pid_detail(state, p) for p in pids], } diff --git a/tests/test_incident.py b/tests/test_incident.py new file mode 100644 index 0000000..fa24ad0 --- /dev/null +++ b/tests/test_incident.py @@ -0,0 +1,180 @@ +# 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()