From 34bc09041af04970e9ecb57b47f0d1fdf854f183 Mon Sep 17 00:00:00 2001 From: Luna Date: Sun, 31 May 2026 22:27:50 -0700 Subject: [PATCH] Add tamper-evidence: package-DB integrity, self-integrity, dead-man's switch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers two hard questions: "what guards the hashes FIM trusts?" and "how do we make the sensor itself hard to silently disable?" — within the honest limit that a root attacker who shares your privileges can't be fully stopped on-box, only made loud. - pkgdb.py: guards the package DB that `pacman -Qkk` trusts. Anchors a fingerprint of /var/lib/pacman/local (refreshed ONLY by the pacman hook) and cross-checks pacman.log; a DB change with no logged transaction is flagged pkgdb_tamper (CRITICAL, sid 100021) — catches an attacker rewriting a stored checksum to mask a modified binary. Verified end-to-end against a simulated hash-overwrite. - selfprotect.py: Sentinel's own binaries/config/units/hook are always in the FIM watch set (self-integrity); a heartbeat is written each loop; an external `watchdog` command polls a remote dashboard and pushes if the sensor goes silent or unreachable — silence becomes the alarm. - daemon: heartbeat + slow-cadence pkgdb check; DB anchor re-anchored in build_fim_baseline so the pacman hook refreshes it after each transaction - web: /api/status now reports heartbeat_age/stale; dashboard shows it - cli: pkgdb-check, watchdog (--url/--token/--max-age, bypasses min-severity) - config: pkgdb_verify/_interval, heartbeat_max_age - README: threat model (trust-anchor problem) + hardening layers (immutability, signed-package + external anchor); reframes "anti-rootkit" as tamper-evidence - tests: +8 (DB fingerprint, anchor/transaction logic, pacman.log parse, heartbeat + watchdog verdicts). 80/80 pass Co-Authored-By: Claude Opus 4.8 --- README.md | 50 ++++++++++ config/enodia-sentinel.toml | 11 +++ enodia_sentinel/__init__.py | 2 +- enodia_sentinel/cli.py | 36 +++++++ enodia_sentinel/config.py | 13 ++- enodia_sentinel/daemon.py | 22 +++++ enodia_sentinel/pkgdb.py | 132 ++++++++++++++++++++++++++ enodia_sentinel/selfprotect.py | 92 ++++++++++++++++++ enodia_sentinel/static/dashboard.html | 10 +- enodia_sentinel/web.py | 4 + packaging/PKGBUILD | 2 +- pyproject.toml | 2 +- tests/test_tamper.py | 98 +++++++++++++++++++ 13 files changed, 465 insertions(+), 9 deletions(-) create mode 100644 enodia_sentinel/pkgdb.py create mode 100644 enodia_sentinel/selfprotect.py create mode 100644 tests/test_tamper.py diff --git a/README.md b/README.md index 0df498e..cf150fa 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,52 @@ notify_ntfy_url = "https://ntfy.sh" notify_ntfy_topic = "enodia-7Hq2x" # keep this secret — it's the access control ``` +## Tamper-evidence & self-protection + +An attacker's first move against a sensor is to disable or blind it — and on a +box where they have root, *any* purely-local defense is ultimately defeatable +(they share your privileges). Sentinel doesn't pretend otherwise. The goal is to +make tampering **evident** by anchoring trust where the attacker has less +control, and to make going silent *loud*. + +**Package-DB integrity.** `pacman -Qkk` trusts the local package database — so a +root attacker can modify `/usr/bin/ssh` *and* rewrite its stored checksum, and +verification passes. Sentinel guards the DB itself: a legitimate change only +happens during a logged transaction, so it anchors a fingerprint of +`/var/lib/pacman/local` (refreshed *only* by the pacman hook) and cross-checks +`pacman.log`. A DB change with **no corresponding transaction** is flagged +`pkgdb_tamper` (CRITICAL, `sid 100021`) — directly catching the "overwrite the +hashes" attack. + +**Self-integrity.** Sentinel's own binaries, config, systemd units, and pacman +hook are always in the FIM watch set, so tampering with the watchdog trips the +watchdog. + +**Dead-man's switch.** The daemon writes a heartbeat every loop; the dashboard +exposes its age. Run an external watcher on another host (over Tailscale) and +*silence becomes the alarm*: + +```bash +# on a SEPARATE machine — alerts you if the sensor or the whole box goes dark +enodia-sentinel watchdog --url http://100.x.x.x:8787 --token --max-age 120 +``` + +**Hardening (the layers beyond on-box detection):** + +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. +3. **External anchor (next)** — mirror the DB/FIM fingerprints off-box so a + local attacker can't refresh the anchor to cover their tracks. + +> On "anti-rootkit": the robust version isn't stealth (fragile, and a genuine +> dual-use rootkit technique) — it's tamper-*evidence*. Don't hide; be +> un-hideable. The durable defenses move the trust anchor below or outside the +> attacker: signatures, external monitors, and ultimately the kernel (the eBPF +> roadmap) rather than userland the attacker can rewrite. + ## File integrity monitoring (Tripwire-style) Detects tampering with binaries and critical configs by **content hash**, so it @@ -349,6 +395,10 @@ regression suite for both. ## Project status +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. + v0.5 — adds **file integrity monitoring** (SHA-256 baseline + `pacman -Qkk` verification, auto-refreshed by a pacman hook) and **false-positive triage** via package-ownership provenance. diff --git a/config/enodia-sentinel.toml b/config/enodia-sentinel.toml index f0095eb..de97bab 100644 --- a/config/enodia-sentinel.toml +++ b/config/enodia-sentinel.toml @@ -67,6 +67,17 @@ fim_paths = [] # [] = built-in critical set; or your own list fim_pkg_verify = false # run `pacman -Qkk` (thorough but slow) fim_pkg_verify_interval = 21600 # seconds between package verifications +# --- tamper-evidence ----------------------------------------------------- +# Detect out-of-band edits to the pacman DB itself (an attacker rewriting a +# stored checksum to mask a modified binary). The DB anchor is refreshed only +# by the pacman hook, so a DB change with no logged transaction = tampering. +pkgdb_verify = true +pkgdb_interval = 600 # seconds between DB integrity checks +heartbeat_max_age = 120 # daemon heartbeat older than this = stale/silent +# Sentinel's own binaries/config/units/hook are always FIM-watched (self- +# integrity). For real teeth, make them immutable: chattr +i +# and run an external `enodia-sentinel watchdog` against this host's dashboard. + # --- eBPF event layer ---------------------------------------------------- # Event-driven execve monitor: catches short-lived processes the poll loop # misses, matched against the Snort-style rule engine. Requires python-bpfcc diff --git a/enodia_sentinel/__init__.py b/enodia_sentinel/__init__.py index e8d6ed3..e5af105 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.5.0" +__version__ = "0.6.0" diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index 7c5e2a8..d485854 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -68,6 +68,13 @@ def main(argv: list[str] | None = None) -> int: 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") + 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) @@ -93,9 +100,38 @@ def main(argv: list[str] | None = None) -> int: 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 == "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_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 ff41724..388e60e 100644 --- a/enodia_sentinel/config.py +++ b/enodia_sentinel/config.py @@ -67,6 +67,10 @@ class Config: fim_scan_interval: int = 300 # seconds between hash scans fim_pkg_verify: bool = False # run `pacman -Qkk` (slow; opt-in) fim_pkg_verify_interval: int = 21600 # seconds between package verifications + # tamper-evidence + 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 # eBPF on-ramp capture_execve_bpftrace: bool = False @@ -119,11 +123,12 @@ class Config: return self.log_dir / "fim-baseline.json" def fim_path_list(self) -> tuple[str, ...]: - """Configured FIM paths, defaulting to the built-in critical set.""" - if self.fim_paths: - return self.fim_paths + """FIM paths: the configured/critical set, always plus Sentinel's own + footprint so tampering with the watchdog trips the watchdog.""" from .fim import DEFAULT_FIM_PATHS - return DEFAULT_FIM_PATHS + from .selfprotect import SELF_PATHS + base = self.fim_paths or DEFAULT_FIM_PATHS + return tuple(base) + SELF_PATHS # ---- loading --------------------------------------------------------- @classmethod diff --git a/enodia_sentinel/daemon.py b/enodia_sentinel/daemon.py index 5bbeefb..26f27f9 100644 --- a/enodia_sentinel/daemon.py +++ b/enodia_sentinel/daemon.py @@ -42,6 +42,7 @@ class Sentinel: self._last_fim_scan = 0.0 self._fim_pkg_thread: threading.Thread | None = None self._last_fim_pkg = 0.0 + self._last_pkgdb = 0.0 self._stop = threading.Event() # -- baselines --------------------------------------------------------- @@ -90,12 +91,30 @@ class Sentinel: # -- FIM (off the loop thread) ----------------------------------------- def build_fim_baseline(self) -> int: + from . import pkgdb from .fim import scan_paths self.fim_baseline = scan_paths(self.cfg.fim_path_list()) self.cfg.log_dir.mkdir(parents=True, exist_ok=True) self.cfg.fim_baseline.write_text(json.dumps(self.fim_baseline)) + # Re-anchor the package DB here too, so the pacman hook (which calls + # fim-update) refreshes both after every legitimate transaction. + pkgdb.save_anchor(self.cfg) return len(self.fim_baseline) + def _maybe_check_pkgdb(self, now: float) -> None: + if not self.cfg.pkgdb_verify: + return + if (now - self._last_pkgdb) < self.cfg.pkgdb_interval: + return + self._last_pkgdb = now + threading.Thread(target=self._check_pkgdb, daemon=True).start() + + def _check_pkgdb(self) -> None: + from . import pkgdb + alert = pkgdb.check(self.cfg) + if alert: + self._on_exec_alert(alert) + def load_fim_baseline(self) -> None: try: self.fim_baseline = json.loads(self.cfg.fim_baseline.read_text()) @@ -195,10 +214,13 @@ class Sentinel: sweeps = 0 while not self._stop.is_set(): now = time.time() + from .selfprotect import write_heartbeat + write_heartbeat(self.cfg) # dead-man's switch if (now - self.start_time) >= self.cfg.baseline_grace: self._maybe_scan_suid(now) self._maybe_scan_fim(now) self._maybe_pkg_verify(now) + self._maybe_check_pkgdb(now) alerts = self.sweep() fresh = self.fresh_alerts(alerts, now) if fresh: diff --git a/enodia_sentinel/pkgdb.py b/enodia_sentinel/pkgdb.py new file mode 100644 index 0000000..b6e4752 --- /dev/null +++ b/enodia_sentinel/pkgdb.py @@ -0,0 +1,132 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Package-database integrity — guard the thing that FIM trusts. + +``pacman -Qkk`` verifies files against the *local package database*. A root +attacker can defeat that by editing a binary and rewriting its stored checksum +in the DB. So we also watch the DB itself. + +The defense: a legitimate change to ``/var/lib/pacman/local`` only ever happens +during a logged pacman transaction. We anchor a fingerprint of the DB, refreshed +*only* by the pacman PostTransaction hook, and cross-check ``pacman.log``. If the +DB fingerprint changes with **no corresponding transaction**, the checksum +database may have been tampered with to mask file modification — which is exactly +the attack of "overwriting the hashes." + +This is layer 1 (out-of-band detection). It does not stop an attacker who also +runs `fim-update` or forges a transaction; for that you want the anchor stored +off-box and verification against signed packages (see README hardening notes). +""" +from __future__ import annotations + +import hashlib +import json +import os +import re +from datetime import datetime + +from .alert import Alert, Severity +from .config import Config + +DB_DIR = "/var/lib/pacman/local" +PACMAN_LOG = "/var/log/pacman.log" +SID_PKGDB = 100021 +_SLOP = 120 # seconds of clock slop allowed between a transaction and the anchor + +_TS_RE = re.compile(r"^\[([0-9T:+\-]+)\]") + + +def db_fingerprint(db_dir: str = DB_DIR) -> str | None: + """Deterministic SHA-256 over every file in the local package DB.""" + h = hashlib.sha256() + try: + files = [] + for dp, _dn, fn in os.walk(db_dir): + files.extend(os.path.join(dp, f) for f in fn) + for p in sorted(files): + h.update(p.encode("utf-8", "replace") + b"\0") + with open(p, "rb") as fh: + while chunk := fh.read(1 << 16): + h.update(chunk) + except OSError: + return None + return h.hexdigest() + + +def last_transaction_time(log: str = PACMAN_LOG) -> float | None: + """Epoch of the most recent pacman transaction, from the log.""" + ts = None + try: + with open(log, encoding="utf-8", errors="replace") as fh: + for line in fh: + if "transaction completed" in line or "transaction started" in line: + m = _TS_RE.match(line) + if m: + ts = m.group(1) + except OSError: + return None + if not ts: + return None + try: + return datetime.fromisoformat(ts).timestamp() + except ValueError: + return None + + +def _anchor_path(cfg: Config): + return cfg.log_dir / "pkgdb-anchor.json" + + +def save_anchor(cfg: Config, fingerprint: str | None = None) -> None: + fp = fingerprint or db_fingerprint() + if fp is None: + return + cfg.log_dir.mkdir(parents=True, exist_ok=True) + _anchor_path(cfg).write_text(json.dumps({"fingerprint": fp, "time": _now()})) + + +def load_anchor(cfg: Config) -> dict | None: + try: + return json.loads(_anchor_path(cfg).read_text()) + except (OSError, ValueError): + return None + + +def _now() -> float: + import time + return time.time() + + +def check(cfg: Config, + _fingerprint=db_fingerprint, + _last_tx=last_transaction_time) -> Alert | None: + """Return a tamper Alert if the DB changed outside a logged transaction. + + Injectable fingerprint / last-transaction functions for testing. + """ + cur = _fingerprint() + if cur is None: + return None + anchor = load_anchor(cfg) + if anchor is None: + save_anchor(cfg, cur) # first run establishes the anchor + return None + if cur == anchor.get("fingerprint"): + return None # unchanged — fine + + # The DB changed. Legitimate only if a pacman transaction occurred after the + # anchor was taken (the hook normally refreshes the anchor in that case). + last_tx = _last_tx() + if last_tx is not None and last_tx >= anchor.get("time", 0) - _SLOP: + save_anchor(cfg, cur) # legitimate upgrade; re-anchor + return None + + return Alert( + severity=Severity.CRITICAL, + signature="pkgdb_tamper", + key="pkgdb:tamper", + detail=("pacman local DB changed with no corresponding transaction — " + "the package checksum database may have been altered to mask " + "file tampering"), + sid=SID_PKGDB, + classtype="anti-tamper", + ) diff --git a/enodia_sentinel/selfprotect.py b/enodia_sentinel/selfprotect.py new file mode 100644 index 0000000..4f71a10 --- /dev/null +++ b/enodia_sentinel/selfprotect.py @@ -0,0 +1,92 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Tamper-evidence for Sentinel itself. + +An attacker's first move against a security tool is to disable or blind it. We +can't make the agent un-killable on a box where the attacker is root — but we +can make killing or modifying it *loud*: + +* **self-integrity** — Sentinel's own binaries, config, units, and pacman hook + are added to the FIM watch set, so tampering with the watchdog trips the + watchdog. +* **dead-man's switch** — the daemon writes a heartbeat every loop; an external + watcher (on another host, over Tailscale) polls the dashboard and alerts if + the heartbeat goes stale or the dashboard becomes unreachable. Silence is the + alarm. + +The durable posture is tamper-*evidence*, not stealth: don't hide, be +un-hideable. For real teeth, make these files immutable (`chattr +i`) and keep +the heartbeat watcher on a separate machine. +""" +from __future__ import annotations + +import time + +from .config import Config + +# Sentinel's own on-disk footprint (both /usr and /usr/local installs); missing +# paths are simply skipped by the FIM scanner. +SELF_PATHS = ( + "/usr/bin/enodia-sentinel", "/usr/local/bin/enodia-sentinel", + "/usr/bin/sentinel-redteam", "/usr/local/bin/sentinel-redteam", + "/usr/lib/enodia-sentinel", "/usr/local/lib/enodia-sentinel", + "/etc/enodia-sentinel.toml", + "/etc/systemd/system/enodia-sentinel.service", + "/etc/systemd/system/enodia-sentinel-web.service", + "/usr/lib/systemd/system/enodia-sentinel.service", + "/usr/lib/systemd/system/enodia-sentinel-web.service", + "/etc/pacman.d/hooks/enodia-sentinel-fim.hook", + "/usr/share/libalpm/hooks/enodia-sentinel-fim.hook", +) + + +def heartbeat_path(cfg: Config): + return cfg.log_dir / "heartbeat" + + +def write_heartbeat(cfg: Config) -> None: + try: + heartbeat_path(cfg).write_text(str(int(time.time()))) + except OSError: + pass + + +def read_heartbeat(cfg: Config) -> int | None: + try: + return int(heartbeat_path(cfg).read_text().strip()) + except (OSError, ValueError): + return None + + +def heartbeat_age(cfg: Config) -> float | None: + hb = read_heartbeat(cfg) + return None if hb is None else max(0.0, time.time() - hb) + + +# --- external dead-man's-switch watcher ----------------------------------- + +def poll_status(url: str, token: str = "", timeout: int = 8) -> dict | None: + """Fetch a remote dashboard's /api/status. None if unreachable.""" + import json + import urllib.request + base = url.rstrip("/") + api = base if base.endswith("/api/status") else base + "/api/status" + req = urllib.request.Request(api) + if token: + req.add_header("Authorization", f"Bearer {token}") + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read()) + except Exception: + return None + + +def watchdog_verdict(status: dict | None, max_age: float) -> tuple[bool, str]: + """Return (ok, message). Runs on the *watcher* side, against poll_status().""" + if status is None: + return False, "Sentinel UNREACHABLE — dashboard down or host offline" + if not status.get("running"): + return False, "Sentinel daemon NOT running on the target host" + age = status.get("heartbeat_age") + if age is not None and age > max_age: + return False, f"Sentinel heartbeat STALE ({int(age)}s > {int(max_age)}s)" + return True, "ok" diff --git a/enodia_sentinel/static/dashboard.html b/enodia_sentinel/static/dashboard.html index 9981018..71b828d 100644 --- a/enodia_sentinel/static/dashboard.html +++ b/enodia_sentinel/static/dashboard.html @@ -100,8 +100,14 @@ async function refresh(){ document.getElementById("host").textContent = st.host || ""; document.getElementById("ver").textContent = "v"+st.version; document.getElementById("ebpf").textContent = "eBPF: "+(st.ebpf||"?"); - document.getElementById("status").textContent = - st.running ? "daemon running" : "daemon NOT running"; + let hb = ""; + if (st.heartbeat_age != null) + hb = st.heartbeat_stale + ? ` ⚠ heartbeat STALE (${Math.round(st.heartbeat_age)}s) — possible tampering` + : ` ♥ ${Math.round(st.heartbeat_age)}s ago`; + const st_el = document.getElementById("status"); + st_el.textContent = (st.running ? "daemon running" : "daemon NOT running") + hb; + st_el.style.color = (st.running && !st.heartbeat_stale) ? "" : "var(--crit)"; const c = st.counts||{}; const cards = document.getElementById("cards"); cards.innerHTML=""; for(const [lbl,key] of [["CRITICAL","crit"],["HIGH","high"],["MEDIUM","med"]]){ diff --git a/enodia_sentinel/web.py b/enodia_sentinel/web.py index fcf3cc1..16b9960 100644 --- a/enodia_sentinel/web.py +++ b/enodia_sentinel/web.py @@ -141,6 +141,8 @@ def daemon_status(cfg: Config) -> dict: if "eBPF exec monitor:" in line: ebpf = line.split("eBPF exec monitor:", 1)[1].strip() break + from .selfprotect import heartbeat_age + age = heartbeat_age(cfg) return { "version": __version__, "running": pid_alive, @@ -149,6 +151,8 @@ def daemon_status(cfg: Config) -> dict: "last_alert": alerts[0]["time"] if alerts else None, "ebpf": ebpf, "host": os.uname().nodename, + "heartbeat_age": age, + "heartbeat_stale": (age is not None and age > cfg.heartbeat_max_age), } diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD index 86cccfb..49e19c1 100644 --- a/packaging/PKGBUILD +++ b/packaging/PKGBUILD @@ -1,6 +1,6 @@ # Maintainer: Enodia pkgname=enodia-sentinel -pkgver=0.5.0 +pkgver=0.6.0 pkgrel=1 pkgdesc="Host intrusion-detection daemon — signature-based detection of reverse shells, LD_PRELOAD rootkits, fileless malware, and persistence tampering" arch=('any') diff --git a/pyproject.toml b/pyproject.toml index bdde5a9..9ea5f9f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "enodia-sentinel" -version = "0.5.0" +version = "0.6.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_tamper.py b/tests/test_tamper.py new file mode 100644 index 0000000..3ab3c6c --- /dev/null +++ b/tests/test_tamper.py @@ -0,0 +1,98 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Tests for package-DB tamper detection and the dead-man's-switch watchdog.""" +import tempfile +import time +import unittest +from pathlib import Path + +from enodia_sentinel import pkgdb, selfprotect +from enodia_sentinel.alert import Severity +from enodia_sentinel.config import Config + + +def cfg_with_dir(tmp: Path) -> Config: + c = Config() + c.log_dir = tmp + return c + + +class TestPkgdbFingerprint(unittest.TestCase): + def test_fingerprint_changes_with_content(self): + with tempfile.TemporaryDirectory() as d: + db = Path(d) + (db / "pkgA").mkdir() + (db / "pkgA" / "desc").write_text("sha256 = AAAA") + f1 = pkgdb.db_fingerprint(str(db)) + self.assertIsNotNone(f1) + (db / "pkgA" / "desc").write_text("sha256 = BBBB") # attacker rewrites hash + self.assertNotEqual(f1, pkgdb.db_fingerprint(str(db))) + + +class TestPkgdbCheck(unittest.TestCase): + def setUp(self): + self.d = tempfile.TemporaryDirectory() + self.cfg = cfg_with_dir(Path(self.d.name)) + + def tearDown(self): + self.d.cleanup() + + def test_first_run_establishes_anchor(self): + v = pkgdb.check(self.cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + self.assertIsNone(v) + self.assertEqual(pkgdb.load_anchor(self.cfg)["fingerprint"], "FP1") + + def test_unchanged_is_silent(self): + pkgdb.check(self.cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + v = pkgdb.check(self.cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + self.assertIsNone(v) + + def test_change_with_no_transaction_is_tamper(self): + pkgdb.check(self.cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + v = pkgdb.check(self.cfg, _fingerprint=lambda: "FP2", _last_tx=lambda: None) + self.assertIsNotNone(v) + self.assertEqual(v.severity, Severity.CRITICAL) + self.assertEqual(v.sid, pkgdb.SID_PKGDB) + + def test_change_with_recent_transaction_is_legit(self): + pkgdb.check(self.cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + future = time.time() + 10 # a transaction occurred after the anchor + v = pkgdb.check(self.cfg, _fingerprint=lambda: "FP2", _last_tx=lambda: future) + self.assertIsNone(v) # re-anchored, no alert + self.assertEqual(pkgdb.load_anchor(self.cfg)["fingerprint"], "FP2") + + +class TestPacmanLogParse(unittest.TestCase): + def test_last_transaction_time(self): + with tempfile.NamedTemporaryFile("w", suffix=".log", delete=False) as f: + f.write("[2026-05-30T10:00:00-0700] [ALPM] transaction completed\n") + f.write("[2026-05-31T12:34:56-0700] [ALPM] transaction completed\n") + f.write("[2026-05-31T12:35:00-0700] [ALPM] upgraded foo (1 -> 2)\n") + path = f.name + ts = pkgdb.last_transaction_time(path) + self.assertIsNotNone(ts) + Path(path).unlink() + + +class TestHeartbeatWatchdog(unittest.TestCase): + def test_heartbeat_roundtrip(self): + with tempfile.TemporaryDirectory() as d: + cfg = cfg_with_dir(Path(d)) + selfprotect.write_heartbeat(cfg) + self.assertLess(selfprotect.heartbeat_age(cfg), 5) + + def test_watchdog_verdicts(self): + ok, _ = selfprotect.watchdog_verdict({"running": True, "heartbeat_age": 5}, 120) + self.assertTrue(ok) + bad, msg = selfprotect.watchdog_verdict(None, 120) + self.assertFalse(bad) + self.assertIn("UNREACHABLE", msg) + bad2, _ = selfprotect.watchdog_verdict({"running": False}, 120) + self.assertFalse(bad2) + bad3, msg3 = selfprotect.watchdog_verdict( + {"running": True, "heartbeat_age": 9999}, 120) + self.assertFalse(bad3) + self.assertIn("STALE", msg3) + + +if __name__ == "__main__": + unittest.main()