enodia-sentinal/tests/test_tamper.py

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