enodia-sentinal/tests/test_detectors.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""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()