diff --git a/README.md b/README.md index 05eadda..cd07047 100644 --- a/README.md +++ b/README.md @@ -236,6 +236,43 @@ notify_ntfy_url = "https://ntfy.sh" notify_ntfy_topic = "enodia-7Hq2x" # keep this secret — it's the access control ``` +## False positives & triage + +A host IDS that cries wolf gets ignored, so Sentinel ships explicit tooling to +separate benign noise from real findings — built on **provenance**: a binary +shipped by your package manager (`pacman`/`dpkg`/`rpm`) is overwhelmingly likely +to be legitimate (the same idea behind OSSEC's rootcheck and AIDE). + +```bash +enodia-sentinel triage # classify captured alerts, suggest allowlist entries +``` + +``` +12 distinct detections — 11 likely false-positive, 1 to review. + +[FP ] new_listener x99 listener binary is package-owned (qbittorrent) +[FP ] new_listener x1 loopback-only listener (not externally reachable) +[REVIEW] new_listener x1 unrecognized listener *:1740 (?) +... +To suppress the false positives, add to your config: + # listener_allow_comms += "qbittorrent" +``` + +Triage is deliberately conservative: `reverse_shell`, `egress`, and the eBPF +exec rules are **always** flagged review (provenance can't clear a network +shell), and any listener it *can't* attribute to a process is reviewed rather +than cleared. Suppression is never automatic — you choose what to allowlist. + +Knobs to quiet known-good activity: + +| Config | Effect | +|---|---| +| `listener_allow_comms` | never alert on listeners owned by these apps | +| `listener_allow_ports` | never alert on these ports | +| `suppress_package_owned_listeners` | drop `new_listener` when the binary is package-owned (best single knob for a desktop/seedbox) | +| `egress_allow_cidrs` | trusted public ranges for the egress detector | +| `suid_hot_dirs` / `exec_rules_file` | tune SUID criticality / add custom exec rules | + ## Security model Sentinel runs as root because it must read every process's `/proc`, the full diff --git a/config/enodia-sentinel.toml b/config/enodia-sentinel.toml index 697e07d..103de66 100644 --- a/config/enodia-sentinel.toml +++ b/config/enodia-sentinel.toml @@ -31,6 +31,14 @@ egress_allow_cidrs = [] # Listener ports that never alert even if they appear after baseline. listener_allow_ports = [] +# Listeners owned by these process names never alert (e.g. P2P clients that +# churn ports: qbittorrent, transmission, nicotine, kdeconnectd, syncthing…). +listener_allow_comms = [] +# Suppress new_listener when the listening binary ships with an installed +# package (strong provenance signal). The single best knob for a desktop / +# seedbox that runs lots of legitimate networked apps. Off by default so the +# detector stays honest; turn on if new_listener is noisy. +suppress_package_owned_listeners = false # Dirs where any SUID binary is treated as CRITICAL (attacker-writable). suid_hot_dirs = ["/tmp", "/dev/shm", "/var/tmp", "/home", "/run/user"] diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index 7de2d78..67966d8 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -62,6 +62,7 @@ def main(argv: list[str] | None = None) -> int: 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") args = parser.parse_args(argv) cfg = Config.load(args.config) @@ -79,8 +80,53 @@ def main(argv: list[str] | None = None) -> int: from .web import serve serve(cfg) return 0 + if args.cmd == "triage": + return _cmd_triage(cfg) return _cmd_run(cfg) +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()) diff --git a/enodia_sentinel/config.py b/enodia_sentinel/config.py index 1eb13cd..a1db3e2 100644 --- a/enodia_sentinel/config.py +++ b/enodia_sentinel/config.py @@ -46,6 +46,11 @@ class Config: interpreters: frozenset[str] = frozenset(_DEFAULT_INTERPRETERS) egress_allow_cidrs: tuple[str, ...] = () listener_allow_ports: frozenset[str] = frozenset() + # Listeners owned by these process names never alert (e.g. P2P clients). + listener_allow_comms: frozenset[str] = frozenset() + # Suppress new_listener when the listening binary is package-owned (strong + # provenance signal; off by default so the detector stays honest). + suppress_package_owned_listeners: bool = False suid_hot_dirs: tuple[str, ...] = ( "/tmp", "/dev/shm", "/var/tmp", "/home", "/run/user", ) diff --git a/enodia_sentinel/detectors/new_listener.py b/enodia_sentinel/detectors/new_listener.py index b1b6a63..a463e95 100644 --- a/enodia_sentinel/detectors/new_listener.py +++ b/enodia_sentinel/detectors/new_listener.py @@ -11,6 +11,7 @@ from collections.abc import Iterator from ..alert import Alert, Severity from ..config import Config +from ..provenance import is_package_owned from ..system import SystemState @@ -25,6 +26,14 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: continue if port in cfg.listener_allow_ports: continue + if comm in cfg.listener_allow_comms: + continue + # Optional provenance gate: a listener whose binary ships with an + # installed package is very likely legitimate (P2P clients, servers). + if cfg.suppress_package_owned_listeners and s.pid: + proc = state.process(s.pid) + if proc and proc.exe and is_package_owned(proc.exe): + continue yield Alert( severity=Severity.HIGH, signature="new_listener", diff --git a/enodia_sentinel/provenance.py b/enodia_sentinel/provenance.py new file mode 100644 index 0000000..7aae627 --- /dev/null +++ b/enodia_sentinel/provenance.py @@ -0,0 +1,63 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Provenance checks — is a file shipped by an installed package? + +The single most effective false-positive filter on a managed Linux system is +*known-good provenance*: a binary owned by the package manager is overwhelmingly +likely to be legitimate (the same principle behind OSSEC rootcheck and AIDE). +We support pacman / dpkg / rpm and cache lookups. + +Caveat by design: package ownership *raises confidence*, it does not prove +safety — a compromised or hijacked package-owned binary is still possible. So +this is used to *rank* and *explain* alerts, never to silently delete them. +""" +from __future__ import annotations + +import functools +import shutil +import subprocess + + +@functools.lru_cache(maxsize=1) +def _tool() -> str | None: + for name in ("pacman", "dpkg", "rpm"): + if shutil.which(name): + return name + return None + + +def _query(tool: str, path: str) -> str | None: + try: + if tool == "pacman": + r = subprocess.run(["pacman", "-Qo", path], capture_output=True, + text=True, timeout=5) + if r.returncode == 0 and "owned by" in r.stdout: + return r.stdout.split("owned by", 1)[1].strip() + elif tool == "dpkg": + r = subprocess.run(["dpkg", "-S", path], capture_output=True, + text=True, timeout=5) + if r.returncode == 0 and ":" in r.stdout: + return r.stdout.split(":", 1)[0].strip() + elif tool == "rpm": + r = subprocess.run(["rpm", "-qf", path], capture_output=True, + text=True, timeout=5) + if r.returncode == 0 and "not owned" not in r.stdout: + return r.stdout.strip().splitlines()[0] + except (OSError, subprocess.SubprocessError): + return None + return None + + +@functools.lru_cache(maxsize=2048) +def package_owner(path: str) -> str | None: + """Return the owning package (e.g. ``qbittorrent 5.2.1-1``) or None.""" + if not path or path == "(deleted)": + return None + path = path.replace(" (deleted)", "") + tool = _tool() + if not tool: + return None + return _query(tool, path) + + +def is_package_owned(path: str) -> bool: + return package_owner(path) is not None diff --git a/enodia_sentinel/triage.py b/enodia_sentinel/triage.py new file mode 100644 index 0000000..ed3becc --- /dev/null +++ b/enodia_sentinel/triage.py @@ -0,0 +1,91 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Alert triage — label a detection likely-false-positive vs needs-review. + +Applies cheap, explainable heuristics (provenance, loopback binds, allowlists, +drill artifacts) so a human isn't drowned in benign noise. Verdicts are advice, +never auto-deletion. ``is_owned`` is injectable for testing. +""" +from __future__ import annotations + +import re +from collections.abc import Callable +from dataclasses import dataclass + +from .config import Config +from .provenance import is_package_owned + +LIKELY_FP = "likely-fp" +REVIEW = "review" + +_LOOPBACK = ("127.", "::1", "[::1]") +_WRITABLE = ("/tmp/", "/dev/shm/", "/var/tmp/", "/run/user/") +# Signatures that provenance can never clear — a network shell or C2 channel is +# worth a look regardless of which binary is involved. +_ALWAYS_REVIEW = {"reverse_shell", "egress"} + + +@dataclass(frozen=True) +class Verdict: + label: str # LIKELY_FP or REVIEW + reason: str + suggest: str = "" # optional config line to suppress future copies + + +def _exe_for(alert: dict, processes: list[dict]) -> str: + pids = set(alert.get("pids", [])) + for p in processes: + if p.get("pid") in pids: + return p.get("exe", "") + return "" + + +def triage_alert(alert: dict, processes: list[dict], cfg: Config, + is_owned: Callable[[str], bool] = is_package_owned) -> Verdict: + sig = alert.get("signature", "") + detail = alert.get("detail", "") + exe = _exe_for(alert, processes) + + if sig.startswith("exec_rule.") or sig in _ALWAYS_REVIEW: + return Verdict(REVIEW, "network/exec signature — provenance can't clear it") + + if sig == "ld_preload": + m = re.search(r"LD_PRELOAD=\[([^\]]+)\]", detail) + lib = m.group(1) if m else "" + if lib.startswith(_WRITABLE): + return Verdict(REVIEW, f"LD_PRELOAD from writable path: {lib}") + if lib and is_owned(lib): + return Verdict(LIKELY_FP, f"preload library is package-owned: {lib}") + return Verdict(REVIEW, "library injection — verify the .so by hand") + + if sig == "deleted_exe": + if "memfd:" in detail: + return Verdict(REVIEW, "memfd-backed (fileless) execution") + if exe and is_owned(exe): + return Verdict(LIKELY_FP, "deleted exe is package-owned (post-upgrade daemon)") + return Verdict(REVIEW, "process running from a deleted binary") + + if sig == "new_listener": + addr = detail.split("socket ", 1)[-1].split(" by ", 1)[0] + comm = detail.rsplit("comm=", 1)[-1].strip() + port = addr.rsplit(":", 1)[-1] + if addr.startswith(_LOOPBACK): + return Verdict(LIKELY_FP, "loopback-only listener (not externally reachable)") + if comm in cfg.listener_allow_comms: + return Verdict(LIKELY_FP, f"comm '{comm}' is allow-listed") + if exe and is_owned(exe): + owner = comm + return Verdict(LIKELY_FP, + f"listener binary is package-owned ({comm})", + suggest=f'listener_allow_comms += "{comm}"') + return Verdict(REVIEW, f"unrecognized listener {addr} ({comm})") + + if sig == "new_suid": + path = detail.rsplit(": ", 1)[-1] + if is_owned(path): + return Verdict(LIKELY_FP, "SUID binary is package-owned") + return Verdict(REVIEW, f"unowned SUID binary: {path}") + + if sig == "persistence": + return Verdict(REVIEW, "a persistence-relevant file changed — confirm intent") + + return Verdict(REVIEW, "no triage rule matched") diff --git a/tests/test_triage.py b/tests/test_triage.py new file mode 100644 index 0000000..91cf9cf --- /dev/null +++ b/tests/test_triage.py @@ -0,0 +1,73 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Tests for false-positive triage (provenance injected, no package manager).""" +import unittest + +from enodia_sentinel.config import Config +from enodia_sentinel.triage import LIKELY_FP, REVIEW, triage_alert + + +def alert(sig, detail, pids=(), sid=0): + return {"signature": sig, "detail": detail, "pids": list(pids), "sid": sid} + + +OWNED = lambda p: p in {"/usr/bin/qbittorrent", "/usr/bin/sudo"} +NONE = lambda p: False + + +class TestNewListener(unittest.TestCase): + def test_package_owned_is_fp(self): + a = alert("new_listener", "new listening socket 1.2.3.4:6881 by comm=qbittorrent", + pids=(50,)) + procs = [{"pid": 50, "exe": "/usr/bin/qbittorrent"}] + v = triage_alert(a, procs, Config(), is_owned=OWNED) + self.assertEqual(v.label, LIKELY_FP) + self.assertIn("qbittorrent", v.suggest) + + def test_loopback_is_fp(self): + a = alert("new_listener", "new listening socket 127.0.0.1:9000 by comm=foo") + v = triage_alert(a, [], Config(), is_owned=NONE) + self.assertEqual(v.label, LIKELY_FP) + + def test_allowlisted_comm_is_fp(self): + c = Config() + c.listener_allow_comms = frozenset({"nicotine"}) + a = alert("new_listener", "new listening socket 1.2.3.4:2234 by comm=nicotine") + self.assertEqual(triage_alert(a, [], c, is_owned=NONE).label, LIKELY_FP) + + def test_unknown_listener_is_review(self): + a = alert("new_listener", "new listening socket 9.9.9.9:31337 by comm=weird", + pids=(77,)) + procs = [{"pid": 77, "exe": "/tmp/weird"}] + self.assertEqual(triage_alert(a, procs, Config(), is_owned=NONE).label, REVIEW) + + +class TestOtherSignatures(unittest.TestCase): + def test_reverse_shell_always_review(self): + a = alert("reverse_shell", "pid=1 comm=bash stdio=net-socket ...") + self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, REVIEW) + + def test_ld_preload_writable_is_review(self): + a = alert("ld_preload", "pid=1 comm=x LD_PRELOAD=[/tmp/evil.so]") + self.assertEqual(triage_alert(a, [], Config(), is_owned=NONE).label, REVIEW) + + def test_suid_package_owned_is_fp(self): + a = alert("new_suid", "new SUID/SGID binary: /usr/bin/sudo") + self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, LIKELY_FP) + + def test_suid_unowned_is_review(self): + a = alert("new_suid", "SUID/SGID binary in writable dir: /tmp/x") + self.assertEqual(triage_alert(a, [], Config(), is_owned=NONE).label, REVIEW) + + def test_deleted_exe_package_owned_is_fp(self): + a = alert("deleted_exe", "pid=1 comm=foo exe=[/usr/bin/foo (deleted)]", pids=(1,)) + procs = [{"pid": 1, "exe": "/usr/bin/foo (deleted)"}] + owned = lambda p: "foo" in p + self.assertEqual(triage_alert(a, procs, Config(), is_owned=owned).label, LIKELY_FP) + + def test_exec_rule_is_review(self): + a = alert("exec_rule.fileless-execution", "sid=100001 ...") + self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, REVIEW) + + +if __name__ == "__main__": + unittest.main()