diff --git a/README.md b/README.md index 863554b..f08b16c 100644 --- a/README.md +++ b/README.md @@ -224,8 +224,14 @@ enodia-sentinel tui enodia-sentinel-tui ``` -Keys: `1` status, `2` alerts, `3` incidents, `4` rules, `r` refresh, `/` filter, -`:` command mode, `Tab` complete command names, `q` quit. +The TUI mirrors the read-only management console for terminal operators: +status, recent alerts, incidents, posture findings, integrity/watchdog state, +incident timelines, event-rule metadata, event tail, and dry-run response-plan +previews. + +Keys: `1` status, `2` alerts, `3` incidents, `4` timelines, `5` posture, +`6` integrity, `7` rules, `8` events, `9` response plans, `r` refresh, `/` filter, `:` +command mode, `Tab` complete command names, `q` quit. Shell completion is generated without extra dependencies: @@ -344,8 +350,10 @@ sudo systemctl enable --now enodia-sentinel-web CLI-generated response plans are saved for handoff/review under `/response-plans/`, with a JSONL trail in `/response-audit.log`. Dashboard plan previews stay read-only and do -not create artifacts. Plans also include read-only recovery checks for package -verification, rootcheck, persistence re-checks, and off-host watchdog visibility. +not create artifacts. Plans also include pacman package-restore guidance +(`pacman -Qo`, signed-cache verification, trusted reinstall, and FIM re-anchor) +plus read-only recovery checks for package verification, rootcheck, persistence +re-checks, and off-host watchdog visibility. Alert snapshots include a best-effort enrichment block that annotates flagged processes and paths with package ownership, executable hashes, parent chains, diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index 795d294..79440c5 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -324,11 +324,14 @@ enodia-sentinel respond plan --json Builds a dry-run response plan from an incident's retained JSON snapshots. The plan may include evidence preservation, process freeze/terminate candidates, outbound IP blocks, systemd unit disablement, suspicious-file quarantine, -package-restore lookup, and follow-up verification checks. Recovery checks stay -read-only too: plans can ask the operator to re-run `fim-check --packages`, -verify package files with `pkgdb-verify`, re-run `rootcheck`, confirm -persistence alerts are gone with `check --json`, and run `watchdog` from another -host to prove the heartbeat/dashboard path is visible. +package-restore planning, and follow-up verification checks. Package restore +guidance stays explicit: identify the owner with `pacman -Qo`, verify signed +package-cache metadata with `pkgdb-verify`, reinstall the owner from trusted +package media, then re-run `fim-check --packages` and `fim-update` only after +operator review. Recovery checks stay read-only too: plans can ask the operator +to re-run `rootcheck`, confirm persistence alerts are gone with `check --json`, +and run `watchdog` from another host to prove the heartbeat/dashboard path is +visible. This command is intentionally read-only: it prints commands for operator review and never executes them. Each CLI-generated plan is also saved under @@ -374,11 +377,14 @@ enodia-sentinel-tui Opens the stdlib curses terminal dashboard for SSH/tmux operators. It does not require the optional tray dependencies and reads the same local state as the CLI: -daemon status, recent alert snapshots, incident index, and event-rule metadata. +daemon status, recent alert snapshots, incident index and timelines, posture +findings, integrity/watchdog state, event-rule metadata, event tail, and +read-only response-plan previews. Keys: -- `1` / `2` / `3` / `4`: status, alerts, incidents, rules. +- `1` / `2` / `3` / `4`: status, alerts, incidents, timelines. +- `5` / `6` / `7` / `8` / `9`: posture, integrity, rules, events, response plans. - `j` / `k` or arrows: scroll. - `r`: refresh. - `/`: filter the current view. @@ -387,8 +393,9 @@ Keys: - `Ctrl-L`: redraw. - `q`: quit. -Command mode supports `status`, `alerts`, `incidents`, `rules`, `help`, -`refresh`, `filter `, `clear`, and `quit`. +Command mode supports `status`, `alerts`, `incidents`, `timeline`, `posture`, +`integrity`, `rules`, `events`, `response`, `help`, `refresh`, +`filter `, `clear`, and `quit`. ### `web` diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index aa434c4..61e52fb 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -78,8 +78,10 @@ summarizes FIM/package anchors plus heartbeat freshness under the Integrity tab for quick remote review. For SSH/tmux operators, `enodia-sentinel tui` provides a terminal dashboard over -the same local state: status, recent alerts, incidents, and rules. Use `:` for -command mode and `Tab` to complete TUI commands. +the same local read-only management state: status, recent alerts, incidents, +incident timelines, posture findings, integrity/watchdog state, event-rule +metadata, event tail, and dry-run response-plan previews. Use `:` for command +mode and `Tab` to complete TUI commands. On desktop hosts the optional tray applet (`enodia-sentinel-tray`, `[tray]` extra) wraps the same workflow: it drives the same `systemctl` units operators @@ -107,11 +109,16 @@ When Sentinel fires: commands. CLI-generated plans are saved under `/var/log/enodia-sentinel/response-plans/` and indexed in `/var/log/enodia-sentinel/response-audit.log` for later review. -5. If the alert involves a binary, run: +5. If the alert involves a package-owned binary, review the package recovery + actions in the response plan and run the restore sequence only after the + owning package and package source are trusted: ```bash - enodia-sentinel fim-check --packages + pacman -Qo /path/to/file enodia-sentinel pkgdb-verify --sample 200 + pacman -S --overwrite '*' + enodia-sentinel fim-check --packages + enodia-sentinel fim-update ``` 6. If the alert involves hiding or tampering, run: diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 8d71653..fe8ce8f 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -67,8 +67,10 @@ Purpose: move from "tell me" to "help me act" without unsafe automation. - ✅ Add dry-run first actions for evidence freeze, process freeze/terminate, outbound IP block, systemd stop/disable, suspicious-file quarantine, package-owner lookup, and follow-up verification. -- Add package reinstall/restore planning after trusted-media and package-manager - semantics are designed for pacman, dpkg, and rpm. +- ✅ Add package reinstall/restore planning for pacman-owned file findings: + identify the owning package, verify signed package-cache metadata, reinstall + from trusted package media, re-run integrity checks, and re-anchor FIM only + after operator review. Debian/RPM support remains future work. - Require explicit `--apply` for changes; default to read-only plans. - Extend response audit logs to any future state-changing `--apply` workflow. - ✅ Add baseline reconciliation: accept legitimate FIM/package/listener/SUID diff --git a/docs/RUNBOOKS.md b/docs/RUNBOOKS.md index acd92e5..f4b90d5 100644 --- a/docs/RUNBOOKS.md +++ b/docs/RUNBOOKS.md @@ -162,12 +162,15 @@ transaction is the suspicious case. - Identify the owning package and reinstall from a trusted source: ```bash + enodia-sentinel respond plan pacman -Qo /path/to/binary + enodia-sentinel pkgdb-verify pacman -S --overwrite '*' # reinstall the verified file enodia-sentinel fim-check --packages # confirm it now matches + enodia-sentinel fim-update # re-anchor only after review ``` -- Then re-anchor: `enodia-sentinel fim-update` and re-run `pkgdb-check`. +- Then re-run `pkgdb-check`. **False positives:** a mid-upgrade snapshot can momentarily disagree. Re-run after the transaction completes; a persistent mismatch is the real finding. diff --git a/docs/SPECIFICATION.md b/docs/SPECIFICATION.md index 4050114..95ea2b6 100644 --- a/docs/SPECIFICATION.md +++ b/docs/SPECIFICATION.md @@ -114,7 +114,8 @@ create artifacts. Current IPS scope is explicit prevention workflow, not transparent inline enforcement. Sentinel can recommend evidence preservation, process freeze/termination, outbound IP blocks, systemd unit disablement, suspicious-file -quarantine, package-owner lookup, and follow-up verification. It does not +quarantine, pacman package-owner lookup, signed-cache verification, trusted +package reinstall, FIM re-anchoring, and follow-up verification. It does not execute those commands until a future audited `--apply` workflow is designed, tested, and documented. diff --git a/enodia_sentinel/respond.py b/enodia_sentinel/respond.py index 60ddced..5eaf491 100644 --- a/enodia_sentinel/respond.py +++ b/enodia_sentinel/respond.py @@ -155,12 +155,30 @@ def build_plan(bundle: dict, cfg: Config) -> dict: add(ResponseAction( id="", category="recovery", - title=f"Restore package-owned file {path}", + title=f"Identify package owner for {path}", command=("pacman", "-Qo", path), reason="Identify the owning package before reinstalling it from trusted media.", risk="low", targets={"path": path}, )) + add(ResponseAction( + id="", + category="recovery", + title=f"Reinstall owner of {path} from trusted package media", + command=("pacman", "-S", "--overwrite", "*", ""), + reason="After pacman -Qo identifies the owner, reinstall that package from a trusted, signature-verified source.", + risk="high", + targets={"path": path, "package": ""}, + )) + add(ResponseAction( + id="", + category="recovery", + title=f"Re-anchor integrity after restoring {path}", + command=("enodia-sentinel", "fim-update"), + reason="Refresh the FIM baseline only after the operator verifies the restored file is legitimate.", + risk="medium", + targets={"path": path}, + )) if {"fim_modified", "fim_pkg_modified", "pkg_signature_mismatch", "pkgdb_tamper"} & sigs: diff --git a/enodia_sentinel/tui.py b/enodia_sentinel/tui.py index c9b0124..460d724 100644 --- a/enodia_sentinel/tui.py +++ b/enodia_sentinel/tui.py @@ -8,16 +8,32 @@ import textwrap from dataclasses import dataclass, field from typing import Iterable -from . import __version__, incident, ruleops +from . import __version__, incident from .config import Config -VIEWS = ("status", "alerts", "incidents", "rules", "help") +VIEWS = ( + "status", + "alerts", + "incidents", + "timeline", + "posture", + "integrity", + "rules", + "events", + "response", + "help", +) TUI_COMMANDS = { "status": "show daemon health and heartbeat", "alerts": "show recent captured alert snapshots", "incidents": "show grouped incidents", + "timeline": "show incident timelines", + "posture": "show advisory posture findings", + "integrity": "show watchdog, FIM, package, and reconciliation state", "rules": "show event-rule catalog", + "events": "tail local events.log", + "response": "preview read-only response plans for incidents", "help": "show keys and commands", "refresh": "reload data from disk", "filter": "filter current view: filter ", @@ -57,20 +73,50 @@ def complete_command(prefix: str, commands: Iterable[str] = TUI_COMMANDS) -> str def collect_model(cfg: Config) -> dict: - from .web import daemon_status + from .web import ( + daemon_status, + integrity_report, + posture_report, + rules_catalog, + tail_events, + ) - return { - "status": daemon_status(cfg), - "alerts": _recent_alerts(cfg), - "incidents": sorted( + status = _safe_section("status", lambda: daemon_status(cfg), {}) + incidents = _safe_section( + "incidents", + lambda: sorted( incident.load_index(cfg).values(), key=lambda i: i.get("last_ts", 0.0), reverse=True, ), - "rules": ruleops.list_rules(cfg), + [], + ) + return { + "status": status, + "alerts": _recent_alerts(cfg), + "incidents": incidents, + "timeline": _incident_views(cfg, incidents), + "posture": _safe_section("posture", lambda: posture_report(cfg), {}), + "integrity": _safe_section( + "integrity", lambda: integrity_report(cfg, status), {} + ), + "rules": _safe_section("rules", lambda: rules_catalog(cfg), {}), + "events": _safe_section("events", lambda: tail_events(cfg), []), + "response": _response_plans(cfg, incidents), } +def _safe_section(name: str, fn, fallback): + try: + return fn() + except Exception as exc: # pragma: no cover - defensive TUI degradation + if isinstance(fallback, dict): + data = dict(fallback) + data["error"] = f"{name}: {exc}" + return data + return [f"{name}: {exc}"] + + def run_tui_command(state: TuiState, command: str) -> TuiState: raw = command.strip() if not raw: @@ -109,8 +155,18 @@ def render_lines(state: TuiState, width: int = 100) -> list[str]: lines = _render_alerts(model.get("alerts") or [], width) elif view == "incidents": lines = _render_incidents(model.get("incidents") or [], width) + elif view == "timeline": + lines = _render_timelines(model.get("timeline") or [], width) + elif view == "posture": + lines = _render_posture(model.get("posture") or {}, width) + elif view == "integrity": + lines = _render_integrity(model.get("integrity") or {}, width) elif view == "rules": - lines = _render_rules(model.get("rules") or [], width) + lines = _render_rules(model.get("rules") or {}, width) + elif view == "events": + lines = _render_events(model.get("events") or [], width) + elif view == "response": + lines = _render_response(model.get("response") or [], width) else: lines = _render_help(width) if state.filter_text: @@ -164,7 +220,17 @@ def _loop(stdscr, cfg: Config) -> None: elif key in (ord("3"),): state.view, state.scroll = "incidents", 0 elif key in (ord("4"),): + state.view, state.scroll = "timeline", 0 + elif key in (ord("5"),): + state.view, state.scroll = "posture", 0 + elif key in (ord("6"),): + state.view, state.scroll = "integrity", 0 + elif key in (ord("7"),): state.view, state.scroll = "rules", 0 + elif key in (ord("8"),): + state.view, state.scroll = "events", 0 + elif key in (ord("9"),): + state.view, state.scroll = "response", 0 elif key in (ord("h"), ord("?")): state.view, state.scroll = "help", 0 elif key == ord(":"): @@ -210,9 +276,7 @@ def _draw(stdscr, state: TuiState) -> None: height, width = stdscr.getmaxyx() width = max(width, 20) stdscr.erase() - title = f" Enodia Sentinel TUI v{__version__} " - tabs = " 1 Status 2 Alerts 3 Incidents 4 Rules h Help r Refresh q Quit" - stdscr.addnstr(0, 0, (title + tabs)[:width - 1], width - 1, curses.A_REVERSE) + stdscr.addnstr(0, 0, _header_text(state, width), width - 1, curses.A_REVERSE) meta = f"view={state.view} filter={state.filter_text or '-'} {state.message}" stdscr.addnstr(1, 0, meta[:width - 1], width - 1) lines = render_lines(state, width) @@ -228,6 +292,24 @@ def _draw(stdscr, state: TuiState) -> None: stdscr.refresh() +def _header_text(state: TuiState, width: int) -> str: + title = f" Enodia Sentinel TUI v{__version__} " + full_tabs = ( + "1 Status 2 Alerts 3 Incidents 4 Timeline 5 Posture " + "6 Integrity 7 Rules 8 Events 9 Response h Help r Refresh q Quit" + ) + compact_tabs = ( + "1 Status 2 Alerts 3 Inc 4 Time 5 Post 6 Int 7 Rules 8 Event 9 Resp" + ) + if width >= len(title) + len(full_tabs) + 2: + text = title + full_tabs + elif width >= len(title) + len(compact_tabs) + 2: + text = title + compact_tabs + else: + text = f"{title}view={state.view} 1-9 switch h help r refresh q quit" + return text[:max(0, width - 1)] + + def _render_status(status: dict, width: int) -> list[str]: if not status: return ["No status available."] @@ -248,7 +330,8 @@ def _render_status(status: dict, width: int) -> list[str]: f"Exec probe: {status.get('ebpf_exec', 'unknown')}", f"Syscall: {status.get('ebpf_syscall', 'unknown')}", "", - "Use 2/3/4 to inspect alerts, incidents, and rules. Use ':' for commands.", + "Use 2-9 to inspect alerts, incidents, timelines, posture, integrity, rules, events, and response plans.", + "Use ':' for commands.", ] @@ -281,10 +364,133 @@ def _render_incidents(incidents: list[dict], width: int) -> list[str]: return lines -def _render_rules(rules: list[dict], width: int) -> list[str]: +def _render_timelines(views: list[dict], width: int) -> list[str]: + if not views: + return ["No incident timelines."] + lines = [] + for view in views: + if view.get("error"): + lines.append(f"{view.get('incident_id', '?')}: {view['error']}") + lines.append("") + continue + inc = view.get("incident") or {} + lines.append( + f"{inc.get('id', '?')} severity={inc.get('severity', '?')} " + f"alerts={inc.get('alert_count', 0)}" + ) + sigs = ", ".join(inc.get("signatures") or []) + if sigs: + lines.append(f" signatures: {_clip(sigs, max(20, width - 15))}") + timeline = view.get("timeline") or [] + if not timeline: + lines.append(" No timeline rows.") + for row in timeline: + if row.get("missing"): + lines.append( + f" {row.get('time', '?'):<24} missing snapshot {row.get('snapshot', '?')}" + ) + continue + sig_text = ", ".join(row.get("signatures") or []) + pid_text = ",".join(str(p) for p in row.get("pids") or []) + lines.append( + f" {row.get('time', '?')[:24]:<24} {row.get('severity', '?'):<9} " + f"{_clip(sig_text, max(20, width - 45))} pids={pid_text or '-'}" + ) + lines.append("") + return lines + + +def _render_posture(report: dict, width: int) -> list[str]: + if report.get("error"): + return [f"Posture unavailable: {report['error']}"] + findings = report.get("findings") or [] + counts = report.get("counts") or {} + count_text = ", ".join(f"{k}:{v}" for k, v in sorted(counts.items())) or "none" + lines = [ + f"Posture findings: {report.get('count', len(findings))} ({count_text})", + "", + ] + if not findings: + lines.append("No posture findings.") + return lines + lines.append(f"{'SEV':<9} {'SID':<7} {'SIGNATURE':<22} DETAIL") + for finding in findings: + detail = finding.get("detail", "") + lines.append( + f"{finding.get('severity', '?'):<9} {finding.get('sid', 0):<7} " + f"{finding.get('signature', '?'):<22} {_clip(detail, max(20, width - 42))}" + ) + return lines + + +def _render_integrity(report: dict, width: int) -> list[str]: + if report.get("error"): + return [f"Integrity unavailable: {report['error']}"] + if not report: + return ["No integrity report available."] + checks = report.get("checks") or {} + watchdog = report.get("watchdog") or {} + anchors = report.get("anchors") or {} + fim = anchors.get("fim_baseline") or {} + pkgdb = anchors.get("pkgdb") or {} + pacman = anchors.get("pacman") or {} + footprint = report.get("sentinel_footprint") or {} + reconciliation = report.get("reconciliation") or {} + lines = [ + f"Overall: {report.get('status', 'unknown')}", + f"Generated: {report.get('generated_at', '?')}", + "", + "Checks:", + ] + for name, state in sorted(checks.items()): + lines.append(f" {name:<18} {state}") + lines.extend([ + "", + "Watchdog:", + f" Status: {'ok' if watchdog.get('ok') else 'review'}", + f" Message: {_clip(watchdog.get('message', '?'), max(20, width - 17))}", + f" Heartbeat age: {_format_age(watchdog.get('heartbeat_age'))}", + f" Max age: {watchdog.get('max_age', '?')}s", + f" Daemon: {'running' if watchdog.get('daemon_running') else 'stopped'}", + "", + "Anchors:", + f" FIM baseline: {fim.get('status', 'unknown')} ({fim.get('path', '?')})", + f" Package DB: {pkgdb.get('status', 'unknown')} prefix={pkgdb.get('fingerprint_prefix', '')}", + f" Pacman keyring:{' present' if pacman.get('keyring_present') else ' missing'}", + f" SigLevel: {pacman.get('siglevel_status', 'unknown')}", + "", + "Sentinel footprint:", + f" Present: {footprint.get('present', 0)}/{footprint.get('configured', 0)}", + f" Missing: {footprint.get('missing', 0)}", + "", + "Reconciliation:", + f" Total: {reconciliation.get('total', 0)}", + f" Stale: {reconciliation.get('stale', 0)}", + ]) + for item in reconciliation.get("stale_items") or []: + lines.append( + f" stale {item.get('kind', '?')} {item.get('target', '?')} - " + f"{_clip(item.get('reason', ''), max(20, width - 20))}" + ) + return lines + + +def _render_rules(catalog, width: int) -> list[str]: + if isinstance(catalog, dict) and catalog.get("error"): + return [f"Rules unavailable: {catalog['error']}"] + rules = catalog.get("rules", []) if isinstance(catalog, dict) else catalog if not rules: return ["No rules loaded."] - lines = [f"{'SID':<7} {'EVENT':<8} {'SEV':<9} {'CLASSTYPE':<24} MESSAGE"] + lines = [] + if isinstance(catalog, dict): + lines.append( + "Rules: " + f"{catalog.get('count', len(rules))} total; " + f"events={_dict_counts(catalog.get('by_event') or {})}; " + f"severity={_dict_counts(catalog.get('by_severity') or {})}" + ) + lines.append("") + lines.append(f"{'SID':<7} {'EVENT':<8} {'SEV':<9} {'CLASSTYPE':<24} MESSAGE") for rule in rules: lines.append( f"{rule.get('sid', 0):<7} {rule.get('event', '?'):<8} " @@ -294,10 +500,48 @@ def _render_rules(rules: list[dict], width: int) -> list[str]: return lines +def _render_events(events: list[str], width: int) -> list[str]: + if not events: + return ["No events.log entries found."] + return [_clip(line, width) for line in events] + + +def _render_response(plans: list[dict], width: int) -> list[str]: + if not plans: + return ["No incidents with response plans."] + lines = [ + "Read-only response previews. Commands are not executed from the TUI.", + "", + ] + for plan in plans: + if plan.get("error"): + lines.append(f"{plan.get('incident_id', '?')}: {plan['error']}") + lines.append("") + continue + summary = plan.get("summary") or {} + lines.append( + f"{plan.get('incident_id', '?')} severity={summary.get('severity', '?')} " + f"actions={summary.get('action_count', len(plan.get('actions') or []))} " + f"mode={plan.get('mode', 'dry-run')}" + ) + sigs = ", ".join(summary.get("signatures") or []) + if sigs: + lines.append(f" signatures: {_clip(sigs, max(20, width - 15))}") + for action in plan.get("actions") or []: + cmd = " ".join(str(part) for part in action.get("command") or []) + lines.append( + f" {action.get('id', '?'):<4} {action.get('category', '?'):<12} " + f"{action.get('risk', '?'):<7} {_clip(action.get('title', ''), max(20, width - 31))}" + ) + lines.append(f" {_clip(cmd, max(20, width - 7))}") + lines.append("") + return lines + + def _render_help(width: int) -> list[str]: lines = [ "Keys:", - " 1/2/3/4 switch views: status, alerts, incidents, rules", + " 1-9 switch views: status, alerts, incidents, timeline, posture, integrity, rules, events, response", " j/k or arrows scroll", " PgUp/PgDn page scroll", " r refresh", @@ -327,5 +571,63 @@ def _recent_alerts(cfg: Config, limit: int = 50) -> list[dict]: return rows +def _response_plans(cfg: Config, incidents: list[dict], limit: int = 20) -> list[dict]: + from .web import response_plan + + plans = [] + for inc in incidents[:limit]: + incident_id = inc.get("id") + if not incident_id: + continue + try: + plan = response_plan(cfg, str(incident_id)) + except Exception as exc: # pragma: no cover - defensive TUI degradation + plans.append({"incident_id": incident_id, "error": f"response: {exc}"}) + continue + if isinstance(plan, dict): + plans.append(plan) + return plans + + +def _incident_views(cfg: Config, incidents: list[dict], limit: int = 20) -> list[dict]: + from .web import get_incident + + views = [] + for inc in incidents[:limit]: + incident_id = inc.get("id") + if not incident_id: + continue + try: + view = get_incident(cfg, str(incident_id)) + except Exception as exc: # pragma: no cover - defensive TUI degradation + views.append({"incident_id": incident_id, "error": f"timeline: {exc}"}) + continue + if isinstance(view, dict): + views.append(view) + return views + + +def _clip(text, width: int) -> str: + text = str(text) + if width <= 0 or len(text) <= width: + return text + if width <= 3: + return text[:width] + return text[:width - 3] + "..." + + +def _format_age(value) -> str: + if value is None: + return "none" + try: + return f"{int(value)}s" + except (TypeError, ValueError): + return str(value) + + +def _dict_counts(data: dict) -> str: + return ", ".join(f"{k}:{v}" for k, v in sorted(data.items())) or "none" + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/tests/test_respond.py b/tests/test_respond.py index 5d08a30..dcdcedb 100644 --- a/tests/test_respond.py +++ b/tests/test_respond.py @@ -103,6 +103,13 @@ class TestRespondPlan(unittest.TestCase): self.assertIn(["enodia-sentinel", "fim-check", "--packages"], commands) self.assertIn(["enodia-sentinel", "pkgdb-verify"], commands) self.assertIn(["pacman", "-Qo", "/usr/bin/ssh"], commands) + self.assertIn(["pacman", "-S", "--overwrite", "*", ""], commands) + self.assertIn(["enodia-sentinel", "fim-update"], commands) + restore = next(a for a in plan["actions"] + if a["title"].startswith("Reinstall owner")) + self.assertEqual(restore["targets"]["path"], "/usr/bin/ssh") + self.assertEqual(restore["targets"]["package"], "") + self.assertTrue(restore["requires_review"]) self.assertIn(["enodia-sentinel", "watchdog", "--url", "https://100.64.0.2:8787", "--token", "", "--insecure-tls"], commands) diff --git a/tests/test_tui.py b/tests/test_tui.py index 52eb070..aeee578 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -10,7 +10,10 @@ from pathlib import Path from enodia_sentinel.cli import main from enodia_sentinel.config import Config from enodia_sentinel.tui import ( + TUI_COMMANDS, + VIEWS, TuiState, + _header_text, collect_model, complete_command, render_lines, @@ -30,13 +33,18 @@ class TestTuiCore(unittest.TestCase): def test_command_completion_unique_and_common_prefix(self): self.assertEqual(complete_command("inc"), "incidents") + self.assertEqual(complete_command("resp"), "response") self.assertEqual(complete_command("r"), "r") self.assertEqual(complete_command("zzz"), "zzz") def test_run_tui_command_switches_views_and_filters(self): state = TuiState() - run_tui_command(state, "incidents") - self.assertEqual(state.view, "incidents") + for view in VIEWS: + if view == "help": + continue + with self.subTest(view=view): + run_tui_command(state, view) + self.assertEqual(state.view, view) run_tui_command(state, "filter ssh") self.assertEqual(state.filter_text, "ssh") run_tui_command(state, "clear") @@ -44,8 +52,14 @@ class TestTuiCore(unittest.TestCase): run_tui_command(state, "quit") self.assertTrue(state.should_quit) - def test_collect_model_and_render_status(self): + def test_collect_model_contains_dashboard_parity_sections(self): model = collect_model(self.cfg) + for key in ( + "status", "alerts", "incidents", "timeline", "posture", "integrity", + "rules", "events", "response", + ): + self.assertIn(key, model) + state = TuiState(model=model) lines = render_lines(state, width=80) self.assertTrue(any("Daemon:" in line for line in lines)) @@ -66,6 +80,92 @@ class TestTuiCore(unittest.TestCase): state.filter_text = "nomatch" self.assertEqual(render_lines(state), ["No rows match filter: nomatch"]) + def test_render_posture_integrity_events_and_response_views(self): + model = { + "posture": { + "count": 1, + "counts": {"HIGH": 1}, + "findings": [{ + "severity": "HIGH", + "sid": 100040, + "signature": "ssh_root_login", + "detail": "PermitRootLogin yes", + }], + }, + "timeline": [{ + "incident": { + "id": "inc-1", + "severity": "CRITICAL", + "alert_count": 1, + "signatures": ["reverse_shell"], + }, + "timeline": [{ + "time": "2026-07-09T01:02:03-07:00", + "severity": "CRITICAL", + "signatures": ["reverse_shell"], + "pids": [4242], + }], + }], + "integrity": { + "status": "review", + "generated_at": 1.0, + "checks": {"watchdog": "ok", "fim_baseline": "missing"}, + "watchdog": { + "ok": True, + "message": "heartbeat fresh", + "heartbeat_age": 2, + "max_age": 120, + "daemon_running": True, + }, + "anchors": { + "fim_baseline": {"status": "missing", "path": "/tmp/fim.json"}, + "pkgdb": {"status": "ok", "fingerprint_prefix": "abc123"}, + "pacman": {"keyring_present": True, "siglevel_status": "ok"}, + }, + "sentinel_footprint": {"present": 2, "configured": 3, "missing": 1}, + "reconciliation": { + "total": 1, + "stale": 1, + "stale_items": [{ + "kind": "fim", + "target": "/etc/passwd", + "reason": "ttl expired", + }], + }, + }, + "events": ["2026-07-09T01:02:03 event happened"], + "response": [{ + "incident_id": "inc-1", + "mode": "dry-run", + "summary": { + "severity": "CRITICAL", + "signatures": ["reverse_shell"], + "action_count": 1, + }, + "actions": [{ + "id": "A001", + "category": "evidence", + "risk": "low", + "title": "Freeze Sentinel evidence bundle", + "command": ["tar", "-czf", "/tmp/evidence.tgz"], + }], + }], + } + + posture = render_lines(TuiState(view="posture", model=model), width=100) + self.assertTrue(any("ssh_root_login" in line for line in posture)) + timeline = render_lines(TuiState(view="timeline", model=model), width=100) + self.assertTrue(any("inc-1" in line for line in timeline)) + self.assertTrue(any("pids=4242" in line for line in timeline)) + integrity = render_lines(TuiState(view="integrity", model=model), width=100) + self.assertTrue(any("Overall:" in line for line in integrity)) + self.assertTrue(any("ttl expired" in line for line in integrity)) + events = render_lines(TuiState(view="events", model=model), width=100) + self.assertEqual(events, ["2026-07-09T01:02:03 event happened"]) + response = render_lines(TuiState(view="response", model=model), width=100) + self.assertTrue(any("inc-1" in line for line in response)) + self.assertTrue(any("tar -czf /tmp/evidence.tgz" in line for line in response)) + def test_cli_completion_outputs_scripts(self): for shell in ("bash", "zsh"): with self.subTest(shell=shell): @@ -76,6 +176,21 @@ class TestTuiCore(unittest.TestCase): self.assertIn("enodia-sentinel", buf.getvalue()) self.assertIn("tui", buf.getvalue()) + def test_help_lists_all_commands(self): + help_lines = render_lines(TuiState(view="help"), width=100) + for command in TUI_COMMANDS: + self.assertTrue(any(command in line for line in help_lines), command) + + def test_header_scales_to_terminal_width(self): + state = TuiState(view="integrity") + for width in (60, 100, 180): + with self.subTest(width=width): + header = _header_text(state, width) + self.assertLessEqual(len(header), width - 1) + self.assertIn("view=integrity", _header_text(state, 60)) + self.assertIn("6 Int", _header_text(state, 100)) + self.assertIn("6 Integrity", _header_text(state, 180)) + if __name__ == "__main__": unittest.main()