457c7604ba
Add two optional windowed frontends over the same local state the TUI and web console read: `enodia-sentinel gui` (stdlib tkinter) and `enodia-sentinel gui-qt` (PySide6, new `[qt]` extra). Both share `gui/model.py`, a GUI-free presentation model over `tui.collect_model`, so every formatter is unit-testable without a display server. Only `app.py` and `qt_app.py` import GUI libraries, enforced by import-guard tests mirroring the tray applet's boundary. Tabs cover status, alerts, incidents, posture, integrity, response plans, and the event tail, plus daemon-control buttons via systemctl. Response plans are display-only; the GUIs never execute containment commands. Core runtime dependencies stay empty. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JX86xeoBJVBb16qHkDf53K
768 lines
30 KiB
Python
768 lines
30 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Command-line entry point.
|
|
|
|
enodia-sentinel run # daemon loop (default; used by systemd)
|
|
enodia-sentinel check # run every detector once, print alerts, exit
|
|
enodia-sentinel baseline # (re)build listener/SUID baselines and exit
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import signal
|
|
import sys
|
|
|
|
from . import __version__, detectors
|
|
from . import schemas
|
|
from .config import Config
|
|
from .daemon import Sentinel
|
|
from .system import SystemState, scan_suid_binaries
|
|
|
|
|
|
def _cmd_run(cfg: Config) -> int:
|
|
sentinel = Sentinel(cfg)
|
|
signal.signal(signal.SIGTERM, sentinel.stop)
|
|
signal.signal(signal.SIGINT, sentinel.stop)
|
|
sentinel.run()
|
|
return 0
|
|
|
|
|
|
def _cmd_baseline(cfg: Config) -> int:
|
|
sentinel = Sentinel(cfg)
|
|
sentinel.build_baselines()
|
|
print(f"Baselines written under {cfg.log_dir}")
|
|
return 0
|
|
|
|
|
|
def _live_fingerprint(cfg: Config, kind: str, target: str) -> tuple[dict, bool]:
|
|
"""Build a fingerprint from the *current* live state and report whether the
|
|
target was actually observed (so accept can warn before blindly recording)."""
|
|
import os
|
|
import stat
|
|
|
|
from . import fim, reconcile
|
|
from .system import SystemState
|
|
|
|
if kind == "fim":
|
|
entry = fim.scan_paths([target]).get(target)
|
|
return reconcile.build_fingerprint("fim", target, entry), entry is not None
|
|
if kind == "pkgfile":
|
|
reasons = [r for _pkg, p, r in fim.pacman_verify() if p == target]
|
|
return reconcile.build_fingerprint("pkgfile", target, reasons), bool(reasons)
|
|
if kind == "listener":
|
|
present = target in SystemState().listener_keys()
|
|
return reconcile.build_fingerprint("listener", target, None), present
|
|
if kind == "suid":
|
|
try:
|
|
mode = os.lstat(target).st_mode
|
|
present = bool(mode & (stat.S_ISUID | stat.S_ISGID))
|
|
except OSError:
|
|
present = False
|
|
return reconcile.build_fingerprint("suid", target, None), present
|
|
raise ValueError(kind)
|
|
|
|
|
|
def _cmd_reconcile(cfg: Config, args) -> int:
|
|
from . import reconcile
|
|
store = reconcile.ReconcileStore.load(cfg)
|
|
if args.action == "accept":
|
|
return _reconcile_accept(cfg, store, args)
|
|
if args.action == "revoke":
|
|
return _reconcile_revoke(store, args)
|
|
return _reconcile_list(cfg, store, args)
|
|
|
|
|
|
def _reconcile_accept(cfg: Config, store, args) -> int:
|
|
from . import reconcile
|
|
if args.kind not in reconcile.KINDS:
|
|
print(f"error: kind must be one of {', '.join(reconcile.KINDS)}",
|
|
file=sys.stderr)
|
|
return 2
|
|
if not args.target:
|
|
print("error: 'baseline accept' needs a target (path or port/comm)",
|
|
file=sys.stderr)
|
|
return 2
|
|
if not args.reason:
|
|
print("error: 'baseline accept' requires --reason", file=sys.stderr)
|
|
return 2
|
|
expires_at = None
|
|
if args.expires:
|
|
try:
|
|
expires_at = reconcile._utcnow() + reconcile.parse_duration(args.expires)
|
|
except ValueError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 2
|
|
fingerprint, present = _live_fingerprint(cfg, args.kind, args.target)
|
|
if not present and not args.force:
|
|
print(f"warning: no live {args.kind} state for {args.target!r}; it cannot "
|
|
f"be fingerprinted now. Re-run with --force to accept anyway.",
|
|
file=sys.stderr)
|
|
return 1
|
|
existed = store.accept(args.kind, args.target, fingerprint, args.reason,
|
|
reconcile.current_actor(), expires_at=expires_at)
|
|
verb = "updated" if existed else "accepted"
|
|
ttl = f" (expires {expires_at:%Y-%m-%d %H:%M} UTC)" if expires_at else ""
|
|
print(f"{verb} {args.kind} {args.target}{ttl}")
|
|
return 0
|
|
|
|
|
|
def _reconcile_revoke(store, args) -> int:
|
|
from . import reconcile
|
|
if args.kind not in reconcile.KINDS or not args.target:
|
|
print("error: 'baseline revoke' needs a kind and target", file=sys.stderr)
|
|
return 2
|
|
if store.revoke(args.kind, args.target):
|
|
print(f"revoked {args.kind} {args.target}")
|
|
return 0
|
|
print(f"error: not found: {args.kind} {args.target}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def _reconcile_list(cfg: Config, store, args) -> int:
|
|
from . import fim, reconcile
|
|
|
|
# Re-derive FIM staleness against live state (cheap — only the acked paths).
|
|
fim_targets = [r.target for r in store.list() if r.kind == "fim"]
|
|
live_fps = {"fim": fim.scan_paths(fim_targets)} if fim_targets else {}
|
|
records = store.list(stale_only=args.stale, live_fps=live_fps)
|
|
any_stale = any(r.status == "stale" for r in store.list())
|
|
|
|
if args.json:
|
|
import json
|
|
print(json.dumps({"schema": schemas.RECONCILE_V1,
|
|
"records": [r.to_dict() for r in records]}, indent=2))
|
|
return 1 if any_stale else 0
|
|
|
|
if not records:
|
|
print("No acknowledged drift." if not args.stale
|
|
else "No stale acknowledgements.")
|
|
return 1 if any_stale else 0
|
|
print(f"{'KIND':<9} {'TARGET':<28} {'STATUS':<6} {'ACCEPTED':<17} "
|
|
f"{'EXPIRES':<11} REASON")
|
|
for r in records:
|
|
accepted = r.accepted_at[:16].replace("T", " ")
|
|
expires = r.expires_at[:10] if r.expires_at else "-"
|
|
print(f"{r.kind:<9} {r.target:<28} {r.status:<6} {accepted:<17} "
|
|
f"{expires:<11} {r.reason}")
|
|
return 1 if any_stale else 0
|
|
|
|
|
|
def _cmd_check(cfg: Config, as_json: bool = False) -> int:
|
|
# One-shot: arm everything immediately, force the SUID scan.
|
|
sentinel = Sentinel(cfg)
|
|
sentinel.start_time = 0.0 # past the grace window
|
|
sentinel.load_baselines()
|
|
if not sentinel.listener_baseline:
|
|
sentinel.build_baselines()
|
|
alerts = sentinel.sweep(force_suid=True)
|
|
ordered = sorted(alerts, key=lambda x: -x.severity)
|
|
if as_json:
|
|
import json
|
|
print(json.dumps([a.to_dict() for a in ordered], indent=2))
|
|
return 0
|
|
if not alerts:
|
|
print("No alerts.")
|
|
return 0
|
|
for a in ordered:
|
|
print(f"[{a.severity}] {a.signature:<14} {a.detail}")
|
|
return 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="enodia-sentinel",
|
|
description="Linux IDS/IPS/EDR platform.",
|
|
)
|
|
parser.add_argument("--version", action="version",
|
|
version=f"enodia-sentinel {__version__}")
|
|
parser.add_argument("-c", "--config", help="path to TOML config")
|
|
sub = parser.add_subparsers(dest="cmd")
|
|
sub.add_parser("run", help="run the daemon loop (default)")
|
|
chk = sub.add_parser("check", help="run detectors once and print alerts")
|
|
chk.add_argument("--json", action="store_true", help="emit alerts as JSON")
|
|
bl = sub.add_parser(
|
|
"baseline",
|
|
help="rebuild baselines (default), or accept/revoke/list drift")
|
|
bl.add_argument("action", nargs="?", default="build",
|
|
choices=["build", "accept", "revoke", "list"],
|
|
help="build (default) rebuilds listener/SUID baselines; "
|
|
"accept/revoke/list manage acknowledged drift")
|
|
bl.add_argument("kind", nargs="?",
|
|
help="drift kind for accept/revoke: fim|pkgfile|listener|suid")
|
|
bl.add_argument("target", nargs="?",
|
|
help="path, or port/comm key, to accept/revoke")
|
|
bl.add_argument("--reason", help="why the drift is acceptable (accept)")
|
|
bl.add_argument("--expires", help="optional TTL for accept: e.g. 7d, 12h, 30m")
|
|
bl.add_argument("--force", action="store_true",
|
|
help="accept even when no live fingerprint can be read")
|
|
bl.add_argument("--stale", action="store_true",
|
|
help="list only stale/expired acknowledgements")
|
|
bl.add_argument("--json", action="store_true",
|
|
help="emit acknowledgements as JSON (list)")
|
|
sub.add_parser("list-detectors", help="list available detectors")
|
|
rules = sub.add_parser("rules",
|
|
help="list/show/test/docs built-in and configured rules")
|
|
rules.add_argument("action", nargs="?", default="list",
|
|
choices=["list", "show", "test", "docs"],
|
|
help="rule action (default: list)")
|
|
rules.add_argument("target", nargs="?",
|
|
help="sid for show, event JSON path for test, or '-'")
|
|
rules.add_argument("--json", action="store_true", help="emit JSON")
|
|
sub.add_parser("web", help="serve the read-only dashboard")
|
|
sub.add_parser("tui", help="open the terminal dashboard (curses)")
|
|
sub.add_parser("gui", help="open the desktop dashboard (tkinter)")
|
|
sub.add_parser("gui-qt", help="open the desktop dashboard (Qt6)")
|
|
sub.add_parser("triage", help="classify captured alerts as likely-FP vs review")
|
|
sub.add_parser("fim-baseline", help="build the file-integrity baseline")
|
|
sub.add_parser("fim-update", help="refresh the FIM baseline (run by the pacman hook)")
|
|
fc = sub.add_parser("fim-check", help="scan monitored files and report changes")
|
|
fc.add_argument("--packages", action="store_true",
|
|
help="also verify package-owned files via pacman -Qkk")
|
|
sub.add_parser("pkgdb-check", help="check the package DB for out-of-band tampering")
|
|
pv = sub.add_parser("pkgdb-verify",
|
|
help="verify on-disk files against the signed cache packages")
|
|
pv.add_argument("--sample", type=int, default=0,
|
|
help="packages to verify (0 = config default; rotates)")
|
|
sub.add_parser("rootcheck",
|
|
help="anti-rootkit cross-view and kernel/module taint checks")
|
|
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"],
|
|
help="posture action (default: check)")
|
|
po.add_argument("--json", action="store_true", help="emit findings as JSON")
|
|
rsp = sub.add_parser("respond",
|
|
help="build and rehearse response plans for incidents")
|
|
rsp.add_argument("action", nargs="?", default="plan", choices=["plan", "apply"],
|
|
help="response action (default: plan)")
|
|
rsp.add_argument("id", nargs="?",
|
|
help="incident id for plan, or saved plan path/id for apply")
|
|
rsp.add_argument("--json", action="store_true", help="emit JSON")
|
|
rsp.add_argument("--dry-run", action="store_true",
|
|
help="for respond apply, audit and print actions without executing")
|
|
wd = sub.add_parser("watchdog",
|
|
help="poll a remote dashboard and push if Sentinel is silent")
|
|
wd.add_argument("--url", required=True, help="dashboard base URL")
|
|
wd.add_argument("--token", default="", help="dashboard bearer token")
|
|
wd.add_argument("--max-age", type=int, default=120,
|
|
help="heartbeat staleness threshold (seconds)")
|
|
wd.add_argument("--insecure-tls", action="store_true",
|
|
help="allow self-signed/untrusted dashboard certificates")
|
|
comp = sub.add_parser("completion",
|
|
help="print shell completion script for bash or zsh")
|
|
comp.add_argument("shell", choices=["bash", "zsh"],
|
|
help="shell to generate completion for")
|
|
|
|
args = parser.parse_args(argv)
|
|
cfg = Config.load(args.config)
|
|
|
|
if args.cmd == "list-detectors":
|
|
for det in detectors.REGISTRY:
|
|
mark = "on " if cfg.enabled(det.name) else "off"
|
|
print(f" [{mark}] {det.name}")
|
|
return 0
|
|
if args.cmd == "rules":
|
|
return _cmd_rules(cfg, args.action, args.target, args.json)
|
|
if args.cmd == "baseline":
|
|
if args.action == "build":
|
|
return _cmd_baseline(cfg)
|
|
return _cmd_reconcile(cfg, args)
|
|
if args.cmd == "check":
|
|
return _cmd_check(cfg, args.json)
|
|
if args.cmd == "web":
|
|
from .web import serve
|
|
serve(cfg)
|
|
return 0
|
|
if args.cmd == "tui":
|
|
from .tui import run as run_tui
|
|
return run_tui(cfg)
|
|
if args.cmd == "gui":
|
|
from .gui.app import main as run_gui
|
|
return run_gui(cfg=cfg)
|
|
if args.cmd == "gui-qt":
|
|
from .gui.qt_app import main as run_gui_qt
|
|
return run_gui_qt(cfg=cfg)
|
|
if args.cmd == "triage":
|
|
return _cmd_triage(cfg)
|
|
if args.cmd in ("fim-baseline", "fim-update"):
|
|
n = Sentinel(cfg).build_fim_baseline()
|
|
print(f"FIM baseline written: {n} files under {cfg.log_dir}")
|
|
return 0
|
|
if args.cmd == "fim-check":
|
|
return _cmd_fim_check(cfg, args.packages)
|
|
if args.cmd == "pkgdb-check":
|
|
from . import pkgdb
|
|
alert = pkgdb.check(cfg)
|
|
if alert:
|
|
print(f"[CRITICAL] {alert.detail}")
|
|
return 1
|
|
print("Package DB: consistent with the anchor (no out-of-band changes).")
|
|
return 0
|
|
if args.cmd == "pkgdb-verify":
|
|
return _cmd_pkgdb_verify(cfg, args.sample)
|
|
if args.cmd == "rootcheck":
|
|
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 == "respond":
|
|
return _cmd_respond(cfg, args.action, args.id, args.json, args.dry_run)
|
|
if args.cmd == "watchdog":
|
|
return _cmd_watchdog(cfg, args.url, args.token, args.max_age,
|
|
not args.insecure_tls)
|
|
if args.cmd == "completion":
|
|
from .completion import script
|
|
print(script(args.shell), end="")
|
|
return 0
|
|
return _cmd_run(cfg)
|
|
|
|
|
|
def _cmd_rules(cfg: Config, action: str, target: str | None,
|
|
as_json: bool) -> int:
|
|
import json
|
|
|
|
from . import ruleops
|
|
|
|
if action == "docs":
|
|
print(ruleops.render_markdown(cfg), end="")
|
|
return 0
|
|
|
|
if action == "list":
|
|
rules = ruleops.list_rules(cfg)
|
|
if as_json:
|
|
print(json.dumps(rules, indent=2))
|
|
return 0
|
|
print(f"{'SID':<7} {'EVENT':<8} {'ORIGIN':<10} {'SEV':<9} "
|
|
f"{'CLASSTYPE':<24} MESSAGE")
|
|
for r in rules:
|
|
print(f"{r['sid']:<7} {r['event']:<8} {r['origin']:<10} "
|
|
f"{r['severity']:<9} {r['classtype']:<24} {r['msg']}")
|
|
return 0
|
|
|
|
if action == "show":
|
|
if not target:
|
|
print("error: 'rules show' needs a sid", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
sid = int(target)
|
|
except ValueError:
|
|
print(f"error: invalid sid: {target}", file=sys.stderr)
|
|
return 2
|
|
rule = ruleops.find_rule(cfg, sid)
|
|
if rule is None:
|
|
print(f"error: no such rule sid: {sid}", file=sys.stderr)
|
|
return 1
|
|
if as_json:
|
|
print(json.dumps(rule, indent=2))
|
|
return 0
|
|
print(f"Rule {rule['sid']} [{rule['severity']}] {rule['signature']}")
|
|
print(f" event: {rule['event']}")
|
|
print(f" origin: {rule['origin']}")
|
|
print(f" classtype: {rule['classtype']}")
|
|
print(f" message: {rule['msg']}")
|
|
print(" conditions:")
|
|
for key, value in rule["conditions"].items():
|
|
print(f" {key}: {value}")
|
|
return 0
|
|
|
|
if action == "test":
|
|
if not target:
|
|
print("error: 'rules test' needs an event JSON path or '-'",
|
|
file=sys.stderr)
|
|
return 2
|
|
try:
|
|
alerts = ruleops.test_event(cfg, ruleops.load_event_json(target))
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"error: cannot test event: {exc}", file=sys.stderr)
|
|
return 2
|
|
if as_json:
|
|
print(json.dumps([a.to_dict() for a in alerts], indent=2))
|
|
return 0 if alerts else 1
|
|
if not alerts:
|
|
print("No rules matched.")
|
|
return 1
|
|
for a in alerts:
|
|
print(f"[{a.severity}] sid={a.sid} {a.signature} {a.detail}")
|
|
return 0
|
|
|
|
print(f"error: unsupported rules action: {action}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
def _cmd_watchdog(cfg: Config, url: str, token: str, max_age: int,
|
|
verify_tls: bool = True) -> int:
|
|
from . import notify
|
|
from .selfprotect import poll_status, watchdog_verdict
|
|
ok, msg = watchdog_verdict(poll_status(url, token, verify_tls=verify_tls),
|
|
max_age)
|
|
print(("OK: " if ok else "ALERT: ") + msg)
|
|
if not ok:
|
|
n = notify.Notification(
|
|
severity=notify.Severity.CRITICAL, host=url,
|
|
signatures=(f"dead-man's-switch — {msg}",), sids=(),
|
|
snapshot_name="-", count=1, dashboard_url=url)
|
|
# bypass the min-severity gate — a silent sensor is always worth a page
|
|
for backend in notify.enabled_backends(cfg):
|
|
try:
|
|
backend.send(cfg, n)
|
|
except Exception:
|
|
pass
|
|
return 0 if ok else 1
|
|
|
|
|
|
def _cmd_pkgdb_verify(cfg: Config, sample: int) -> int:
|
|
from . import pkgdb
|
|
if sample > 0:
|
|
cfg.pkgdb_pkgverify_sample = sample
|
|
sl = pkgdb.siglevel_alert()
|
|
if sl:
|
|
print(f"[CRITICAL] {sl.detail}")
|
|
if not pkgdb.keyring_present():
|
|
print("[WARN] pacman keyring not found — signature trust may be unestablished.")
|
|
alerts = [a for a in pkgdb.verify_alerts(cfg) if a.signature != "pacman_siglevel_disabled"]
|
|
for a in alerts:
|
|
print(f"[CRITICAL] {a.detail}")
|
|
if not sl and not alerts:
|
|
print(f"Package verify: sampled files match the signed cache packages "
|
|
f"(sample={cfg.pkgdb_pkgverify_sample}).")
|
|
return 1 if (sl or alerts) else 0
|
|
|
|
|
|
def _cmd_rootcheck(cfg: Config) -> int:
|
|
from . import rootcheck
|
|
alerts = list(rootcheck.run(cfg))
|
|
if not alerts:
|
|
print("Rootcheck: no hidden processes/modules/sockets, process-tool "
|
|
"hiding, sniffers, known rootkit modules, or kernel taint found.")
|
|
return 0
|
|
for a in sorted(alerts, key=lambda x: -x.severity):
|
|
print(f"[{a.severity}] {a.signature:<22} {a.detail}")
|
|
return 1
|
|
|
|
|
|
def _cmd_status(cfg: Config, as_json: bool) -> int:
|
|
from .web import daemon_status
|
|
s = daemon_status(cfg)
|
|
if as_json:
|
|
import json
|
|
print(json.dumps(s, indent=2))
|
|
else:
|
|
age = s["heartbeat_age"]
|
|
if age is None:
|
|
hb = "none (daemon never wrote one)"
|
|
elif s["heartbeat_stale"]:
|
|
hb = f"STALE — {int(age)}s ago (> {cfg.heartbeat_max_age}s)"
|
|
else:
|
|
hb = f"{int(age)}s ago"
|
|
counts = ", ".join(f"{k}: {v}" for k, v in sorted(s["counts"].items())) or "none"
|
|
print(f"enodia-sentinel {s['version']} on {s['host']}")
|
|
print(f" daemon: {'running' if s['running'] else 'NOT running'}")
|
|
print(f" heartbeat: {hb}")
|
|
print(f" alerts: {s['total_alerts']} total ({counts})")
|
|
print(f" last alert: {s['last_alert'] or 'none'}")
|
|
print(f" eBPF: {s['ebpf']}")
|
|
# Health gate for monitoring: unhealthy if the daemon is down or silent.
|
|
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 = {
|
|
"schema": schemas.INCIDENT_BUNDLE_V1,
|
|
"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({
|
|
"schema": schemas.INCIDENT_VIEW_V1,
|
|
"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)
|
|
if as_json:
|
|
import json
|
|
print(json.dumps([f.to_dict() for f in findings], indent=2))
|
|
return 1 if findings else 0
|
|
if not findings:
|
|
print("Posture: no findings — SSH, sudo, PATH, file perms, and package "
|
|
"signature policy look sound.")
|
|
return 0
|
|
for a in findings:
|
|
print(f"[{a.severity}] {a.signature:<24} {a.detail}")
|
|
return 1
|
|
|
|
|
|
def _cmd_respond(cfg: Config, action: str, iid: str | None, as_json: bool,
|
|
dry_run: bool = False) -> int:
|
|
import json
|
|
|
|
from . import respond
|
|
|
|
if action == "apply":
|
|
return _cmd_respond_apply(cfg, iid, as_json, dry_run)
|
|
if not iid:
|
|
print("error: 'respond plan' needs an incident id "
|
|
"(see 'incident list')", file=sys.stderr)
|
|
return 2
|
|
bundle = respond.load_bundle(cfg, iid)
|
|
if bundle is None:
|
|
print(f"error: no such incident: {iid}", file=sys.stderr)
|
|
return 1
|
|
plan = respond.build_plan(bundle, cfg)
|
|
artifacts = respond.persist_plan(cfg, plan)
|
|
plan = dict(plan)
|
|
plan["artifacts"] = artifacts
|
|
if as_json:
|
|
print(json.dumps(plan, indent=2))
|
|
return 0
|
|
|
|
summary = plan["summary"]
|
|
print(f"Response plan {plan['plan_id']} [{summary['severity']}]")
|
|
print(f" incident: {plan['incident_id']}")
|
|
print(f" mode: {plan['mode']} (no commands executed)")
|
|
print(f" signatures: {', '.join(summary['signatures']) or '—'}")
|
|
print(f" snapshots: {summary['snapshot_count']}")
|
|
print(f" saved: {artifacts['plan_path']}")
|
|
print(f" audit: {artifacts['audit_log']}")
|
|
print(" actions:")
|
|
for a in plan["actions"]:
|
|
cmd = " ".join(a["command"])
|
|
print(f" {a['id']} [{a['risk']}] {a['title']}")
|
|
print(f" {cmd}")
|
|
print(f" reason: {a['reason']}")
|
|
return 0
|
|
|
|
|
|
def _cmd_respond_apply(cfg: Config, ref: str | None, as_json: bool,
|
|
dry_run: bool) -> int:
|
|
import json
|
|
|
|
from . import respond
|
|
|
|
if not ref:
|
|
print("error: 'respond apply' needs a saved plan path or plan id",
|
|
file=sys.stderr)
|
|
return 2
|
|
if not dry_run:
|
|
print("error: response apply execution is not implemented; rerun with "
|
|
"--dry-run to rehearse and audit the reviewed plan",
|
|
file=sys.stderr)
|
|
return 2
|
|
plan, plan_path = respond.load_persisted_plan(cfg, ref)
|
|
if plan is None or plan_path is None:
|
|
print(f"error: no such response plan: {ref}", file=sys.stderr)
|
|
return 1
|
|
result = respond.rehearse_apply(cfg, plan, plan_path)
|
|
output = {
|
|
"schema": schemas.RESPONSE_AUDIT_V1,
|
|
"plan": {
|
|
"plan_id": plan.get("plan_id"),
|
|
"incident_id": plan.get("incident_id"),
|
|
"plan_path": str(plan_path),
|
|
"action_count": len(plan.get("actions", [])),
|
|
},
|
|
"audit": result["record"],
|
|
"audit_log": result["audit_log"],
|
|
"executed": False,
|
|
}
|
|
if as_json:
|
|
print(json.dumps(output, indent=2))
|
|
return 0
|
|
|
|
print(f"Response apply rehearsal {plan.get('plan_id', ref)}")
|
|
print(f" incident: {plan.get('incident_id', '?')}")
|
|
print(f" plan: {plan_path}")
|
|
print(f" audit: {result['audit_log']}")
|
|
print(" executed: no")
|
|
print(" actions:")
|
|
for action in plan.get("actions", []):
|
|
cmd = " ".join(action.get("command", []))
|
|
print(f" {action.get('id', '?')} [{action.get('risk', '?')}] "
|
|
f"{action.get('title', '')}")
|
|
print(f" {cmd}")
|
|
return 0
|
|
|
|
|
|
def _cmd_fim_check(cfg: Config, packages: bool) -> int:
|
|
from . import fim, reconcile
|
|
sentinel = Sentinel(cfg)
|
|
sentinel.load_fim_baseline()
|
|
current = fim.scan_paths(cfg.fim_path_list())
|
|
d = fim.diff(sentinel.fim_baseline, current)
|
|
# Hide drift the operator has acknowledged while it still matches the
|
|
# accepted fingerprint (`current` is the live fingerprint source).
|
|
store = reconcile.ReconcileStore.load(cfg)
|
|
surviving = {a.key for a in
|
|
store.filter_alerts(list(fim.diff_alerts(d)), {"fim": current})}
|
|
changed = 0
|
|
for path, changes in d["modified"]:
|
|
if f"fim:mod:{path}" in surviving:
|
|
print(f"[MODIFIED] {path} ({', '.join(changes)})")
|
|
changed += 1
|
|
for path in d["added"]:
|
|
if f"fim:add:{path}" in surviving:
|
|
print(f"[ADDED] {path}")
|
|
changed += 1
|
|
for path in d["removed"]:
|
|
if f"fim:del:{path}" in surviving:
|
|
print(f"[REMOVED] {path}")
|
|
changed += 1
|
|
if not changed:
|
|
print("FIM: no changes against baseline.")
|
|
if packages:
|
|
print("\nVerifying package-owned files (pacman -Qkk, may take a while)…")
|
|
hits = fim.pacman_verify()
|
|
for pkg, path, reason in hits:
|
|
print(f"[PKG] {path} ({pkg}: {reason})")
|
|
if not hits:
|
|
print("Packages: all verified files match the distro checksums.")
|
|
return 1 if changed else 0
|
|
|
|
|
|
def _cmd_triage(cfg: Config) -> int:
|
|
import json
|
|
from collections import OrderedDict
|
|
|
|
from .triage import LIKELY_FP, triage_alert
|
|
|
|
seen: OrderedDict[tuple, dict] = OrderedDict()
|
|
for p in sorted(cfg.log_dir.glob("alert-*.json")):
|
|
try:
|
|
d = json.loads(p.read_text())
|
|
except (OSError, ValueError):
|
|
continue
|
|
procs = d.get("processes", [])
|
|
for a in d.get("alerts", []):
|
|
key = (a.get("signature"), a.get("detail", "").split(" cmd=")[0])
|
|
if key in seen:
|
|
seen[key]["count"] += 1
|
|
continue
|
|
v = triage_alert(a, procs, cfg)
|
|
seen[key] = {"alert": a, "verdict": v, "count": 1}
|
|
|
|
if not seen:
|
|
print("No alerts to triage.")
|
|
return 0
|
|
|
|
fp = sum(1 for e in seen.values() if e["verdict"].label == LIKELY_FP)
|
|
print(f"{len(seen)} distinct detections — {fp} likely false-positive, "
|
|
f"{len(seen) - fp} to review.\n")
|
|
suggestions = set()
|
|
for e in sorted(seen.values(), key=lambda e: e["verdict"].label):
|
|
a, v = e["alert"], e["verdict"]
|
|
tag = "FP " if v.label == LIKELY_FP else "REVIEW"
|
|
print(f"[{tag}] {a.get('signature'):14} x{e['count']:<3} {v.reason}")
|
|
print(f" {a.get('detail', '')[:100]}")
|
|
if v.suggest:
|
|
suggestions.add(v.suggest)
|
|
if suggestions:
|
|
print("\nSuggested TOML after review (Sentinel does not edit config):")
|
|
for s in sorted(suggestions):
|
|
for line in s.splitlines():
|
|
print(f" {line}")
|
|
print()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|