Re-architect Sentinel as a zero-dependency Python package

Port the bash prototype to a structured, testable Python codebase while
preserving the same control loop and the seven detection signatures. The bash
implementation stays in src/ as the regression oracle.

Highlights:
- enodia_sentinel/ package (stdlib only — no runtime deps):
  - detectors/ : one pure function per signature, detect(state, cfg)->Alerts
  - system.py  : SystemState — one cached /proc + ss snapshot per sweep,
                 fully injectable so detectors are unit-testable
  - daemon.py  : sweep loop, in-process cooldown dedup, threaded snapshot
                 capture, and a SUID filesystem scan moved OFF the loop
                 thread onto a slow background cadence
  - snapshot.py: forensic text + JSON sidecar with per-signature IR guidance
  - config.py  : dataclass config via TOML + env overrides
  - netutil.py : public-IP / CIDR logic via stdlib ipaddress
- tests/ : 25 stdlib-unittest cases (no root, no /proc, no ss needed)
- TOML config, launcher wrapper, Makefile (pip-free install), hardened
  systemd unit (env-resolved ExecStart), updated PKGBUILD, rewritten README

Performance: per-sweep cost ~200 ms (shared cached state); the multi-second
SUID walk no longer blocks detection. scandir-based walk replaces os.walk.

Verified on Arch: all 7 detectors fire on red-team drills (reverse_shell,
ld_preload, deleted_exe, new_listener, new_suid confirmed live end-to-end),
no false positives on a clean sweep, 25/25 tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-05-31 01:50:50 -07:00
parent 45f8acb24a
commit 28d67a1360
28 changed files with 1783 additions and 133 deletions

152
tests/test_detectors.py Normal file
View file

