From 1de5e86fed48687730312be37a498d0c24239fc7 Mon Sep 17 00:00:00 2001 From: Luna Date: Mon, 1 Jun 2026 05:42:57 -0700 Subject: [PATCH] Add signed-package verification and anti-rootkit cross-view (v0.7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the tamper-evidence loop with two trust anchors the attacker can't forge from userland: - pkgdb Layer 2: verify on-disk files against the .MTREE in the *signed* cache package, surviving a rewritten local checksum DB. Rotating-sample cadence keeps it affordable; flags pkg_signature_mismatch (sid 100027) and SigLevel downgrades (sid 100026). Fixes parse_mtree, which required type=file on every line and so matched nothing on real pacman MTREEs (which use a /set type=file default with bare file entries). - rootcheck: anti-rootkit cross-view — hidden processes, modules, ports, and promiscuous interfaces, each caught by diffing two views of the same state (sids 100022-100025). Wired both through config, the daemon (off-loop slow cadence), and CLI (pkgdb-verify, rootcheck). 14 new tests (95 total). Docs + version bump. Co-Authored-By: Claude Opus 4.8 --- README.md | 76 ++++++++++++- enodia_sentinel/__init__.py | 2 +- enodia_sentinel/cli.py | 39 +++++++ enodia_sentinel/config.py | 11 ++ enodia_sentinel/daemon.py | 47 ++++++++ enodia_sentinel/pkgdb.py | 210 +++++++++++++++++++++++++++++++++++ enodia_sentinel/rootcheck.py | 209 ++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_rootcheck.py | 89 +++++++++++++++ tests/test_tamper.py | 86 ++++++++++++++ 10 files changed, 766 insertions(+), 5 deletions(-) create mode 100644 enodia_sentinel/rootcheck.py create mode 100644 tests/test_rootcheck.py diff --git a/README.md b/README.md index cf150fa..88fa5a2 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,9 @@ enodia_sentinel/ ├── web.py read-only dashboard: stdlib http server + JSON API + auth ├── static/ the self-contained dashboard SPA ├── fim.py file integrity monitoring (hash baseline + pacman verify) +├── pkgdb.py package-DB integrity + signed-package verification +├── rootcheck.py anti-rootkit cross-view (hidden procs/modules/ports) +├── selfprotect.py self-integrity footprint + dead-man's-switch heartbeat ├── triage.py false-positive classification ├── provenance.py package-ownership lookups (pacman/dpkg/rpm) ├── detectors/ poll detectors — one module per signature, each a pure @@ -196,6 +199,10 @@ enodia-sentinel.service`. Every key is optional. Highlights: | `suid_scan_extra_dirs` | /tmp … | writable mounts always scanned (tmpfs-safe) | | `capture_execve_bpftrace` | false | add a bpftrace execve trace to snapshots | | `notify_users` | [] | desktop notify-send targets | +| `pkgdb_pkgverify` | false | verify on-disk files against signed cache packages | +| `pkgdb_pkgverify_sample` | 40 | packages verified per pass (rotates over time) | +| `rootcheck_enabled` | true | run the anti-rootkit cross-view sweep | +| `rootcheck_interval` | 300 | seconds between cross-view sweeps | ## Web dashboard @@ -273,9 +280,10 @@ enodia-sentinel watchdog --url http://100.x.x.x:8787 --token --max-age 120 1. **Immutability** — `chattr +i` Sentinel's binaries, config, baselines, and the pacman hook so even root must visibly clear the flag first. -2. **Cryptographic anchor (next)** — verify files against the *signed* package - in the cache rather than the mutable DB; a maintainer's PGP signature is a - root of trust the attacker doesn't hold. +2. ✅ **Cryptographic anchor (done)** — verify on-disk files against the *signed* + package in the cache rather than the mutable DB; a maintainer's PGP signature + is a root of trust the attacker doesn't hold. See *Signed-package verification* + below. 3. **External anchor (next)** — mirror the DB/FIM fingerprints off-box so a local attacker can't refresh the anchor to cover their tracks. @@ -285,6 +293,61 @@ enodia-sentinel watchdog --url http://100.x.x.x:8787 --token --max-age 120 > attacker: signatures, external monitors, and ultimately the kernel (the eBPF > roadmap) rather than userland the attacker can rewrite. +### Signed-package verification (the independent anchor) + +Layer 1 (above) catches an *out-of-band* DB edit. But a root attacker can do the +full job: modify `/usr/bin/sshd`, rewrite its hash in the local DB, **and** run +`fim-update` to re-anchor — defeating both `pacman -Qkk` and the DB-fingerprint +check, because every reference they're checked against is one they can rewrite. + +The one reference they *can't* forge is the distro's signing key. Packages in +the cache are signed, and each carries a `.MTREE` manifest of per-file SHA-256 +hashes. Layer 2 extracts that manifest from the **cached package** and compares +the on-disk files to it — a reference independent of the local DB: + +```bash +enodia-sentinel pkgdb-verify # verify a rotating sample of packages +enodia-sentinel pkgdb-verify --sample 200 # verify more per run +``` + +A divergence is `pkg_signature_mismatch` (CRITICAL, `sid 100027`) — a trojaned +binary that a rewritten checksum DB would have hidden. It also flags a +`SigLevel` downgrade in `pacman.conf` (`pacman_siglevel_disabled`, CRITICAL, +`sid 100026`), since disabling signature checking is how an attacker would slip +an unsigned package past the anchor in the first place. + +Verifying every package each pass is expensive (untar + hash every file), so it +runs on its own slow cadence (`pkgdb_pkgverify_interval`) and checks a **rotating +sample** (`pkgdb_pkgverify_sample`) per pass — over enough passes the whole +installed set is covered, and any single mismatch fires immediately. It's +off by default (needs the package cache populated); enable with `pkgdb_pkgverify += true`. + +## Anti-rootkit (cross-view detection) + +A rootkit hides by lying to one view of the system — but the technique that +hides it is also how you catch it: **ask the same question two different ways and +compare the answers.** A discrepancy is the hiding artifact. + +| Cross-check | Hidden thing it surfaces | `sid` | +|---|---|---| +| `kill(pid, 0)` for every PID vs the `/proc` listing | a process the kernel schedules but `/proc` omits | 100022 | +| `/sys/module` (initstate=live) vs `/proc/modules` | a loaded LKM hidden from the module list | 100023 | +| `/proc/net/tcp` vs `ss` | a listening port a hooked `ss` won't report | 100024 | +| `/sys/class/net/*/flags` | an interface in promiscuous mode (a sniffer) | 100025 | + +```bash +enodia-sentinel rootcheck # one-shot cross-view scan +``` + +The daemon runs it on a slow background cadence (`rootcheck_interval`) and routes +any finding into the normal alert/snapshot/push pipeline. **Honest limits:** this +runs in user space, so it reliably catches userland (`LD_PRELOAD`) rootkits and +the common `/proc`-hiding LKMs, but a kernel rootkit that hooks *every* path +consistently can still evade it. It raises the bar and catches the common cases; +it is not a guarantee against a bespoke ring-0 implant — pair it with the off-box +dead-man's switch. (Lineage: OSSEC's rootcheck / `chkrootkit`'s cross-view idea.) + ## File integrity monitoring (Tripwire-style) Detects tampering with binaries and critical configs by **content hash**, so it @@ -395,6 +458,13 @@ regression suite for both. ## Project status +v0.7 — closes the tamper-evidence loop with the **independent anchor**: +signed-package verification (compares on-disk files to the `.MTREE` in the signed +cache package, surviving a rewritten checksum DB) plus a `SigLevel`-downgrade +check, and an **anti-rootkit cross-view** layer (hidden processes, modules, +ports, and promiscuous interfaces — each caught by asking the same question two +ways and diffing the answers). + v0.6 — adds **tamper-evidence**: out-of-band package-DB integrity (catches rewritten checksums), self-integrity of Sentinel's own footprint, and a dead-man's-switch heartbeat with an external watchdog. diff --git a/enodia_sentinel/__init__.py b/enodia_sentinel/__init__.py index e5af105..7b95879 100644 --- a/enodia_sentinel/__init__.py +++ b/enodia_sentinel/__init__.py @@ -6,4 +6,4 @@ captures a forensic snapshot with incident-response guidance whenever a known attack signature appears. The Python re-architecture of the bash v0 prototype. """ -__version__ = "0.6.0" +__version__ = "0.7.0" diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index d485854..19bc262 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -69,6 +69,12 @@ def main(argv: list[str] | None = None) -> int: 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") @@ -108,6 +114,10 @@ def main(argv: list[str] | None = None) -> int: 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) @@ -132,6 +142,35 @@ def _cmd_watchdog(cfg: Config, url: str, token: str, max_age: int) -> int: 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) diff --git a/enodia_sentinel/config.py b/enodia_sentinel/config.py index 388e60e..568e9e4 100644 --- a/enodia_sentinel/config.py +++ b/enodia_sentinel/config.py @@ -71,6 +71,17 @@ class Config: pkgdb_verify: bool = True # detect out-of-band package-DB edits pkgdb_interval: int = 600 # seconds between DB integrity checks heartbeat_max_age: int = 120 # heartbeat older than this = stale + # Layer 2 — verify on-disk files against the signed cache package (survives + # a rewritten checksum DB). Expensive, so it samples a rotating slice of + # packages each pass; off by default (needs the package cache populated). + pkgdb_pkgverify: bool = False + pkgdb_pkgverify_interval: int = 21600 # seconds between Layer-2 passes + pkgdb_pkgverify_sample: int = 40 # packages verified per pass (rotates) + + # anti-rootkit (cross-view: ask the same question two ways, compare) + rootcheck_enabled: bool = True + rootcheck_interval: int = 300 # seconds between cross-view sweeps + rootcheck_pid_cap: int = 65536 # upper PID to brute-force via kill(0) # eBPF on-ramp capture_execve_bpftrace: bool = False diff --git a/enodia_sentinel/daemon.py b/enodia_sentinel/daemon.py index 26f27f9..14865a8 100644 --- a/enodia_sentinel/daemon.py +++ b/enodia_sentinel/daemon.py @@ -43,6 +43,13 @@ class Sentinel: self._fim_pkg_thread: threading.Thread | None = None self._last_fim_pkg = 0.0 self._last_pkgdb = 0.0 + # Layer-2 package verification (rotating sample each pass). + self._pkgverify_thread: threading.Thread | None = None + self._last_pkgverify = 0.0 + self._pkgverify_offset = 0 + # Anti-rootkit cross-view sweep (off the loop thread; brute-forces PIDs). + self._rootcheck_thread: threading.Thread | None = None + self._last_rootcheck = 0.0 self._stop = threading.Event() # -- baselines --------------------------------------------------------- @@ -158,6 +165,44 @@ class Sentinel: for alert in pacman_verify_alerts(): self._on_exec_alert(alert) + # -- package signature verification (Layer 2, off the loop thread) ------ + def _maybe_pkgdb_verify(self, now: float) -> None: + if not self.cfg.pkgdb_pkgverify: + return + if self._pkgverify_thread and self._pkgverify_thread.is_alive(): + return + if (now - self._last_pkgverify) < self.cfg.pkgdb_pkgverify_interval: + return + self._last_pkgverify = now + self._pkgverify_thread = threading.Thread( + target=self._pkgdb_verify, daemon=True) + self._pkgverify_thread.start() + + def _pkgdb_verify(self) -> None: + from . import pkgdb + for alert in pkgdb.verify_alerts(self.cfg, offset=self._pkgverify_offset): + self._on_exec_alert(alert) + # Advance the rotating window so the next pass covers a different slice. + self._pkgverify_offset += self.cfg.pkgdb_pkgverify_sample + + # -- anti-rootkit cross-view (off the loop thread) --------------------- + def _maybe_rootcheck(self, now: float) -> None: + if not self.cfg.rootcheck_enabled: + return + if self._rootcheck_thread and self._rootcheck_thread.is_alive(): + return + if (now - self._last_rootcheck) < self.cfg.rootcheck_interval: + return + self._last_rootcheck = now + self._rootcheck_thread = threading.Thread( + target=self._rootcheck, daemon=True) + self._rootcheck_thread.start() + + def _rootcheck(self) -> None: + from . import rootcheck + for alert in rootcheck.run(self.cfg): + self._on_exec_alert(alert) + # -- one sweep --------------------------------------------------------- def sweep(self, *, force_suid: bool = False) -> list[Alert]: now = time.time() @@ -221,6 +266,8 @@ class Sentinel: self._maybe_scan_fim(now) self._maybe_pkg_verify(now) self._maybe_check_pkgdb(now) + self._maybe_pkgdb_verify(now) + self._maybe_rootcheck(now) alerts = self.sweep() fresh = self.fresh_alerts(alerts, now) if fresh: diff --git a/enodia_sentinel/pkgdb.py b/enodia_sentinel/pkgdb.py index b6e4752..14187c2 100644 --- a/enodia_sentinel/pkgdb.py +++ b/enodia_sentinel/pkgdb.py @@ -18,10 +18,12 @@ off-box and verification against signed packages (see README hardening notes). """ from __future__ import annotations +import gzip import hashlib import json import os import re +import subprocess from datetime import datetime from .alert import Alert, Severity @@ -29,7 +31,12 @@ from .config import Config DB_DIR = "/var/lib/pacman/local" PACMAN_LOG = "/var/log/pacman.log" +PACMAN_CONF = "/etc/pacman.conf" +KEYRING_DIR = "/etc/pacman.d/gnupg" +PKG_CACHE = "/var/cache/pacman/pkg" SID_PKGDB = 100021 +SID_SIGLEVEL = 100026 +SID_PKGVERIFY = 100027 _SLOP = 120 # seconds of clock slop allowed between a transaction and the anchor _TS_RE = re.compile(r"^\[([0-9T:+\-]+)\]") @@ -130,3 +137,206 @@ def check(cfg: Config, sid=SID_PKGDB, classtype="anti-tamper", ) + + +# ========================================================================== +# Layer 2 — verification against signed packages (the independent anchor) +# ========================================================================== +# Layer 1 detects an *out-of-band* DB edit. But an attacker with root can edit +# a binary, rewrite its hash in the local DB, AND run `fim-update` to re-anchor. +# The only reference they cannot forge is the distro's signing key: packages and +# repo databases are signed. So we verify on-disk files against the .MTREE inside +# the *cached package* — independent of the local DB the attacker controls. + +def parse_global_siglevel(conf_text: str) -> str: + """Return the [options] SigLevel value from pacman.conf text ('' if unset).""" + section = "" + value = "" + for line in conf_text.splitlines(): + line = line.split("#", 1)[0].strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1].strip().lower() + elif section == "options" and line.lower().startswith("siglevel"): + value = line.split("=", 1)[1].strip() if "=" in line else "" + return value + + +def siglevel_insecure(value: str) -> bool: + """True if signature verification is effectively disabled.""" + tokens = value.replace(",", " ").split() + return any(t in ("Never", "TrustAll") for t in tokens) + + +def siglevel_alert(conf_path: str = PACMAN_CONF) -> Alert | None: + try: + text = open(conf_path, encoding="utf-8", errors="replace").read() + except OSError: + return None + value = parse_global_siglevel(text) + # pacman's compiled-in default ("Required DatabaseOptional") is secure, so an + # unset SigLevel is fine; only an explicit downgrade is suspicious. + if value and siglevel_insecure(value): + return Alert( + severity=Severity.CRITICAL, signature="pacman_siglevel_disabled", + key="pkgdb:siglevel", + detail=(f"pacman SigLevel is insecure ('{value}') — package signature " + "verification is disabled, allowing unsigned/forged packages"), + sid=SID_SIGLEVEL, classtype="anti-tamper") + return None + + +def keyring_present(keyring_dir: str = KEYRING_DIR) -> bool: + return os.path.isfile(os.path.join(keyring_dir, "pubring.gpg")) + + +# --- mtree extraction & comparison ---------------------------------------- + +_MTREE_SHA = re.compile(r"sha256digest=([0-9a-f]{64})") + + +def parse_mtree(text: str) -> dict[str, str]: + """Parse a pacman .MTREE into {absolute_path: sha256} for regular files. + + pacman emits a `/set type=file …` default and then bare path entries that + *omit* `type=` for regular files (only dirs/links carry an explicit type). + So a file is "has a sha256digest and isn't explicitly a non-file type" — + we track the `/set` default and honor per-line overrides. + """ + out: dict[str, str] = {} + set_type = "file" # pacman's /set default; updated by /set lines + for line in text.splitlines(): + line = line.strip() + if line.startswith("/set "): + for tok in line.split()[1:]: + if tok.startswith("type="): + set_type = tok[len("type="):] + continue + if not line.startswith("./"): + continue + # explicit per-line type overrides the /set default + line_type = set_type + for tok in line.split()[1:]: + if tok.startswith("type="): + line_type = tok[len("type="):] + if line_type != "file": + continue + m = _MTREE_SHA.search(line) + if not m: + continue + rel = line.split(None, 1)[0][1:] # './usr/bin/x' -> '/usr/bin/x' + rel = rel.replace("\\040", " ") # mtree space escape + out[rel] = m.group(1) + return out + + +def package_mtree(pkg_path: str) -> dict[str, str]: + """Extract and parse the .MTREE from a cached package archive (via bsdtar).""" + try: + raw = subprocess.run(["bsdtar", "-xOqf", pkg_path, ".MTREE"], + capture_output=True, timeout=60).stdout + text = gzip.decompress(raw).decode("utf-8", "replace") + except (OSError, subprocess.SubprocessError, gzip.BadGzipFile, ValueError): + return {} + return parse_mtree(text) + + +def find_cached_package(name: str, version: str, + cache: str = PKG_CACHE) -> str | None: + import glob + for ext in ("zst", "xz"): + hits = glob.glob(os.path.join(cache, f"{name}-{version}-*.pkg.tar.{ext}")) + if hits: + return sorted(hits)[-1] + return None + + +def verify_package(name: str, version: str, root: str = "/", + cache: str = PKG_CACHE) -> dict: + """Compare a package's on-disk files to the hashes in its signed cache pkg. + + Returns {'pkg', 'cached', 'checked', 'mismatched':[(path, on_disk, expected)]}. + A mismatch means the on-disk file differs from what the distro shipped — + independent of the local DB, so it survives a rewritten checksum database. + """ + from .fim import hash_file + result = {"pkg": f"{name}-{version}", "cached": None, + "checked": 0, "mismatched": []} + pkg = find_cached_package(name, version, cache) + if not pkg: + return result + result["cached"] = pkg + for path, expected in package_mtree(pkg).items(): + full = os.path.join(root, path.lstrip("/")) + if not os.path.isfile(full): + continue + result["checked"] += 1 + actual = hash_file(full) + if actual and actual != expected: + result["mismatched"].append((path, actual, expected)) + return result + + +def installed_packages() -> list[tuple[str, str]]: + """(name, version) for every installed package, via `pacman -Q`.""" + try: + out = subprocess.run(["pacman", "-Q"], capture_output=True, + text=True, timeout=30).stdout + except (OSError, subprocess.SubprocessError): + return [] + pkgs = [] + for line in out.splitlines(): + parts = line.split() + if len(parts) == 2: + pkgs.append((parts[0], parts[1])) + return pkgs + + +# --- the Layer-2 check ---------------------------------------------------- +# Verifying every cached package on each pass is expensive (untar + hash every +# file), so this runs on its own slow cadence and verifies a rotating *sample* +# of packages per pass. Over enough passes the whole installed set is covered, +# and any single mismatch fires immediately — an attacker can't know which slice +# we'll check next. + +def _sample(items: list, n: int, offset: int) -> list: + """A rotating window of `n` items starting at `offset` (wraps around).""" + if n <= 0 or not items: + return items + n = min(n, len(items)) + start = offset % len(items) + return [items[(start + i) % len(items)] for i in range(n)] + + +def verify_alerts(cfg: Config, *, + _installed=installed_packages, + _verify=verify_package, + offset: int = 0) -> list[Alert]: + """Layer 2: flag config that disables signing, and on-disk files that + diverge from the distro's *signed* cache package. + + A mismatch here survives a rewritten local checksum DB — the reference is + the maintainer's signature, not the mutable DB the attacker controls. The + cache package itself is only trusted because pacman verified its signature + on download (so we also flag SigLevel downgrades that would skip that). + """ + alerts: list[Alert] = [] + sl = siglevel_alert() + if sl: + alerts.append(sl) + + pkgs = _installed() + sample = _sample(pkgs, cfg.pkgdb_pkgverify_sample, offset) + for name, version in sample: + res = _verify(name, version) + if res["mismatched"]: + paths = ", ".join(p for p, _a, _e in res["mismatched"][:8]) + alerts.append(Alert( + severity=Severity.CRITICAL, + signature="pkg_signature_mismatch", + key=f"pkgverify:{name}", + detail=(f"on-disk files of package '{res['pkg']}' differ from the " + f"signed cache package — possible trojaned binary that a " + f"rewritten checksum DB would hide: {paths}"), + sid=SID_PKGVERIFY, + classtype="anti-tamper")) + return alerts diff --git a/enodia_sentinel/rootcheck.py b/enodia_sentinel/rootcheck.py new file mode 100644 index 0000000..e78059b --- /dev/null +++ b/enodia_sentinel/rootcheck.py @@ -0,0 +1,209 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Anti-rootkit: detect things that are being hidden from the normal views. + +The technique a rootkit uses to hide is also how you catch it: ask the same +question two different ways and compare the answers. A process the ``/proc`` +listing omits but the scheduler still knows about; a loaded module absent from +``/proc/modules`` but live in ``/sys/module``; a listening port in +``/proc/net/tcp`` that ``ss`` won't show — each discrepancy is a hiding artifact. + +Honest limits: this runs in user space, so it reliably catches userland +(LD_PRELOAD) rootkits and the common ``/proc``-hiding LKMs, but a kernel rootkit +that hooks *every* path (the scheduler, /proc, and /sys consistently) can still +evade it. It raises the bar and catches the common cases; it is not a guarantee +against a bespoke ring-0 implant. Pair it with the off-box dead-man's switch. +""" +from __future__ import annotations + +import os +import re +import subprocess +from collections.abc import Iterator + +from .alert import Alert, Severity +from .config import Config +from .system import SystemState + +SID_HIDDEN_PROC = 100022 +SID_HIDDEN_MODULE = 100023 +SID_HIDDEN_PORT = 100024 +SID_PROMISC = 100025 + +IFF_PROMISC = 0x100 + + +# --- pure diff cores (unit-tested without a live system) ------------------ + +def find_hidden_pids(proc_pids: set[int], alive_pids: set[int]) -> set[int]: + """PIDs the kernel confirms alive but that are missing from /proc.""" + return alive_pids - proc_pids + + +def find_hidden_modules(proc_modules: set[str], sys_live: set[str]) -> set[str]: + """Modules live in /sys/module but absent from /proc/modules.""" + return sys_live - proc_modules + + +def find_hidden_ports(procnet_ports: set[int], ss_ports: set[int]) -> set[int]: + """Listening ports in /proc/net/tcp that ss does not report.""" + return procnet_ports - ss_ports + + +# --- system views --------------------------------------------------------- + +def proc_pids() -> set[int]: + return {int(n) for n in os.listdir("/proc") if n.isdigit()} + + +def pid_max(cap: int) -> int: + try: + return min(int(open("/proc/sys/kernel/pid_max").read().strip()), cap) + except (OSError, ValueError): + return cap + + +def alive_pids(cap: int) -> set[int]: + """Brute-force which PIDs exist via kill(pid, 0). + + kill returns success or EPERM for a live PID, ESRCH (ProcessLookupError) + for a dead one. Goes through the process table, not the /proc readdir a + rootkit may have hooked. + """ + alive: set[int] = set() + for pid in range(1, pid_max(cap) + 1): + try: + os.kill(pid, 0) + alive.add(pid) + except PermissionError: + alive.add(pid) # exists, just not ours to signal + except (ProcessLookupError, OSError): + pass + return alive + + +def proc_modules() -> set[str]: + out: set[str] = set() + try: + with open("/proc/modules") as fh: + for line in fh: + out.add(line.split()[0]) + except OSError: + pass + return out + + +def sys_live_modules() -> set[str]: + """Modules under /sys/module whose initstate is 'live' (i.e. loaded, not + built-in — built-ins have no initstate file).""" + out: set[str] = set() + base = "/sys/module" + try: + for m in os.listdir(base): + try: + with open(os.path.join(base, m, "initstate")) as fh: + if fh.read().strip() == "live": + out.add(m) + except OSError: + continue + except OSError: + pass + return out + + +def _procnet_listen_ports(path: str) -> set[int]: + ports: set[int] = set() + try: + with open(path) as fh: + next(fh, None) # header + for line in fh: + parts = line.split() + if len(parts) < 4 or parts[3] != "0A": # 0A = TCP_LISTEN + continue + local = parts[1] # HEX_ADDR:HEX_PORT + ports.add(int(local.rsplit(":", 1)[1], 16)) + except OSError: + pass + return ports + + +def procnet_listen_ports() -> set[int]: + return _procnet_listen_ports("/proc/net/tcp") | _procnet_listen_ports("/proc/net/tcp6") + + +def ss_listen_ports(state: SystemState) -> set[int]: + ports: set[int] = set() + for s in state.listening_sockets(): + tail = s.local.rsplit(":", 1) + if len(tail) == 2 and tail[1].isdigit(): + ports.add(int(tail[1])) + return ports + + +def promiscuous_interfaces() -> list[str]: + out: list[str] = [] + base = "/sys/class/net" + try: + for iface in os.listdir(base): + try: + with open(os.path.join(base, iface, "flags")) as fh: + flags = int(fh.read().strip(), 16) + if flags & IFF_PROMISC: + out.append(iface) + except (OSError, ValueError): + continue + except OSError: + pass + return out + + +# --- the check ------------------------------------------------------------ + +def run(cfg: Config, state: SystemState | None = None) -> Iterator[Alert]: + state = state or SystemState() + cap = cfg.rootcheck_pid_cap + + # Hidden processes — re-verify each candidate to shake out exit races. + visible = proc_pids() + candidates = find_hidden_pids(visible, alive_pids(cap)) + confirmed = [] + for pid in candidates: + try: + os.kill(pid, 0) + except PermissionError: + pass + except OSError: + continue + if not os.path.exists(f"/proc/{pid}"): + confirmed.append(pid) + if confirmed: + yield Alert( + severity=Severity.CRITICAL, signature="rootkit_hidden_process", + key=f"rk:hidproc:{min(confirmed)}", + detail=("process(es) alive but hidden from /proc: " + + ", ".join(map(str, sorted(confirmed)[:20]))), + pids=tuple(sorted(confirmed)[:20]), + sid=SID_HIDDEN_PROC, classtype="rootkit-hidden-process") + + hidden_mods = find_hidden_modules(proc_modules(), sys_live_modules()) + for m in sorted(hidden_mods): + yield Alert( + severity=Severity.CRITICAL, signature="rootkit_hidden_module", + key=f"rk:hidmod:{m}", + detail=f"kernel module live in /sys/module but hidden from /proc/modules: {m}", + sid=SID_HIDDEN_MODULE, classtype="rootkit-hidden-module") + + hidden_ports = find_hidden_ports(procnet_listen_ports(), ss_listen_ports(state)) + if hidden_ports: + yield Alert( + severity=Severity.HIGH, signature="rootkit_hidden_port", + key=f"rk:hidport:{min(hidden_ports)}", + detail=("listening port(s) in /proc/net/tcp not reported by ss " + "(tool may be hooked): " + ", ".join(map(str, sorted(hidden_ports)))), + sid=SID_HIDDEN_PORT, classtype="rootkit-hidden-port") + + for iface in promiscuous_interfaces(): + yield Alert( + severity=Severity.MEDIUM, signature="promiscuous_interface", + key=f"rk:promisc:{iface}", + detail=f"interface in promiscuous mode (possible sniffer): {iface}", + sid=SID_PROMISC, classtype="network-sniffer") diff --git a/pyproject.toml b/pyproject.toml index 9ea5f9f..56074e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "enodia-sentinel" -version = "0.6.0" +version = "0.7.0" description = "Host intrusion-detection daemon — signature-based detection of reverse shells, LD_PRELOAD rootkits, fileless malware, and persistence tampering" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_rootcheck.py b/tests/test_rootcheck.py new file mode 100644 index 0000000..d6950b9 --- /dev/null +++ b/tests/test_rootcheck.py @@ -0,0 +1,89 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Anti-rootkit cross-view tests — the pure diff cores plus a run() integration +test that injects each system view, so it needs no live /proc, /sys, or ss.""" +import unittest + +from enodia_sentinel import rootcheck +from enodia_sentinel.alert import Severity +from enodia_sentinel.config import Config +from enodia_sentinel.system import Socket, SystemState + + +class TestDiffCores(unittest.TestCase): + def test_hidden_pids_are_alive_minus_visible(self): + # 1337 is alive (kernel knows it) but absent from the /proc listing. + self.assertEqual( + rootcheck.find_hidden_pids({1, 2, 3}, {1, 2, 3, 1337}), {1337}) + + def test_no_hidden_pids_when_views_agree(self): + self.assertEqual(rootcheck.find_hidden_pids({1, 2}, {1, 2}), set()) + + def test_hidden_modules_live_in_sys_but_not_proc(self): + self.assertEqual( + rootcheck.find_hidden_modules({"ext4", "nf_tables"}, + {"ext4", "nf_tables", "evil_rk"}), + {"evil_rk"}) + + def test_hidden_ports_in_procnet_not_in_ss(self): + self.assertEqual( + rootcheck.find_hidden_ports({22, 80, 31337}, {22, 80}), {31337}) + + +class TestSsPortsFromState(unittest.TestCase): + def test_extracts_listening_ports_from_injected_state(self): + socks = [ + Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 10, "sshd", 1), + Socket("LISTEN", "127.0.0.1:631", "0.0.0.0:0", 11, "cupsd", 2), + ] + state = SystemState(sockets=socks) + self.assertEqual(rootcheck.ss_listen_ports(state), {22, 631}) + + +class TestRunIntegration(unittest.TestCase): + """Drive run() with every system view monkeypatched to a known state.""" + + def setUp(self): + self.cfg = Config() + self._saved = {} + for name in ("proc_pids", "alive_pids", "proc_modules", + "sys_live_modules", "procnet_listen_ports", + "promiscuous_interfaces"): + self._saved[name] = getattr(rootcheck, name) + + def tearDown(self): + for name, fn in self._saved.items(): + setattr(rootcheck, name, fn) + + def _patch(self, **views): + rootcheck.proc_pids = lambda: views.get("proc_pids", set()) + rootcheck.alive_pids = lambda cap: views.get("alive_pids", set()) + rootcheck.proc_modules = lambda: views.get("proc_modules", set()) + rootcheck.sys_live_modules = lambda: views.get("sys_live_modules", set()) + rootcheck.procnet_listen_ports = lambda: views.get("procnet_ports", set()) + rootcheck.promiscuous_interfaces = lambda: views.get("promisc", []) + + def test_clean_system_yields_nothing(self): + self._patch(proc_pids={1, 2}, alive_pids={1, 2}, + proc_modules={"ext4"}, sys_live_modules={"ext4"}) + state = SystemState(sockets=[]) + self.assertEqual(list(rootcheck.run(self.cfg, state)), []) + + def test_hidden_module_and_port_and_promisc(self): + self._patch( + proc_pids={1}, alive_pids={1}, + proc_modules={"ext4"}, sys_live_modules={"ext4", "diamorphine"}, + procnet_ports={22, 31337}, promisc=["eth0"]) + state = SystemState(sockets=[ + Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 10, "sshd", 1)]) + alerts = {a.signature: a for a in rootcheck.run(self.cfg, state)} + self.assertIn("rootkit_hidden_module", alerts) + self.assertEqual(alerts["rootkit_hidden_module"].severity, Severity.CRITICAL) + self.assertIn("diamorphine", alerts["rootkit_hidden_module"].detail) + self.assertIn("rootkit_hidden_port", alerts) + self.assertIn("31337", alerts["rootkit_hidden_port"].detail) + self.assertIn("promiscuous_interface", alerts) + self.assertEqual(alerts["promiscuous_interface"].sid, rootcheck.SID_PROMISC) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tamper.py b/tests/test_tamper.py index 3ab3c6c..f983161 100644 --- a/tests/test_tamper.py +++ b/tests/test_tamper.py @@ -73,6 +73,92 @@ class TestPacmanLogParse(unittest.TestCase): Path(path).unlink() +class TestSigLevel(unittest.TestCase): + def test_insecure_tokens_detected(self): + self.assertTrue(pkgdb.siglevel_insecure("Never")) + self.assertTrue(pkgdb.siglevel_insecure("Optional TrustAll")) + self.assertFalse(pkgdb.siglevel_insecure("Required DatabaseOptional")) + + def test_parse_global_only_reads_options_section(self): + conf = ( + "[options]\n" + "SigLevel = Required DatabaseOptional\n" + "[core]\n" + "SigLevel = Never\n" # a repo override must not be read as global + ) + self.assertEqual(pkgdb.parse_global_siglevel(conf), + "Required DatabaseOptional") + + def test_parse_ignores_comments(self): + conf = "[options]\nSigLevel = Never # was: Required\n" + self.assertEqual(pkgdb.parse_global_siglevel(conf), "Never") + + +class TestMtreeParse(unittest.TestCase): + def test_parses_regular_files_with_sha(self): + # Mirrors real pacman output: a /set default + bare file entries with no + # explicit type=, and dirs that carry type=dir. + mtree = ( + "#mtree\n" + "/set type=file uid=0 gid=0 mode=644\n" + "./usr/bin/ssh time=0.0 size=10 " + "sha256digest=" + "a" * 64 + "\n" + "./usr/lib time=0.0 type=dir\n" # dirs are skipped + "./etc/x\\040y time=0.0 " + "sha256digest=" + "b" * 64 + "\n" + ) + out = pkgdb.parse_mtree(mtree) + self.assertEqual(out["/usr/bin/ssh"], "a" * 64) + self.assertEqual(out["/etc/x y"], "b" * 64) # \040 -> space + self.assertNotIn("/usr/lib", out) + + +class TestSampleRotation(unittest.TestCase): + def test_window_wraps_and_covers_all_over_passes(self): + items = list(range(10)) + seen = set() + for off in range(0, 10, 3): + seen.update(pkgdb._sample(items, 3, off)) + self.assertEqual(seen, set(items)) + + def test_sample_larger_than_list_clamps(self): + self.assertEqual(set(pkgdb._sample([1, 2], 99, 0)), {1, 2}) + + +class TestVerifyAlerts(unittest.TestCase): + def setUp(self): + self.d = tempfile.TemporaryDirectory() + self.cfg = cfg_with_dir(Path(self.d.name)) + self.cfg.pkgdb_pkgverify_sample = 10 + + def tearDown(self): + self.d.cleanup() + + def test_mismatch_yields_critical_alert(self): + def fake_verify(name, version): + return {"pkg": f"{name}-{version}", "cached": "/c.pkg", "checked": 1, + "mismatched": [("/usr/bin/ssh", "dead", "beef")]} + alerts = pkgdb.verify_alerts( + self.cfg, _installed=lambda: [("openssh", "9.0")], + _verify=fake_verify) + sigs = [a.signature for a in alerts] + self.assertIn("pkg_signature_mismatch", sigs) + a = next(x for x in alerts if x.signature == "pkg_signature_mismatch") + self.assertEqual(a.severity, Severity.CRITICAL) + self.assertEqual(a.sid, pkgdb.SID_PKGVERIFY) + self.assertIn("/usr/bin/ssh", a.detail) + + def test_all_match_is_silent(self): + clean = lambda name, version: { + "pkg": f"{name}-{version}", "cached": "/c", "checked": 3, + "mismatched": []} + alerts = pkgdb.verify_alerts( + self.cfg, _installed=lambda: [("a", "1"), ("b", "2")], + _verify=clean) + self.assertEqual( + [a for a in alerts if a.signature == "pkg_signature_mismatch"], []) + + class TestHeartbeatWatchdog(unittest.TestCase): def test_heartbeat_roundtrip(self): with tempfile.TemporaryDirectory() as d: