215 lines
10 KiB
Python
215 lines
10 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_ps_hidden_pids_are_proc_minus_ps(self):
|
|
self.assertEqual(rootcheck.find_ps_hidden_pids({1, 2, 3}, {1, 3}), {2})
|
|
|
|
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", "ps_pids", "proc_pid_exists",
|
|
"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.ps_pids = lambda: views.get("ps_pids", set())
|
|
rootcheck.proc_pid_exists = lambda pid: pid in views.get("existing_pids", 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},
|
|
ps_pids={1, 2}, existing_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},
|
|
ps_pids={1}, existing_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_process_visible_in_proc_but_missing_from_ps_alerts(self):
|
|
self._patch(
|
|
proc_pids={1, 4242}, alive_pids={1, 4242},
|
|
ps_pids={1}, existing_pids={1, 4242},
|
|
proc_modules={"ext4"}, sys_live_modules={"ext4"})
|
|
alerts = {a.signature: a for a in rootcheck.run(self.cfg, SystemState(sockets=[]))}
|
|
self.assertIn("rootkit_ps_hidden_process", alerts)
|
|
self.assertEqual(alerts["rootkit_ps_hidden_process"].sid,
|
|
rootcheck.SID_PS_HIDDEN_PROCESS)
|
|
self.assertEqual(alerts["rootkit_ps_hidden_process"].pids, (4242,))
|
|
|
|
def test_tainted_module_and_kernel_taint(self):
|
|
self._patch(
|
|
proc_pids={1}, alive_pids={1},
|
|
ps_pids={1}, existing_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},
|
|
ps_pids={1}, existing_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()
|