# 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="Host intrusion-detection daemon.", ) 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") 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) return _cmd_run(cfg) 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())