@ -0,0 +1,152 @@
"""Detector unit tests using injected fake state — no /proc, no root, no ss."""
import unittest
from dataclasses import dataclass, field
from enodia_sentinel.alert import Severity
from enodia_sentinel.config import Config
from enodia_sentinel.detectors import (
deleted_exe, egress, ld_preload, new_listener, new_suid, reverse_shell,
)
from enodia_sentinel.system import Socket, SystemState
@dataclass
class FakeProc:
"""Duck-typed stand-in for system.Process."""
pid: int
comm: str = "bash"
cmdline: str = ""
exe: str = ""
environ: dict = field(default_factory=dict)
_stdio_inode: int | None = None
ppid: int = 1
uid: int = 0
cwd: str = "/"
def stdio_socket_inode(self):
return self._stdio_inode
def cfg() -> Config:
return Config()
class TestReverseShell(unittest.TestCase):
def test_interpreter_with_network_socket_alerts(self):
proc = FakeProc(pid=100, comm="bash", _stdio_inode=999,
cmdline="bash -i")
sock = Socket("ESTAB", "127.0.0.1:55", "9.9.9.9:443", 999, "bash", 100)
state = SystemState(processes=[proc], sockets=[sock])
alerts = list(reverse_shell.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
self.assertEqual(alerts[0].signature, "reverse_shell")
self.assertEqual(alerts[0].severity, Severity.CRITICAL)
self.assertEqual(alerts[0].pids, (100,))
def test_non_interpreter_does_not_alert(self):
proc = FakeProc(pid=101, comm="nginx", _stdio_inode=999)
sock = Socket("ESTAB", "127.0.0.1:55", "9.9.9.9:443", 999, "nginx", 101)
state = SystemState(processes=[proc], sockets=[sock])
self.assertEqual(list(reverse_shell.detect(state, cfg())), [])
def test_unix_socket_stdio_does_not_alert(self):
# stdio socket inode not present in the network socket table => benign
proc = FakeProc(pid=102, comm="bash", _stdio_inode=777)
state = SystemState(processes=[proc], sockets=[])
self.assertEqual(list(reverse_shell.detect(state, cfg())), [])
class TestLdPreload(unittest.TestCase):
def test_writable_path_alerts(self):
proc = FakeProc(pid=200, comm="sleep",
environ={"LD_PRELOAD": "/tmp/evil.so"})
state = SystemState(processes=[proc])
alerts = list(ld_preload.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
self.assertEqual(alerts[0].severity, Severity.CRITICAL)
def test_system_path_does_not_alert(self):
proc = FakeProc(pid=201, environ={"LD_PRELOAD": "/usr/lib/libfoo.so"})
state = SystemState(processes=[proc])
self.assertEqual(list(ld_preload.detect(state, cfg())), [])
class TestDeletedExe(unittest.TestCase):
def test_deleted_in_tmp_alerts(self):
proc = FakeProc(pid=300, comm="x", exe="/tmp/x (deleted)")
state = SystemState(processes=[proc])
self.assertEqual(len(list(deleted_exe.detect(state, cfg()))), 1)
def test_memfd_alerts(self):
proc = FakeProc(pid=301, comm="x", exe="/memfd:foo (deleted)")
state = SystemState(processes=[proc])
self.assertEqual(len(list(deleted_exe.detect(state, cfg()))), 1)
def test_deleted_system_path_ignored(self):
# daemon still running after a package upgrade — not fileless malware
proc = FakeProc(pid=302, comm="x", exe="/usr/bin/x (deleted)")
state = SystemState(processes=[proc])
self.assertEqual(list(deleted_exe.detect(state, cfg())), [])
class TestNewSuid(unittest.TestCase):
def test_new_in_writable_is_critical(self):
state = SystemState(
suid_binaries=["/tmp/evil", "/usr/bin/sudo"],
suid_baseline={"/usr/bin/sudo"},
)
alerts = list(new_suid.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
self.assertEqual(alerts[0].severity, Severity.CRITICAL)
self.assertIn("/tmp/evil", alerts[0].detail)
def test_new_in_system_path_is_high(self):
state = SystemState(
suid_binaries=["/usr/local/bin/newtool", "/usr/bin/sudo"],
suid_baseline={"/usr/bin/sudo"},
)
alerts = list(new_suid.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
self.assertEqual(alerts[0].severity, Severity.HIGH)
def test_no_scan_no_alert(self):
state = SystemState(suid_binaries=None, suid_baseline={"/usr/bin/sudo"})
self.assertEqual(list(new_suid.detect(state, cfg())), [])
class TestEgress(unittest.TestCase):
def test_interpreter_to_public_alerts(self):
sock = Socket("ESTAB", "10.0.0.2:5000", "8.8.8.8:443", 1, "python3", 400)
state = SystemState(sockets=[sock])
alerts = list(egress.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
def test_interpreter_to_private_ignored(self):
sock = Socket("ESTAB", "10.0.0.2:5000", "10.0.0.9:443", 1, "python3", 401)
state = SystemState(sockets=[sock])
self.assertEqual(list(egress.detect(state, cfg())), [])
def test_allowlisted_cidr_ignored(self):
c = Config()
c.egress_allow_cidrs = ("8.8.8.0/24",)
sock = Socket("ESTAB", "10.0.0.2:5000", "8.8.8.8:443", 1, "bash", 402)
state = SystemState(sockets=[sock])
self.assertEqual(list(egress.detect(state, c)), [])
class TestNewListener(unittest.TestCase):
def test_new_port_alerts(self):
sock = Socket("LISTEN", "0.0.0.0:31337", "0.0.0.0:*", 1, "nc", 500)
state = SystemState(sockets=[sock], listener_baseline=set())
alerts = list(new_listener.detect(state, cfg()))
self.assertEqual(len(alerts), 1)
self.assertEqual(alerts[0].severity, Severity.HIGH)
def test_baseline_port_ignored(self):
sock = Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:*", 1, "sshd", 501)
state = SystemState(sockets=[sock], listener_baseline={"22/sshd"})
self.assertEqual(list(new_listener.detect(state, cfg())), [])
if __name__ == "__main__":
unittest.main()