9ebc355936
Add `enodia-sentinel posture check` — a config-hygiene audit that finds the conditions making a host easy to attack, complementing the live detectors. Checks: SSH root/password/empty-password login (Include-aware, first-wins sshd resolution), passwordless and group/world-writable sudoers, world-writable or non-root PATH directories, loose permissions on sensitive files (/etc/shadow, /etc/passwd, ...), and downgraded package-signature policy (reusing pkgdb.siglevel_alert). Findings reuse the Alert type, so they serialize through the existing JSON/triage pipeline (`posture check --json`). Each check is a pure evaluator over injected content/stat facts plus a thin system reader, so the 18 new tests need no root or live /etc. Posture is command-driven and never runs in the daemon loop, per the v0.8 exit criterion. SIDs 100040-100047 (classtype host-posture). Sample config and docs (COMMAND_REFERENCE, OPERATIONS, ROADMAP, SPECIFICATION) updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
266 lines
10 KiB
Python
266 lines
10 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 .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")
|
|
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")
|
|
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 == "posture":
|
|
return _cmd_posture(cfg, args.json)
|
|
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_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_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())
|