125 lines
5.6 KiB
Python
125 lines
5.6 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Tests for false-positive triage (provenance injected, no package manager)."""
|
|
import unittest
|
|
|
|
from enodia_sentinel.config import Config
|
|
from enodia_sentinel.triage import LIKELY_FP, REVIEW, triage_alert
|
|
|
|
|
|
def alert(sig, detail, pids=(), sid=0):
|
|
return {"signature": sig, "detail": detail, "pids": list(pids), "sid": sid}
|
|
|
|
|
|
OWNED = lambda p: p in {"/usr/bin/qbittorrent", "/usr/bin/sudo"}
|
|
NONE = lambda p: False
|
|
|
|
|
|
class TestNewListener(unittest.TestCase):
|
|
def test_package_owned_is_fp(self):
|
|
a = alert("new_listener", "new listening socket 1.2.3.4:6881 by comm=qbittorrent",
|
|
pids=(50,), sid=100013)
|
|
a["classtype"] = "backdoor-listener"
|
|
procs = [{"pid": 50, "exe": "/usr/bin/qbittorrent"}]
|
|
v = triage_alert(a, procs, Config(), is_owned=OWNED)
|
|
self.assertEqual(v.label, LIKELY_FP)
|
|
self.assertIn("listener_allow_comms", v.suggest)
|
|
self.assertIn('"qbittorrent"', v.suggest)
|
|
self.assertIn("sid=100013", v.suggest)
|
|
self.assertIn("classtype=backdoor-listener", v.suggest)
|
|
|
|
def test_loopback_is_fp(self):
|
|
a = alert("new_listener", "new listening socket 127.0.0.1:9000 by comm=foo")
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, LIKELY_FP)
|
|
|
|
def test_allowlisted_comm_is_fp(self):
|
|
c = Config()
|
|
c.listener_allow_comms = frozenset({"nicotine"})
|
|
a = alert("new_listener", "new listening socket 1.2.3.4:2234 by comm=nicotine")
|
|
self.assertEqual(triage_alert(a, [], c, is_owned=NONE).label, LIKELY_FP)
|
|
|
|
def test_unknown_listener_is_review(self):
|
|
a = alert("new_listener", "new listening socket 9.9.9.9:31337 by comm=weird",
|
|
pids=(77,))
|
|
procs = [{"pid": 77, "exe": "/tmp/weird"}]
|
|
v = triage_alert(a, procs, Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("listener_allow_comms", v.suggest)
|
|
|
|
|
|
class TestOtherSignatures(unittest.TestCase):
|
|
def test_reverse_shell_always_review(self):
|
|
a = alert("reverse_shell", "pid=1 comm=bash stdio=net-socket ...")
|
|
self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, REVIEW)
|
|
|
|
def test_ld_preload_writable_is_review(self):
|
|
a = alert("ld_preload", "pid=1 comm=x LD_PRELOAD=[/tmp/evil.so]")
|
|
self.assertEqual(triage_alert(a, [], Config(), is_owned=NONE).label, REVIEW)
|
|
|
|
def test_suid_package_owned_is_fp(self):
|
|
a = alert("new_suid", "new SUID/SGID binary: /usr/bin/sudo")
|
|
self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, LIKELY_FP)
|
|
|
|
def test_suid_unowned_is_review(self):
|
|
a = alert("new_suid", "SUID/SGID binary in writable dir: /tmp/x")
|
|
self.assertEqual(triage_alert(a, [], Config(), is_owned=NONE).label, REVIEW)
|
|
|
|
def test_deleted_exe_package_owned_is_fp(self):
|
|
a = alert("deleted_exe", "pid=1 comm=foo exe=[/usr/bin/foo (deleted)]", pids=(1,))
|
|
procs = [{"pid": 1, "exe": "/usr/bin/foo (deleted)"}]
|
|
owned = lambda p: "foo" in p
|
|
self.assertEqual(triage_alert(a, procs, Config(), is_owned=owned).label, LIKELY_FP)
|
|
|
|
def test_exec_rule_is_review(self):
|
|
a = alert("exec_rule.fileless-execution", "sid=100001 ...")
|
|
self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, REVIEW)
|
|
|
|
def test_input_snooper_suggests_comm_allowlist_with_context(self):
|
|
a = alert("input_snooper",
|
|
"pid=12 comm=input-remapper fd=7 has input device open: /dev/input/event0",
|
|
sid=100032)
|
|
a["classtype"] = "credential-keylogging"
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("input_snooper_allow_comms", v.suggest)
|
|
self.assertIn('"input-remapper"', v.suggest)
|
|
self.assertIn("sid=100032", v.suggest)
|
|
|
|
def test_credential_access_suggests_comm_allowlist(self):
|
|
a = alert("credential_access",
|
|
"pid=22 comm=backupd fd=3 has private SSH key open: /home/a/.ssh/id_ed25519",
|
|
sid=100033)
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("credential_access_allow_comms", v.suggest)
|
|
self.assertIn('"backupd"', v.suggest)
|
|
|
|
def test_stealth_network_suggests_comm_allowlist(self):
|
|
a = alert("stealth_network",
|
|
"raw socket state=UNCONN local=0.0.0.0:* peer=*:* comm=tcpdump pid=5",
|
|
sid=100036)
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("stealth_network_allow_comms", v.suggest)
|
|
self.assertIn('"tcpdump"', v.suggest)
|
|
|
|
def test_memory_obfuscation_suggests_comm_allowlist(self):
|
|
a = alert("memory_obfuscation",
|
|
"pid=44 comm=jit-runtime writable+executable memory mapping perms=rwxp path=<anonymous>",
|
|
sid=100039)
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("memory_obfuscation_allow_comms", v.suggest)
|
|
self.assertIn('"jit-runtime"', v.suggest)
|
|
|
|
def test_process_injection_library_is_review(self):
|
|
a = alert("process_injection_library",
|
|
"pid=45 comm=sshd executable library mapped from writable/runtime path: /tmp/libinject.so",
|
|
sid=100049)
|
|
v = triage_alert(a, [], Config(), is_owned=NONE)
|
|
self.assertEqual(v.label, REVIEW)
|
|
self.assertIn("writable path", v.reason)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|