# 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 .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 _cmd_check(cfg: Config) -> 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) if not alerts: print("No alerts.") return 0 for a in sorted(alerts, key=lambda x: -x.severity): 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 host security 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)") sub.add_parser("check", help="run detectors once and print alerts") sub.add_parser("baseline", help="rebuild listener/SUID baselines") sub.add_parser("list-detectors", help="list available detectors") sub.add_parser("web", help="serve the read-only dashboard") 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: hidden procs/modules/ports") 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)") 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 == "baseline": return _cmd_baseline(cfg) if args.cmd == "check": return _cmd_check(cfg) if args.cmd == "web": from .web import serve serve(cfg) return 0 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 == "watchdog": return _cmd_watchdog(cfg, args.url, args.token, args.max_age) return _cmd_run(cfg) def _cmd_watchdog(cfg: Config, url: str, token: str, max_age: int) -> int: from . import notify from .selfprotect import poll_status, watchdog_verdict ok, msg = watchdog_verdict(poll_status(url, token), 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, ports, or sniffers 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_fim_check(cfg: Config, packages: bool) -> int: from . import fim sentinel = Sentinel(cfg) sentinel.load_fim_baseline() current = fim.scan_paths(cfg.fim_path_list()) d = fim.diff(sentinel.fim_baseline, current) changed = len(d["added"]) + len(d["removed"]) + len(d["modified"]) for path, changes in d["modified"]: print(f"[MODIFIED] {path} ({', '.join(changes)})") for path in d["added"]: print(f"[ADDED] {path}") for path in d["removed"]: print(f"[REMOVED] {path}") 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("\nTo suppress the false positives, add to your config:") for s in sorted(suggestions): print(f" # {s}") return 0 if __name__ == "__main__": sys.exit(main())