Add signed-package verification and anti-rootkit cross-view (v0.7)

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 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-06-01 05:42:57 -07:00
parent 34bc09041a
commit 1de5e86fed
10 changed files with 766 additions and 5 deletions

89
tests/test_rootcheck.py Normal file
View file

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

View file

@ -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: