Add enrichment and SID coverage gates

This commit is contained in:
Luna 2026-07-08 17:05:17 -07:00
parent 93ac47dd82
commit 66472134d8
32 changed files with 1623 additions and 31 deletions

View file

@ -17,11 +17,15 @@ 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,))
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("qbittorrent", v.suggest)
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")
@ -38,7 +42,9 @@ class TestNewListener(unittest.TestCase):
a = alert("new_listener", "new listening socket 9.9.9.9:31337 by comm=weird",
pids=(77,))
procs = [{"pid": 77, "exe": "/tmp/weird"}]
self.assertEqual(triage_alert(a, procs, Config(), is_owned=NONE).label, REVIEW)
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):
@ -68,6 +74,44 @@ class TestOtherSignatures(unittest.TestCase):
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)
if __name__ == "__main__":
unittest.main()