# 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 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: 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()