146 lines
6.6 KiB
Python
146 lines
6.6 KiB
Python
# 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_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})
|
|
|
|
|
|
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", "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.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},
|
|
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),
|
|
])
|
|
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("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()
|