# 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}) def test_hidden_udp_ports_in_procnet_not_in_ss(self): self.assertEqual( rootcheck.find_hidden_udp_ports({53, 5353, 4444}, {53, 5353}), {4444}) def test_hidden_raw_protocols_in_procnet_not_in_ss(self): self.assertEqual( rootcheck.find_hidden_raw_protocols({1, 58}, {58}), {1}) def test_hidden_protocol_kinds_in_procnet_not_in_ss(self): self.assertEqual( rootcheck.find_hidden_protocol_kinds({"sctp", "packet"}, {"packet"}), {"sctp"}) def test_known_rootkit_module_names_normalized(self): self.assertEqual( rootcheck.find_known_rootkit_modules({"ext4", "diamorphine", "adore-ng"}), {"diamorphine", "adore-ng"}) def test_kernel_taint_decoding_and_severity(self): value = (1 << 12) | (1 << 13) self.assertEqual(rootcheck.decode_kernel_taint(value), ["out-of-tree-module", "unsigned-module"]) self.assertEqual(rootcheck.taint_severity(value), Severity.HIGH) 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}) def test_extracts_udp_ports_from_injected_state(self): socks = [ Socket("UNCONN", "0.0.0.0:53", "0.0.0.0:0", 10, "dnsmasq", 1), Socket("ESTAB", "127.0.0.1:5353", "127.0.0.1:1", 11, "mdns", 2), Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 12, "sshd", 3), ] state = SystemState(sockets=socks) self.assertEqual(rootcheck.ss_udp_ports(state), {53, 5353}) def test_extracts_raw_protocols_from_injected_state(self): socks = [ Socket("UNCONN", "0.0.0.0:icmp", "0.0.0.0:*", 10, "hoxha", 1, "raw"), Socket("UNCONN", "[::]:ipv6-icmp", "[::]:*", 11, "monitor", 2, "raw"), Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 12, "sshd", 3, "tcp"), ] state = SystemState(sockets=socks) self.assertEqual(rootcheck.ss_raw_protocols(state), {1, 58}) def test_extracts_socket_kinds_from_injected_state(self): state = SystemState(sockets=[ Socket("ESTAB", "10.0.0.2:5000", "8.8.8.8:4443", 10, "h", 1, "sctp"), Socket("UNCONN", "*:eth0", "*", 11, "tcpdump", 2, "packet"), ]) self.assertEqual(rootcheck.ss_protocol_kinds(state), {"sctp", "packet"}) def test_ss_parser_handles_netid_column(self): text = 'p_raw UNCONN 0 0 *:eth0 * users:(("dhcpcd",pid=12,fd=7)) ino:123' sock = SystemState._parse_ss(text, kind="packet")[0] self.assertEqual(sock.state, "UNCONN") self.assertEqual(sock.local, "*:eth0") self.assertEqual(sock.peer, "*") self.assertEqual(sock.comm, "dhcpcd") self.assertEqual(sock.pid, 12) self.assertEqual(sock.inode, 123) self.assertEqual(sock.kind, "packet") 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", "procnet_udp_ports", "procnet_raw_protocols", "procnet_protocol_kinds", "promiscuous_interfaces", "module_taints", "kernel_taint"): 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.procnet_udp_ports = lambda: views.get("procnet_udp_ports", set()) rootcheck.procnet_raw_protocols = lambda: views.get("procnet_raw_protocols", set()) rootcheck.procnet_protocol_kinds = lambda: views.get("procnet_protocol_kinds", set()) rootcheck.promiscuous_interfaces = lambda: views.get("promisc", []) rootcheck.module_taints = lambda: views.get("module_taints", {}) rootcheck.kernel_taint = lambda: views.get("kernel_taint", 0) 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}, procnet_udp_ports={53, 4444}, procnet_raw_protocols={1, 58}, procnet_protocol_kinds={"sctp", "packet"}, promisc=["eth0"]) state = SystemState(sockets=[ Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 10, "sshd", 1), Socket("UNCONN", "0.0.0.0:53", "0.0.0.0:0", 11, "dnsmasq", 2), Socket("UNCONN", "0.0.0.0:ipv6-icmp", "0.0.0.0:*", 12, "monitor", 3, "raw"), Socket("UNCONN", "*:eth0", "*", 13, "tcpdump", 4, "packet"), ]) 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_known_module", alerts) self.assertIn("rootkit_hidden_port", alerts) self.assertIn("31337", alerts["rootkit_hidden_port"].detail) self.assertIn("rootkit_hidden_udp_port", alerts) self.assertIn("4444", alerts["rootkit_hidden_udp_port"].detail) self.assertIn("rootkit_hidden_raw_socket", alerts) self.assertIn("1", alerts["rootkit_hidden_raw_socket"].detail) self.assertIn("raw_icmp_socket", alerts) self.assertEqual(alerts["raw_icmp_socket"].sid, rootcheck.SID_RAW_ICMP_SOCKET) self.assertIn("rootkit_hidden_protocol_socket", alerts) self.assertIn("sctp", alerts["rootkit_hidden_protocol_socket"].detail) self.assertIn("promiscuous_interface", alerts) self.assertEqual(alerts["promiscuous_interface"].sid, rootcheck.SID_PROMISC) def test_tainted_module_and_kernel_taint(self): self._patch( proc_pids={1}, alive_pids={1}, proc_modules={"ext4", "vendor_gpu"}, sys_live_modules={"ext4", "vendor_gpu"}, module_taints={"vendor_gpu": "OE"}, kernel_taint=(1 << 12) | (1 << 13)) alerts = {a.signature: a for a in rootcheck.run(self.cfg, SystemState(sockets=[]))} self.assertIn("rootkit_tainted_module", alerts) self.assertEqual(alerts["rootkit_tainted_module"].sid, rootcheck.SID_TAINTED_MODULE) self.assertIn("kernel_tainted", alerts) self.assertEqual(alerts["kernel_tainted"].severity, Severity.HIGH) def test_module_allowlist_suppresses_tainted_module(self): self.cfg.rootcheck_module_allow = ("vendor_gpu",) self._patch( proc_pids={1}, alive_pids={1}, proc_modules={"vendor_gpu"}, sys_live_modules={"vendor_gpu"}, module_taints={"vendor_gpu": "OE"}) alerts = {a.signature: a for a in rootcheck.run(self.cfg, SystemState(sockets=[]))} self.assertNotIn("rootkit_tainted_module", alerts) if __name__ == "__main__": unittest.main()