Add memory obfuscation and ps-hidden process detection
This commit is contained in:
Luna
parent
cb334c0c94
commit
893409b549
17 changed files with 353 additions and 26 deletions
|
|
@ -7,10 +7,10 @@ from enodia_sentinel.alert import Severity
|
|||
from enodia_sentinel.config import Config
|
||||
from enodia_sentinel.detectors import (
|
||||
credential_access, deleted_exe, egress, input_snooper, ld_preload,
|
||||
new_listener, new_suid, reverse_shell,
|
||||
memory_obfuscation, new_listener, new_suid, reverse_shell,
|
||||
stealth_network,
|
||||
)
|
||||
from enodia_sentinel.system import Socket, SystemState
|
||||
from enodia_sentinel.system import MemoryMap, Socket, SystemState, parse_memory_maps
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -26,6 +26,7 @@ class FakeProc:
|
|||
uid: int = 0
|
||||
cwd: str = "/"
|
||||
fd_targets: dict = field(default_factory=dict)
|
||||
memory_maps: list = field(default_factory=list)
|
||||
|
||||
def stdio_socket_inode(self):
|
||||
return self._stdio_inode
|
||||
|
|
@ -158,6 +159,61 @@ class TestStealthNetwork(unittest.TestCase):
|
|||
self.assertEqual(list(stealth_network.detect(SystemState(sockets=[sock]), cfg())), [])
|
||||
|
||||
|
||||
class TestMemoryObfuscation(unittest.TestCase):
|
||||
def test_maps_parser_extracts_path(self):
|
||||
text = "7f00-8000 rwxp 00000000 00:00 0 /memfd:stage (deleted)\n"
|
||||
maps = parse_memory_maps(text)
|
||||
self.assertEqual(maps[0].start, 0x7f00)
|
||||
self.assertEqual(maps[0].end, 0x8000)
|
||||
self.assertEqual(maps[0].perms, "rwxp")
|
||||
self.assertEqual(maps[0].path, "/memfd:stage (deleted)")
|
||||
|
||||
def test_executable_memfd_alerts(self):
|
||||
proc = FakeProc(
|
||||
pid=350, comm="hoxha",
|
||||
memory_maps=[MemoryMap(0x1000, 0x2000, "r-xp", "/memfd:stage (deleted)")],
|
||||
)
|
||||
alerts = list(memory_obfuscation.detect(SystemState(processes=[proc]), cfg()))
|
||||
self.assertEqual(len(alerts), 1)
|
||||
self.assertEqual(alerts[0].signature, "memory_obfuscation")
|
||||
self.assertEqual(alerts[0].severity, Severity.CRITICAL)
|
||||
|
||||
def test_rwx_anonymous_alerts(self):
|
||||
proc = FakeProc(
|
||||
pid=351, comm="packer",
|
||||
memory_maps=[MemoryMap(0x3000, 0x4000, "rwxp", "")],
|
||||
)
|
||||
alerts = list(memory_obfuscation.detect(SystemState(processes=[proc]), cfg()))
|
||||
self.assertEqual(len(alerts), 1)
|
||||
self.assertEqual(alerts[0].severity, Severity.HIGH)
|
||||
|
||||
def test_process_hiding_library_alerts(self):
|
||||
proc = FakeProc(
|
||||
pid=352, comm="bash",
|
||||
memory_maps=[MemoryMap(0x5000, 0x6000, "r-xp", "/usr/lib/libhide.so")],
|
||||
)
|
||||
alerts = list(memory_obfuscation.detect(SystemState(processes=[proc]), cfg()))
|
||||
self.assertEqual(len(alerts), 1)
|
||||
self.assertEqual(alerts[0].signature, "process_hiding_library")
|
||||
self.assertEqual(alerts[0].severity, Severity.CRITICAL)
|
||||
|
||||
def test_jit_allowlisted_comm_ignored(self):
|
||||
proc = FakeProc(
|
||||
pid=353, comm="node",
|
||||
memory_maps=[MemoryMap(0x7000, 0x8000, "rwxp", "")],
|
||||
)
|
||||
self.assertEqual(list(memory_obfuscation.detect(SystemState(processes=[proc]), cfg())), [])
|
||||
|
||||
def test_hide_library_alerts_even_in_jit_allowlisted_comm(self):
|
||||
proc = FakeProc(
|
||||
pid=354, comm="node",
|
||||
memory_maps=[MemoryMap(0x9000, 0xa000, "r-xp", "/tmp/libhide.so")],
|
||||
)
|
||||
alerts = list(memory_obfuscation.detect(SystemState(processes=[proc]), cfg()))
|
||||
self.assertEqual(len(alerts), 1)
|
||||
self.assertEqual(alerts[0].signature, "process_hiding_library")
|
||||
|
||||
|
||||
class TestNewSuid(unittest.TestCase):
|
||||
def test_new_in_writable_is_critical(self):
|
||||
state = SystemState(
|
||||
|
|
|
|||
|
|
@ -18,6 +18,9 @@ class TestDiffCores(unittest.TestCase):
|
|||
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"},
|
||||
|
|
@ -108,7 +111,7 @@ class TestRunIntegration(unittest.TestCase):
|
|||
for name in ("proc_pids", "alive_pids", "proc_modules",
|
||||
"sys_live_modules", "procnet_listen_ports",
|
||||
"procnet_udp_ports", "procnet_raw_protocols",
|
||||
"procnet_protocol_kinds",
|
||||
"procnet_protocol_kinds", "ps_pids", "proc_pid_exists",
|
||||
"promiscuous_interfaces", "module_taints", "kernel_taint"):
|
||||
self._saved[name] = getattr(rootcheck, name)
|
||||
|
||||
|
|
@ -125,12 +128,15 @@ class TestRunIntegration(unittest.TestCase):
|
|||
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)), [])
|
||||
|
|
@ -138,6 +144,7 @@ class TestRunIntegration(unittest.TestCase):
|
|||
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},
|
||||
|
|
@ -167,9 +174,21 @@ class TestRunIntegration(unittest.TestCase):
|
|||
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"},
|
||||
|
|
@ -185,6 +204,7 @@ class TestRunIntegration(unittest.TestCase):
|
|||
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=[]))}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue