diff --git a/.gitignore b/.gitignore index 6b704ac..b3f4958 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,18 @@ +# packaging artifacts *.pkg.tar.zst *.tar.gz pkg/ -src/*.o +dist/ +build/ +*.egg-info/ + +# python +__pycache__/ +*.py[cod] +.pytest_cache/ + +# editor / agent .claude/ + +# logs / local test output *.log diff --git a/Makefile b/Makefile index f0a7855..d099fb0 100644 --- a/Makefile +++ b/Makefile @@ -1,30 +1,36 @@ PREFIX ?= /usr/local BINDIR := $(PREFIX)/bin +LIBDIR := $(PREFIX)/lib/enodia-sentinel SYSTEMDDIR := /etc/systemd/system CONFDIR := /etc LOGDIR := /var/log/enodia-sentinel -.PHONY: install uninstall enable disable status logs check baseline drill clean +.PHONY: install uninstall enable disable status logs check baseline drill test clean +# Installs the stdlib-only Python package as a plain directory + launcher +# wrapper (no pip, no virtualenv, no site-packages). Zero runtime deps. install: - install -Dm755 src/sentinel.sh $(DESTDIR)$(BINDIR)/sentinel.sh + install -d $(DESTDIR)$(LIBDIR) + cp -r enodia_sentinel $(DESTDIR)$(LIBDIR)/ + install -Dm755 packaging/enodia-sentinel.wrapper $(DESTDIR)$(BINDIR)/enodia-sentinel install -Dm755 src/sentinel-redteam $(DESTDIR)$(BINDIR)/sentinel-redteam install -Dm644 systemd/enodia-sentinel.service $(DESTDIR)$(SYSTEMDDIR)/enodia-sentinel.service - @if [ ! -e "$(DESTDIR)$(CONFDIR)/enodia-sentinel.conf" ]; then \ - install -Dm644 config/enodia-sentinel.conf $(DESTDIR)$(CONFDIR)/enodia-sentinel.conf; \ - echo "Installed default config at $(CONFDIR)/enodia-sentinel.conf"; \ + @if [ ! -e "$(DESTDIR)$(CONFDIR)/enodia-sentinel.toml" ]; then \ + install -Dm644 config/enodia-sentinel.toml $(DESTDIR)$(CONFDIR)/enodia-sentinel.toml; \ + echo "Installed default config at $(CONFDIR)/enodia-sentinel.toml"; \ else \ - install -Dm644 config/enodia-sentinel.conf $(DESTDIR)$(CONFDIR)/enodia-sentinel.conf.new; \ - echo "Existing config preserved; new template at $(CONFDIR)/enodia-sentinel.conf.new"; \ + install -Dm644 config/enodia-sentinel.toml $(DESTDIR)$(CONFDIR)/enodia-sentinel.toml.new; \ + echo "Existing config preserved; new template at $(CONFDIR)/enodia-sentinel.toml.new"; \ fi install -dm750 $(DESTDIR)$(LOGDIR) @echo "Installed. Run 'sudo make enable' to start the service." uninstall: - rm -f $(DESTDIR)$(BINDIR)/sentinel.sh + rm -rf $(DESTDIR)$(LIBDIR) + rm -f $(DESTDIR)$(BINDIR)/enodia-sentinel rm -f $(DESTDIR)$(BINDIR)/sentinel-redteam rm -f $(DESTDIR)$(SYSTEMDDIR)/enodia-sentinel.service - rm -f $(DESTDIR)$(CONFDIR)/enodia-sentinel.conf.new + rm -f $(DESTDIR)$(CONFDIR)/enodia-sentinel.toml.new @echo "Uninstalled. Config and logs preserved." enable: @@ -40,14 +46,18 @@ status: logs: journalctl -u enodia-sentinel.service -f +# Dev targets — run from the repo without installing. +test: + python3 -m unittest discover -s tests -v + check: - sentinel.sh --check + python3 -m enodia_sentinel.cli check baseline: - sentinel.sh --baseline + python3 -m enodia_sentinel.cli baseline drill: - sentinel-redteam + ./src/sentinel-redteam clean: - @echo "Nothing to clean (no build artifacts)." + find . -type d -name __pycache__ -prune -exec rm -rf {} + diff --git a/README.md b/README.md index 71d7b1e..9199118 100644 --- a/README.md +++ b/README.md @@ -3,62 +3,72 @@ A host intrusion-detection daemon for Linux. It continuously runs a set of detectors over live system state — processes, sockets, file descriptors, the SUID inventory, and sensitive files — and writes a detailed forensic snapshot -with incident-response guidance the moment a known attack signature appears. +(text **and** JSON) with incident-response guidance the moment a known attack +signature appears. Think of it as the security counterpart to a performance watchdog: instead of "I/O pressure spiked, here's the kernel state," it's *"a shell just wired itself to a socket — here's the process tree, the peer, and what to do about it."* -> **Design lineage.** Sentinel reuses the proven control loop of a performance -> anomaly-capture daemon (sample → threshold/trigger → rich snapshot → classify -> → notify/retain) and swaps the inputs and signatures from *performance* to -> *security*. The forensic-snapshot-on-trigger pattern is the same; the -> detectors and the response guidance are new. +> **Two implementations, on purpose.** The project began as a bash prototype +> (`src/sentinel.sh`, kept as the regression **oracle**) and was re-architected +> into a zero-dependency Python package with a unit-test suite, structured +> detectors, and JSON output. The bash version and the Python version share one +> red-team harness, so every signature is exercised against both. ## Why these detectors Every detector keys on a behavior that is **cheap to observe** and **expensive -for an attacker to avoid** — the high-signal, low-false-positive heuristics that -real EDRs are built on: +for an attacker to avoid** — the high-signal, low-false-positive heuristics real +EDRs are built on: | Signature | What it catches | Why it's hard to evade | |---|---|---| -| `reverse_shell` | A shell/interpreter with a **socket on fd 0/1/2** | Interactive shells get a pty, not a socket — a socket on stdio is almost always `nc -e` / `bash -i >& /dev/tcp/...` | -| `ld_preload_global` | A non-empty `/etc/ld.so.preload` | Injecting into *every* process is the whole point of a userland rootkit; the file has to exist | -| `ld_preload_proc` | `LD_PRELOAD` pointing into `/tmp`, `/dev/shm`, … | Hooking libc to hide files/creds needs the library somewhere writable | -| `deleted_exe` | A process running from a **deleted / `memfd:`** binary | Fileless malware deletes its dropper but the kernel still names the inode `(deleted)` | -| `new_listener` | A listening port absent from the startup baseline | Bind shells and backdoors have to listen somewhere | -| `new_suid` | A new SUID/SGID binary (critical in writable dirs) | A SUID `/tmp` binary is a textbook privesc persistence trick | -| `persistence` | Changes to cron, systemd units, `authorized_keys`, rc files | Persistence has to write *somewhere* that survives reboot | -| `suspicious_egress` | An interpreter holding an outbound conn to a public IP | C2 beacons and exfil need to phone home | +| `reverse_shell` | An interpreter with a **network socket on fd 0/1/2** | Interactive shells get a pty and daemons get unix sockets — a *network* socket on stdio is `nc -e` / `bash -i >& /dev/tcp/...` | +| `ld_preload` | Non-empty `/etc/ld.so.preload`, or `LD_PRELOAD` into a writable dir | Injecting into processes needs the library to exist somewhere | +| `deleted_exe` | A process running from a **deleted / `memfd:`** binary | Fileless malware deletes its dropper; the kernel still names the inode `(deleted)` | +| `new_listener` | A listening port absent from the startup baseline | Bind shells/backdoors have to listen somewhere | +| `new_suid` | A new SUID/SGID binary (critical in a writable dir) | A SUID `/tmp` binary is a textbook privesc trick | +| `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 | -## The control loop +## Architecture ``` - ┌────────────────────────────────────────────┐ - │ every SAMPLE_INTERVAL seconds │ - │ │ - │ run_detectors() ──► alert lines │ - │ │ SEV signature detail │ - │ ▼ │ - │ cooldown dedup (per signature+key) │ - │ │ │ - │ ▼ (fresh alerts only) │ - │ capture_snapshot() │ - │ • flagged-pid deep dive (/proc) │ - │ • full process tree + sockets │ - │ • ld.so.preload, recent file mods │ - │ • logins, failed auth, kernel modules │ - │ • per-signature response guidance │ - │ │ │ - │ ▼ │ - │ events.log + alert-*.log + desktop notify │ - └────────────────────────────────────────────┘ +enodia_sentinel/ +├── cli.py run / check / baseline / list-detectors +├── daemon.py sweep loop · cooldown dedup · backgrounded SUID scan +├── system.py SystemState — one cached snapshot of /proc + ss per sweep +├── 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] ``` -Output lives in `/var/log/enodia-sentinel/`: -- `events.log` — one line per alert (timestamp, severity, signatures, file) -- `alert-YYYYMMDD-HHMMSS.log` — the full forensic snapshot +The loop is deliberately the same control flow as the bash prototype, but the +state lives in real objects: + +``` +every sample_interval seconds: + state = SystemState() # /proc + ss gathered once, cached + alerts = run_all(detectors, state, cfg) + fresh = drop alerts still within cooldown + if fresh: + snapshot.capture(fresh) # on a worker thread +``` + +Two design choices keep it fast and unobtrusive: + +- **One `SystemState` per sweep.** Detectors read shared, cached `/proc`/`ss` + data instead of each shelling out — a sweep costs ~200 ms regardless of how + many detectors run. +- **The filesystem-wide SUID scan runs off the loop thread** on a slow cadence, + so the multi-second walk never stalls live detection. + +Everything in `SystemState` is **injectable**, which is what makes the detectors +unit-testable without root or a live system (see `tests/`). ## Quick start @@ -72,103 +82,111 @@ sudo tail -f /var/log/enodia-sentinel/events.log sentinel-redteam # safe, self-cleaning attack simulations ``` -You'll watch the drills trip `reverse_shell`, `ld_preload_proc`, `deleted_exe`, -`new_listener`, and `new_suid` in real time, each producing a snapshot with -response guidance. +You'll watch the drills trip `reverse_shell`, `ld_preload`, `deleted_exe`, +`new_listener`, and `new_suid` in real time, each producing a `.log` + `.json` +snapshot with response guidance. -### Without installing (try it in place) +Output lives in `/var/log/enodia-sentinel/`: +- `events.log` — one line per alert +- `alert-YYYYMMDD-HHMMSS.log` — human-readable forensic snapshot +- `alert-YYYYMMDD-HHMMSS.json` — same data, structured (SIEM-ready) + +### Without installing ```bash -sudo ./src/sentinel.sh --baseline # establish listener/SUID baselines -sudo ./src/sentinel.sh --check # run every detector once, print findings +make test # run the unit suite +python3 -m enodia_sentinel.cli baseline # establish baselines +python3 -m enodia_sentinel.cli check # run every detector once, print findings ``` +No pip, no virtualenv, no dependencies — it's stdlib-only and installs as a +plain package directory plus a launcher wrapper. + ## The red-team harness -`sentinel-redteam` is the demo and the regression test in one. It simulates each -threat with **safe, clearly-labeled stand-ins** (everything tagged -`enodia-drill`, auto-cleaned on exit): - -- **reverse_shell** — a `bash` whose stdio is a TCP socket to a *local* listener - (no traffic leaves the box) -- **ld_preload** — a process with `LD_PRELOAD=/tmp/...` (an empty file; never - actually loaded, and it does **not** touch the system-wide `/etc/ld.so.preload`) -- **deleted_exe** — a copy of `/bin/sleep` run from `/tmp`, then deleted -- **new_listener** — an ephemeral `nc -l` on loopback -- **new_suid** — a `chmod 4755` copy of `/bin/true` in `/tmp` -- **persistence** — writes to a sandbox `authorized_keys` (only fires if you opt - the sandbox into `WATCH_PERSISTENCE`; it refuses to touch your real dotfiles) +`sentinel-redteam` is the demo and the integration test in one. It simulates +each threat with **safe, clearly-labeled stand-ins** (everything tagged +`enodia-drill`, auto-cleaned on exit), using a local Python TCP listener so no +traffic ever leaves the host: ```bash -sentinel-redteam --list # list drills -sentinel-redteam reverse_shell new_suid # run specific ones -HOLD=30 sentinel-redteam # keep artifacts alive 30s +sentinel-redteam --list # list drills +sentinel-redteam reverse_shell new_suid # run specific ones +HOLD=30 sentinel-redteam # keep artifacts alive 30s +``` + +It never touches your real dotfiles or `/etc/ld.so.preload`; the LD_PRELOAD +drill only sets the env var on a throwaway process. + +## Testing + +```bash +make test # 25 unit tests, stdlib unittest, no deps +``` + +Detectors are pure functions over an injectable `SystemState`, so tests build +fake processes/sockets and assert on the alerts — no root, no `/proc`, no `ss`: + +```python +proc = FakeProc(pid=100, comm="bash", _stdio_inode=999) +sock = Socket("ESTAB", "127.0.0.1:55", "9.9.9.9:443", 999, "bash", 100) +state = SystemState(processes=[proc], sockets=[sock]) +assert list(reverse_shell.detect(state, Config()))[0].signature == "reverse_shell" ``` ## Configuration -Edit `/etc/enodia-sentinel.conf`, then `sudo systemctl restart -enodia-sentinel.service`. Key options: +Edit `/etc/enodia-sentinel.toml`, then `sudo systemctl restart +enodia-sentinel.service`. Every key is optional. Highlights: -| Variable | Default | Purpose | +| Key | Default | Purpose | |---|---|---| -| `SAMPLE_INTERVAL` | 4 | seconds between detector sweeps | -| `COOLDOWN` | 60 | min seconds before re-alerting the same signature | -| `DET_*` | 1 | per-detector on/off switches | -| `INTERPRETERS` | bash sh … | process names treated as shells for rsh/egress | -| `EGRESS_ALLOW_CIDRS` | "" | trusted public ranges that won't trip egress | -| `LISTENER_ALLOW_PORTS` | "" | ports that never trip `new_listener` | -| `SUID_HOT_DIRS` | /tmp /dev/shm … | dirs where SUID = CRITICAL | -| `WATCH_PERSISTENCE` | cron/systemd/… | files watched for tampering | -| `CAPTURE_EXECVE_BPFTRACE` | 0 | add a bpftrace execve trace to snapshots | -| `MAX_SNAPSHOTS` / `MAX_SNAPSHOT_AGE_DAYS` | 300 / 60 | retention | -| `NOTIFY_USERS` | "" | desktop notify-send targets | - -## Tuning out false positives - -- A long-running daemon legitimately running a deleted exe after a package - upgrade won't alert — `deleted_exe` only fires for `/tmp`, `/dev/shm`, - `/run`, `/var/tmp`, or `memfd:` paths. -- Your own admin shells over SSH won't trip `reverse_shell` — they're on a pty. -- Browsers and chat apps that interpreters open to public IPs *can* trip - `suspicious_egress` if you add them to `INTERPRETERS`; the default list is - deliberately scripting-focused. Use `EGRESS_ALLOW_CIDRS` for known-good ranges. +| `sample_interval` | 4 | seconds between sweeps | +| `cooldown` | 60 | min seconds before re-alerting a signature | +| `detectors` | all 7 | the enabled detector list | +| `interpreters` | bash sh … | process names treated as shells | +| `egress_allow_cidrs` | [] | trusted public ranges (won't trip egress) | +| `suid_hot_dirs` | /tmp … | dirs where a SUID binary is CRITICAL | +| `suid_scan_extra_dirs` | /tmp … | writable mounts always scanned (tmpfs-safe) | +| `capture_execve_bpftrace` | false | add a bpftrace execve trace to snapshots | +| `notify_users` | [] | desktop notify-send targets | ## Security model Sentinel runs as root because it must read every process's `/proc`, the full socket table, and root-owned files like `authorized_keys`. The systemd unit -constrains that power hard: `ProtectSystem=strict` with the log dir as the only +constrains that power: `ProtectSystem=strict` with the log dir as the only writable path, `ProtectHome=read-only`, `NoNewPrivileges`, -`MemoryDenyWriteExecute`, and a minimal capability set +`MemoryDenyWriteExecute`, `RestrictNamespaces`, and a minimal capability set (`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 -This v0 is **poll-based**: it sweeps `/proc` and `ss` every few seconds. That's -robust, dependency-light, and catches anything that lingers — but it can miss -sub-second processes, and a sweep is observable. The planned evolution: +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: -1. **bpftrace tracepoints (now, optional)** — `CAPTURE_EXECVE_BPFTRACE=1` adds a - live `execve` trace to each snapshot. The gentle on-ramp to kernel tracing. -2. **Event-driven execve/connect detection** — replace the polling gap with - `bpftrace` probes on `sys_enter_execve`, `security_bprm_check`, and - `tcp_connect`, streamed to the daemon so short-lived reverse shells can't - slip between sweeps. -3. **A libbpf + CO-RE agent (Go or Rust userland)** — the real EDR core: LSM - hooks, ring-buffer event streaming, per-process lineage tracking, and - tamper-resistance. This is the production-grade rewrite the polling daemon - prototypes the detection logic for. +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. 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 for both. +shared regression suite. ## Project status -v0.1 — working poll-based detection across 7 signatures, forensic snapshots, -red-team harness, systemd packaging. Built and tested on Arch Linux. +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. +Built and tested on Arch Linux. ## License diff --git a/config/enodia-sentinel.toml b/config/enodia-sentinel.toml new file mode 100644 index 0000000..9793255 --- /dev/null +++ b/config/enodia-sentinel.toml @@ -0,0 +1,60 @@ +# Enodia Sentinel configuration (TOML). +# Read at startup from /etc/enodia-sentinel.toml (or $ENODIA_CONFIG). +# Restart after editing: sudo systemctl restart enodia-sentinel.service +# +# Every key is optional; omitted keys keep their built-in default. + +# --- timing -------------------------------------------------------------- +sample_interval = 4.0 # seconds between detector sweeps +cooldown = 60 # min seconds before re-alerting the same signature +baseline_grace = 10 # seconds after start before new-listener/suid arm +suid_refresh = 3600 # seconds between SUID baseline refreshes +suid_scan_interval = 60 # seconds between (backgrounded) SUID filesystem scans + +# --- detectors (omit one to disable it) ---------------------------------- +detectors = [ + "reverse_shell", "ld_preload", "deleted_exe", + "new_listener", "new_suid", "persistence", "egress", +] + +# --- tuning -------------------------------------------------------------- +# Process names treated as "interpreters" for reverse-shell / egress checks. +interpreters = [ + "bash", "sh", "dash", "zsh", "ksh", "ash", "nc", "ncat", "netcat", + "socat", "telnet", "python", "python2", "python3", "perl", "ruby", + "php", "lua", "awk", +] + +# Trusted public CIDRs that should NOT trip the egress detector +# (your VPS, VPN exit, monitoring endpoints, …). +egress_allow_cidrs = [] + +# Listener ports that never alert even if they appear after baseline. +listener_allow_ports = [] + +# Dirs where any SUID binary is treated as CRITICAL (attacker-writable). +suid_hot_dirs = ["/tmp", "/dev/shm", "/var/tmp", "/home", "/run/user"] + +# Writable dirs always scanned for SUID even if on a separate filesystem +# (a tmpfs /tmp would otherwise be skipped by the on-device walk). +suid_scan_extra_dirs = ["/tmp", "/dev/shm", "/var/tmp", "/run/user"] + +# Persistence files/dirs watched for modification. +watch_persistence = [ + "/etc/cron.d", "/etc/crontab", "/etc/cron.daily", "/etc/cron.hourly", + "/etc/cron.weekly", "/var/spool/cron", "/etc/systemd/system", + "/etc/ld.so.preload", "/etc/passwd", "/etc/sudoers", "/etc/sudoers.d", + "/root/.ssh/authorized_keys", "/root/.bashrc", "/root/.profile", +] + +# --- eBPF on-ramp -------------------------------------------------------- +# Add a 3s bpftrace execve trace to each snapshot (requires the bpftrace pkg). +capture_execve_bpftrace = false + +# --- retention ----------------------------------------------------------- +max_snapshots = 300 # auto-delete oldest beyond this count (0 = no cap) +max_snapshot_age_days = 60 # auto-delete older than this (0 = no age cap) + +# --- notifications ------------------------------------------------------- +notify_users = [] # e.g. ["luna"] — desktop notify-send on alert +notify_urgency = "critical" # low / normal / critical diff --git a/enodia_sentinel/__init__.py b/enodia_sentinel/__init__.py new file mode 100644 index 0000000..ec34618 --- /dev/null +++ b/enodia_sentinel/__init__.py @@ -0,0 +1,8 @@ +"""Enodia Sentinel — a host intrusion-detection daemon. + +A poll-based HIDS that runs a set of detectors over live system state and +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" diff --git a/enodia_sentinel/alert.py b/enodia_sentinel/alert.py new file mode 100644 index 0000000..f3833f6 --- /dev/null +++ b/enodia_sentinel/alert.py @@ -0,0 +1,39 @@ +"""Alert and severity types shared by all detectors.""" +from __future__ import annotations + +import enum +from dataclasses import dataclass, field + + +class Severity(enum.IntEnum): + MEDIUM = 1 + HIGH = 2 + CRITICAL = 3 + + def __str__(self) -> str: # pragma: no cover - trivial + return self.name + + +@dataclass(frozen=True) +class Alert: + """A single detection. + + `key` is a stable dedup identity (e.g. ``rsh:1234``) used by the daemon's + cooldown so the same condition doesn't re-fire every sweep. `pids` lists + processes the snapshot should deep-dive. + """ + + severity: Severity + signature: str + key: str + detail: str + pids: tuple[int, ...] = field(default_factory=tuple) + + def to_dict(self) -> dict: + return { + "severity": str(self.severity), + "signature": self.signature, + "key": self.key, + "detail": self.detail, + "pids": list(self.pids), + } diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py new file mode 100644 index 0000000..07e12ac --- /dev/null +++ b/enodia_sentinel/cli.py @@ -0,0 +1,80 @@ +"""Command-line entry point. + + enodia-sentinel run # daemon loop (default; used by systemd) + enodia-sentinel check # run every detector once, print alerts, exit + enodia-sentinel baseline # (re)build listener/SUID baselines and exit +""" +from __future__ import annotations + +import argparse +import signal +import sys + +from . import __version__, detectors +from .config import Config +from .daemon import Sentinel +from .system import SystemState, scan_suid_binaries + + +def _cmd_run(cfg: Config) -> int: + sentinel = Sentinel(cfg) + signal.signal(signal.SIGTERM, sentinel.stop) + signal.signal(signal.SIGINT, sentinel.stop) + sentinel.run() + return 0 + + +def _cmd_baseline(cfg: Config) -> int: + sentinel = Sentinel(cfg) + sentinel.build_baselines() + print(f"Baselines written under {cfg.log_dir}") + return 0 + + +def _cmd_check(cfg: Config) -> int: + # One-shot: arm everything immediately, force the SUID scan. + sentinel = Sentinel(cfg) + sentinel.start_time = 0.0 # past the grace window + sentinel.load_baselines() + if not sentinel.listener_baseline: + sentinel.build_baselines() + alerts = sentinel.sweep(force_suid=True) + if not alerts: + print("No alerts.") + return 0 + for a in sorted(alerts, key=lambda x: -x.severity): + print(f"[{a.severity}] {a.signature:<14} {a.detail}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="enodia-sentinel", + description="Host intrusion-detection daemon.", + ) + parser.add_argument("--version", action="version", + version=f"enodia-sentinel {__version__}") + parser.add_argument("-c", "--config", help="path to TOML config") + sub = parser.add_subparsers(dest="cmd") + sub.add_parser("run", help="run the daemon loop (default)") + sub.add_parser("check", help="run detectors once and print alerts") + sub.add_parser("baseline", help="rebuild listener/SUID baselines") + sub.add_parser("list-detectors", help="list available detectors") + + args = parser.parse_args(argv) + cfg = Config.load(args.config) + + if args.cmd == "list-detectors": + for det in detectors.REGISTRY: + mark = "on " if cfg.enabled(det.name) else "off" + print(f" [{mark}] {det.name}") + return 0 + if args.cmd == "baseline": + return _cmd_baseline(cfg) + if args.cmd == "check": + return _cmd_check(cfg) + return _cmd_run(cfg) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/enodia_sentinel/config.py b/enodia_sentinel/config.py new file mode 100644 index 0000000..fd3885d --- /dev/null +++ b/enodia_sentinel/config.py @@ -0,0 +1,117 @@ +"""Runtime configuration. + +Defaults live here; overrides come from a TOML file (default +``/etc/enodia-sentinel.toml``, or ``$ENODIA_CONFIG``) and a couple of env vars +useful for testing without root (``$ENODIA_LOG_DIR``). +""" +from __future__ import annotations + +import os +import tomllib +from dataclasses import dataclass, field, fields +from pathlib import Path + +_DEFAULT_INTERPRETERS = ( + "bash sh dash zsh ksh ash nc ncat netcat socat telnet " + "python python2 python3 perl ruby php lua awk" +).split() + +_DEFAULT_WATCH = ( + "/etc/cron.d", "/etc/crontab", "/etc/cron.daily", "/etc/cron.hourly", + "/etc/cron.weekly", "/var/spool/cron", "/etc/systemd/system", + "/etc/ld.so.preload", "/etc/passwd", "/etc/sudoers", "/etc/sudoers.d", + "/root/.ssh/authorized_keys", "/root/.bashrc", "/root/.profile", +) + +_ALL_DETECTORS = ( + "reverse_shell", "ld_preload", "deleted_exe", + "new_listener", "new_suid", "persistence", "egress", +) + + +@dataclass +class Config: + # timing + sample_interval: float = 4.0 + cooldown: int = 60 + baseline_grace: int = 10 + suid_refresh: int = 3600 + suid_scan_interval: int = 60 + + # detector toggles + detectors: frozenset[str] = frozenset(_ALL_DETECTORS) + + # tuning + interpreters: frozenset[str] = frozenset(_DEFAULT_INTERPRETERS) + egress_allow_cidrs: tuple[str, ...] = () + listener_allow_ports: frozenset[str] = frozenset() + suid_hot_dirs: tuple[str, ...] = ( + "/tmp", "/dev/shm", "/var/tmp", "/home", "/run/user", + ) + # Writable dirs always scanned for SUID even if on a separate filesystem + # (tmpfs /tmp etc.). Kept small so the gated scan stays cheap. + suid_scan_extra_dirs: tuple[str, ...] = ( + "/tmp", "/dev/shm", "/var/tmp", "/run/user", + ) + watch_persistence: tuple[str, ...] = _DEFAULT_WATCH + + # eBPF on-ramp + capture_execve_bpftrace: bool = False + + # retention + max_snapshots: int = 300 + max_snapshot_age_days: int = 60 + + # notifications + notify_users: tuple[str, ...] = () + notify_urgency: str = "critical" + + # paths + log_dir: Path = Path("/var/log/enodia-sentinel") + + # ---- derived paths --------------------------------------------------- + @property + def events_log(self) -> Path: + return self.log_dir / "events.log" + + @property + def listener_baseline(self) -> Path: + return self.log_dir / "listener-baseline.json" + + @property + def suid_baseline(self) -> Path: + return self.log_dir / "suid-baseline.json" + + # ---- loading --------------------------------------------------------- + @classmethod + def load(cls, path: str | os.PathLike | None = None) -> "Config": + cfg = cls() + + env_log = os.environ.get("ENODIA_LOG_DIR") + if env_log: + cfg.log_dir = Path(env_log) + + cfg_path = Path(path or os.environ.get("ENODIA_CONFIG") + or "/etc/enodia-sentinel.toml") + if cfg_path.is_file(): + cfg._apply_toml(cfg_path) + return cfg + + def _apply_toml(self, path: Path) -> None: + with open(path, "rb") as fh: + data = tomllib.load(fh) + known = {f.name for f in fields(self)} + for key, value in data.items(): + if key not in known: + continue + current = getattr(self, key) + if isinstance(current, frozenset): + value = frozenset(value) + elif isinstance(current, tuple): + value = tuple(value) + elif isinstance(current, Path): + value = Path(value) + setattr(self, key, value) + + def enabled(self, detector: str) -> bool: + return detector in self.detectors diff --git a/enodia_sentinel/daemon.py b/enodia_sentinel/daemon.py new file mode 100644 index 0000000..c9215d4 --- /dev/null +++ b/enodia_sentinel/daemon.py @@ -0,0 +1,144 @@ +"""The detection daemon: sweep loop, cooldown dedup, baseline management. + +Unlike the bash prototype, all loop state (cooldowns, last-scan timestamps, +baselines) lives in this object — no subshell-state surprises — and the +expensive filesystem-wide SUID scan is gated to its own slow cadence. +""" +from __future__ import annotations + +import json +import threading +import time +from pathlib import Path + +from . import detectors, snapshot +from .alert import Alert +from .config import Config +from .system import SystemState, scan_suid_binaries + + +class Sentinel: + def __init__(self, cfg: Config) -> None: + self.cfg = cfg + self.start_time = time.time() + self.cooldowns: dict[str, float] = {} + self.last_persist_scan = self.start_time + self.listener_baseline: set[str] = set() + self.suid_baseline: set[str] = set() + # SUID scan runs off the loop thread; the loop reads the latest result. + self._suid_current: list[str] | None = None + self._suid_thread: threading.Thread | None = None + self._last_suid_scan = 0.0 + self._last_suid_baseline_refresh = self.start_time + self._stop = threading.Event() + + # -- baselines --------------------------------------------------------- + def build_baselines(self) -> None: + self.listener_baseline = SystemState().listener_keys() + self.suid_baseline = set(scan_suid_binaries( + extra_dirs=self.cfg.suid_scan_extra_dirs)) + self.cfg.log_dir.mkdir(parents=True, exist_ok=True) + self._save(self.cfg.listener_baseline, sorted(self.listener_baseline)) + self._save(self.cfg.suid_baseline, sorted(self.suid_baseline)) + + def load_baselines(self) -> None: + self.listener_baseline = set(self._load(self.cfg.listener_baseline)) + self.suid_baseline = set(self._load(self.cfg.suid_baseline)) + + @staticmethod + def _save(path: Path, data: list[str]) -> None: + path.write_text(json.dumps(data)) + + @staticmethod + def _load(path: Path) -> list[str]: + try: + return json.loads(path.read_text()) + except (OSError, ValueError): + return [] + + # -- SUID scan (off the loop thread) ----------------------------------- + def _maybe_scan_suid(self, now: float) -> None: + if self._suid_thread and self._suid_thread.is_alive(): + return + if (now - self._last_suid_scan) < self.cfg.suid_scan_interval: + return + self._last_suid_scan = now + self._suid_thread = threading.Thread(target=self._scan_suid, daemon=True) + self._suid_thread.start() + + def _scan_suid(self) -> None: + result = scan_suid_binaries(extra_dirs=self.cfg.suid_scan_extra_dirs) + self._suid_current = result + # Periodically fold the current state into the baseline so legitimately + # installed SUID binaries stop alerting after a while. + if (time.time() - self._last_suid_baseline_refresh) >= self.cfg.suid_refresh: + self.suid_baseline = set(result) + self._last_suid_baseline_refresh = time.time() + self._save(self.cfg.suid_baseline, sorted(self.suid_baseline)) + + # -- one sweep --------------------------------------------------------- + def sweep(self, *, force_suid: bool = False) -> list[Alert]: + now = time.time() + armed = (now - self.start_time) >= self.cfg.baseline_grace + + if force_suid: + suid_binaries = scan_suid_binaries( + extra_dirs=self.cfg.suid_scan_extra_dirs) + else: + suid_binaries = self._suid_current # latest async result (may be None) + + state = SystemState( + listener_baseline=self.listener_baseline if armed or force_suid else None, + suid_baseline=self.suid_baseline, + suid_binaries=suid_binaries if armed or force_suid else None, + persist_since=self.last_persist_scan if armed or force_suid else None, + ) + alerts = list(detectors.run_all(state, self.cfg)) + self.last_persist_scan = now + return alerts + + def fresh_alerts(self, alerts: list[Alert], now: float) -> list[Alert]: + """Drop alerts whose dedup key is still within cooldown.""" + 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) + return out + + # -- main loop --------------------------------------------------------- + def run(self) -> None: + self.cfg.log_dir.mkdir(parents=True, exist_ok=True) + with open(self.cfg.events_log, "a") as fh: + fh.write(f"{time.strftime('%FT%T%z')} enodia-sentinel started\n") + self.build_baselines() + snapshot.prune(self.cfg) + sweeps = 0 + while not self._stop.is_set(): + now = time.time() + if (now - self.start_time) >= self.cfg.baseline_grace: + self._maybe_scan_suid(now) + alerts = self.sweep() + fresh = self.fresh_alerts(alerts, now) + if fresh: + # capture off the loop thread so a slow snapshot never stalls + # detection; the SystemState used for forensics is rebuilt fresh + # inside the thread for accuracy. + threading.Thread( + target=self._capture, args=(fresh,), daemon=True + ).start() + sweeps += 1 + if sweeps % 20 == 0: + snapshot.prune(self.cfg) + self._stop.wait(self.cfg.sample_interval) + + def _capture(self, alerts: list[Alert]) -> None: + try: + snapshot.capture(alerts, SystemState(), self.cfg) + except Exception as exc: # never let a capture crash the daemon + with open(self.cfg.events_log, "a") as fh: + fh.write(f"{time.strftime('%FT%T%z')} capture error: {exc!r}\n") + + def stop(self, *_a) -> None: + self._stop.set() diff --git a/enodia_sentinel/detectors/__init__.py b/enodia_sentinel/detectors/__init__.py new file mode 100644 index 0000000..9b80499 --- /dev/null +++ b/enodia_sentinel/detectors/__init__.py @@ -0,0 +1,62 @@ +"""Detector registry. + +Each detector is a ``Detector`` with a ``name`` and a ``detect(state, cfg)`` +callable yielding ``Alert``s. They are pure functions of the injected +``SystemState`` + ``Config``, which is what makes them unit-testable without +root or a live system. +""" +from __future__ import annotations + +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass + +from ..alert import Alert +from ..config import Config +from ..system import SystemState +from . import ( + deleted_exe, + egress, + ld_preload, + new_listener, + new_suid, + persistence, + reverse_shell, +) + +DetectFn = Callable[[SystemState, Config], Iterable[Alert]] + + +@dataclass(frozen=True) +class Detector: + name: str + detect: DetectFn + needs_baseline: bool = False # gated until after the grace window + expensive: bool = False # gated to the slow scan cadence + + +# Order here is the order signatures appear in a snapshot. +REGISTRY: tuple[Detector, ...] = ( + Detector("reverse_shell", reverse_shell.detect), + Detector("ld_preload", ld_preload.detect), + Detector("deleted_exe", deleted_exe.detect), + Detector("egress", egress.detect), + Detector("new_listener", new_listener.detect, needs_baseline=True), + Detector("persistence", persistence.detect, needs_baseline=True), + Detector("new_suid", new_suid.detect, needs_baseline=True, expensive=True), +) + + +def run_all( + state: SystemState, + cfg: Config, + *, + include: Iterable[str] | None = None, +) -> Iterator[Alert]: + """Run the enabled detectors whose names are in ``include`` (or all).""" + wanted = set(include) if include is not None else None + for det in REGISTRY: + if not cfg.enabled(det.name): + continue + if wanted is not None and det.name not in wanted: + continue + yield from det.detect(state, cfg) diff --git a/enodia_sentinel/detectors/deleted_exe.py b/enodia_sentinel/detectors/deleted_exe.py new file mode 100644 index 0000000..8bafe94 --- /dev/null +++ b/enodia_sentinel/detectors/deleted_exe.py @@ -0,0 +1,39 @@ +"""deleted_exe — a process running from a deleted or memfd-backed binary. + +Fileless malware unlinks its dropper (or runs straight from ``memfd``) so there +is nothing on disk to scan. The kernel still labels the ``/proc//exe`` +symlink ``(deleted)`` or ``memfd:``. Normal-path deleted exes (a daemon still +running after a package upgrade) are ignored — only attacker-controlled/fileless +locations alert. +""" +from __future__ import annotations + +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + +_SUSPICIOUS_PREFIXES = ("/tmp/", "/dev/shm/", "/var/tmp/", "/run/") + + +def _is_fileless(exe: str) -> bool: + if "memfd:" in exe: + return True + if "(deleted)" not in exe: + return False + return exe.startswith(_SUSPICIOUS_PREFIXES) + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + for proc in state.processes: + exe = proc.exe + if not exe or not _is_fileless(exe): + continue + yield Alert( + severity=Severity.CRITICAL, + signature="deleted_exe", + key=f"del:{proc.pid}", + detail=f"pid={proc.pid} comm={proc.comm} exe=[{exe}]", + pids=(proc.pid,), + ) diff --git a/enodia_sentinel/detectors/egress.py b/enodia_sentinel/detectors/egress.py new file mode 100644 index 0000000..22a7633 --- /dev/null +++ b/enodia_sentinel/detectors/egress.py @@ -0,0 +1,35 @@ +"""egress — an interpreter holding an outbound connection to a public IP. + +C2 beacons and exfil phone home. A scripting interpreter (bash/python/perl/nc…) +with an established connection to a globally-routable address — that isn't in +the operator's trust list — is worth a look. +""" +from __future__ import annotations + +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..netutil import ip_in_cidrs, is_public_ip, split_host_port +from ..system import SystemState + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + for s in state.established_sockets(): + if s.comm not in cfg.interpreters: + continue + host, port = split_host_port(s.peer) + if not is_public_ip(host): + continue + if ip_in_cidrs(host, cfg.egress_allow_cidrs): + continue + yield Alert( + severity=Severity.HIGH, + signature="egress", + key=f"egr:{s.pid}:{host}", + detail=( + f"pid={s.pid} comm={s.comm} -> {host}:{port} " + "(interpreter to public IP)" + ), + pids=(s.pid,) if s.pid else (), + ) diff --git a/enodia_sentinel/detectors/ld_preload.py b/enodia_sentinel/detectors/ld_preload.py new file mode 100644 index 0000000..31597e3 --- /dev/null +++ b/enodia_sentinel/detectors/ld_preload.py @@ -0,0 +1,46 @@ +"""ld_preload — userland rootkit / library-injection indicators. + +Two signatures: + * ``/etc/ld.so.preload`` non-empty: injects into *every* dynamically-linked + process — the classic system-wide rootkit hook. + * a process whose ``LD_PRELOAD`` points into a writable/temp dir — per-process + function hooking (hiding files/procs, stealing credentials). +""" +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + +_SUSPICIOUS_PREFIXES = ("/tmp/", "/dev/shm/", "/var/tmp/", "/run/user/", "./") + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + preload = Path("/etc/ld.so.preload") + try: + contents = preload.read_text().strip() if preload.is_file() else "" + except OSError: + contents = "" + if contents: + yield Alert( + severity=Severity.CRITICAL, + signature="ld_preload", + key="ldp:global", + detail=f"/etc/ld.so.preload is non-empty: [{contents.replace(chr(10), ' ')}]", + ) + + for proc in state.processes: + pre = proc.environ.get("LD_PRELOAD", "") + if not pre: + continue + if pre.startswith(_SUSPICIOUS_PREFIXES): + yield Alert( + severity=Severity.CRITICAL, + signature="ld_preload", + key=f"ldp:{proc.pid}", + detail=f"pid={proc.pid} comm={proc.comm} LD_PRELOAD=[{pre}]", + pids=(proc.pid,), + ) diff --git a/enodia_sentinel/detectors/new_listener.py b/enodia_sentinel/detectors/new_listener.py new file mode 100644 index 0000000..f8abd30 --- /dev/null +++ b/enodia_sentinel/detectors/new_listener.py @@ -0,0 +1,33 @@ +"""new_listener — a listening socket absent from the startup baseline. + +Bind shells and backdoors have to listen somewhere. We diff the current set of +``port/comm`` listeners against the baseline captured at start; anything new +(and not explicitly allow-listed) alerts. +""" +from __future__ import annotations + +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + if state.listener_baseline is None: + return + for s in state.listening_sockets(): + port = s.local.rpartition(":")[2] + comm = s.comm or "?" + key = f"{port}/{comm}" + if key in state.listener_baseline: + continue + if port in cfg.listener_allow_ports: + continue + yield Alert( + severity=Severity.HIGH, + signature="new_listener", + key=f"lis:{key}", + detail=f"new listening socket {s.local} by comm={comm}", + pids=(s.pid,) if s.pid else (), + ) diff --git a/enodia_sentinel/detectors/new_suid.py b/enodia_sentinel/detectors/new_suid.py new file mode 100644 index 0000000..277ec72 --- /dev/null +++ b/enodia_sentinel/detectors/new_suid.py @@ -0,0 +1,35 @@ +"""new_suid — a SUID/SGID binary that wasn't in the baseline. + +A new setuid binary is a privilege-escalation persistence trick; one in a +world-writable dir (``/tmp``, ``/home`` …) is almost never legitimate, so it is +escalated to CRITICAL. The filesystem walk is expensive, so the daemon only +populates ``state.suid_binaries`` on its slow scan cadence — when it's absent +this detector is a no-op. +""" +from __future__ import annotations + +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + if state.suid_binaries is None or state.suid_baseline is None: + return + for path in state.suid_binaries: + if path in state.suid_baseline: + continue + hot = any( + path.startswith(d.rstrip("/") + "/") for d in cfg.suid_hot_dirs + ) + yield Alert( + severity=Severity.CRITICAL if hot else Severity.HIGH, + signature="new_suid", + key=f"suid:{path}", + detail=( + ("SUID/SGID binary in writable dir: " if hot + else "new SUID/SGID binary: ") + path + ), + ) diff --git a/enodia_sentinel/detectors/persistence.py b/enodia_sentinel/detectors/persistence.py new file mode 100644 index 0000000..2b29114 --- /dev/null +++ b/enodia_sentinel/detectors/persistence.py @@ -0,0 +1,42 @@ +"""persistence — modification of a watched persistence-relevant file. + +Persistence has to land somewhere that survives a reboot: cron, systemd units, +``authorized_keys``, shell rc files. We watch a configurable set and alert on +anything modified since the previous scan (``state.persist_since``). +""" +from __future__ import annotations + +import os +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + + +def _iter_files(roots) -> Iterator[str]: + for root in roots: + if os.path.isdir(root): + for dirpath, _dirs, files in os.walk(root, onerror=lambda e: None): + for f in files: + yield os.path.join(dirpath, f) + elif os.path.lexists(root): + yield root + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + since = state.persist_since + if since is None: + return + for path in _iter_files(cfg.watch_persistence): + try: + mtime = os.lstat(path).st_mtime + except OSError: + continue + if mtime > since: + yield Alert( + severity=Severity.HIGH, + signature="persistence", + key=f"persist:{path}", + detail=f"persistence file modified: {path}", + ) diff --git a/enodia_sentinel/detectors/reverse_shell.py b/enodia_sentinel/detectors/reverse_shell.py new file mode 100644 index 0000000..0b41f0e --- /dev/null +++ b/enodia_sentinel/detectors/reverse_shell.py @@ -0,0 +1,37 @@ +"""reverse_shell — an interpreter with a network socket on its stdio. + +A real reverse shell dups a TCP/UDP socket onto fd 0/1/2 (``nc -e /bin/bash``, +``bash -i >& /dev/tcp/...``). Interactive shells get a pty, and daemons get +unix sockets/pipes — neither is a network socket, so requiring the stdio socket +to appear in the network socket table is what keeps false positives near zero. +""" +from __future__ import annotations + +from collections.abc import Iterator + +from ..alert import Alert, Severity +from ..config import Config +from ..system import SystemState + + +def detect(state: SystemState, cfg: Config) -> Iterator[Alert]: + peer_by_inode = state.net_peer_by_inode + for proc in state.processes: + if proc.comm not in cfg.interpreters: + continue + inode = proc.stdio_socket_inode() + if inode is None: + continue + peer = peer_by_inode.get(inode) + if not peer: # unix socket / pipe — benign + continue + yield Alert( + severity=Severity.CRITICAL, + signature="reverse_shell", + key=f"rsh:{proc.pid}", + detail=( + f"pid={proc.pid} comm={proc.comm} stdio=net-socket " + f"peer=[{peer}] cmd=[{proc.cmdline[:90]}]" + ), + pids=(proc.pid,), + ) diff --git a/enodia_sentinel/netutil.py b/enodia_sentinel/netutil.py new file mode 100644 index 0000000..6849d7c --- /dev/null +++ b/enodia_sentinel/netutil.py @@ -0,0 +1,62 @@ +"""Network address helpers, backed by the stdlib ``ipaddress`` module. + +The bash prototype hand-rolled integer/CIDR math; here we lean on ``ipaddress`` +for correct handling of private ranges, link-local, CGNAT, and IPv6. +""" +from __future__ import annotations + +import ipaddress +from collections.abc import Iterable + + +def parse_addr(addr: str) -> ipaddress._BaseAddress | None: + """Parse an address that may carry a ``%iface`` zone or ``[..]`` brackets.""" + if not addr: + return None + addr = addr.strip().strip("[]") + addr = addr.split("%", 1)[0] # drop IPv6 zone id + try: + return ipaddress.ip_address(addr) + except ValueError: + return None + + +def is_public_ip(addr: str) -> bool: + """True if ``addr`` is a globally-routable address. + + Excludes loopback, RFC1918 private, link-local, CGNAT (100.64/10), and + other non-global ranges — i.e. a connection to such an address is one that + actually leaves the host to the public internet. + """ + ip = parse_addr(addr) + if ip is None: + return False + return ip.is_global + + +def ip_in_cidrs(addr: str, cidrs: Iterable[str]) -> bool: + """True if ``addr`` falls inside any CIDR in ``cidrs`` (the trust list).""" + ip = parse_addr(addr) + if ip is None: + return False + for cidr in cidrs: + cidr = cidr.strip() + if not cidr: + continue + try: + net = ipaddress.ip_network(cidr, strict=False) + except ValueError: + continue + if ip.version == net.version and ip in net: + return True + return False + + +def split_host_port(endpoint: str) -> tuple[str, str]: + """Split an ``ss`` endpoint like ``10.0.0.2%eth0:443`` or ``[::1]:22``.""" + endpoint = endpoint.strip() + if endpoint.startswith("["): + host, _, port = endpoint.partition("]") + return host.lstrip("["), port.lstrip(":") + host, _, port = endpoint.rpartition(":") + return host, port diff --git a/enodia_sentinel/snapshot.py b/enodia_sentinel/snapshot.py new file mode 100644 index 0000000..a58ac91 --- /dev/null +++ b/enodia_sentinel/snapshot.py @@ -0,0 +1,247 @@ +"""Forensic snapshot writer. + +When detections fire, this captures a timestamped report — both a human-readable +``.log`` and a structured ``.json`` sidecar — with per-signature incident- +response guidance, the flagged processes' /proc detail, the process tree, +sockets, recent file changes, and auth events. +""" +from __future__ import annotations + +import json +import os +import subprocess +import time +from datetime import datetime +from pathlib import Path + +from .alert import Alert, Severity +from .config import Config +from .system import Process, SystemState + +RESPONSES: dict[str, str] = { + "reverse_shell": ( + "A shell/interpreter has a network socket wired to its stdin/stdout — " + "the canonical reverse-shell signature. RESPONSE: identify the peer IP, " + "inspect the parent process (web/db parent = RCE), preserve /proc/ " + "before killing if you can, then kill the pid." + ), + "ld_preload": ( + "Library-injection detected (ld.so.preload or LD_PRELOAD from a writable " + "path) — a userland rootkit / credential-theft hook. RESPONSE: capture " + "the named .so and hash it, check package ownership, inspect the process " + "tree, and assume host compromise until cleared." + ), + "deleted_exe": ( + "A process is executing from a deleted or memfd-backed binary — fileless " + "malware that left no file on disk. RESPONSE: dump /proc//exe to " + "recover the binary BEFORE killing, capture maps and sockets." + ), + "new_listener": ( + "A listening socket appeared that wasn't present at baseline — possible " + "backdoor/bind shell. RESPONSE: identify the binary, confirm it's an " + "expected service, check for matching inbound connections." + ), + "new_suid": ( + "A new SUID/SGID binary appeared (critical if in a writable dir) — a " + "privilege-escalation persistence trick. RESPONSE: verify package " + "ownership; an unowned SUID binary in /tmp or /home is almost never " + "legitimate." + ), + "persistence": ( + "A persistence-relevant file (cron, systemd unit, authorized_keys, shell " + "rc) was modified. RESPONSE: diff against backup/version control, review " + "the change, and check auth logs for who made it." + ), + "egress": ( + "An interpreter is holding an outbound connection to a public IP — " + "possible C2 beacon or exfil. RESPONSE: resolve/geolocate the peer, check " + "reputation, inspect the process and its parentage." + ), +} + + +def _run(cmd: list[str], timeout: int = 8) -> str: + try: + res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + return res.stdout + except (OSError, subprocess.SubprocessError) as exc: + return f" ({' '.join(cmd)} failed: {exc})\n" + + +def _pid_detail(state: SystemState, pid: int) -> dict: + proc = state.process(pid) or Process(pid) + fds: dict[str, str] = {} + fd_dir = f"/proc/{pid}/fd" + try: + for name in sorted(os.listdir(fd_dir), key=lambda x: int(x) if x.isdigit() else 0): + try: + fds[name] = os.readlink(os.path.join(fd_dir, name)) + except OSError: + continue + except OSError: + pass + parent = state.process(proc.ppid) + return { + "pid": pid, + "comm": proc.comm, + "exe": proc.exe, + "cwd": proc.cwd, + "cmdline": proc.cmdline, + "ppid": proc.ppid, + "ppid_comm": parent.comm if parent else "", + "uid": proc.uid, + "ld_preload": proc.environ.get("LD_PRELOAD", ""), + "fds": dict(list(fds.items())[:40]), + } + + +def _format_text(report: dict, extras: dict[str, str]) -> str: + L: list[str] = [] + L.append("=== ENODIA SENTINEL ALERT ===") + L.append(f"Time: {report['time']}") + L.append(f"Host: {report['host']}") + L.append(f"Severity: {report['severity']}") + L.append("") + L.append("## Triggering detections") + for a in report["alerts"]: + L.append(f" [{a['severity']}] {a['signature']} — {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("") + L.append("## Flagged process detail") + if report["processes"]: + for d in report["processes"]: + L.append(f"### pid {d['pid']}") + L.append(f" comm: {d['comm']}") + L.append(f" exe: {d['exe']}") + L.append(f" cwd: {d['cwd']}") + L.append(f" cmdline: {d['cmdline']}") + L.append(f" ppid: {d['ppid']} ({d['ppid_comm']})") + L.append(f" uid: {d['uid']}") + L.append(f" LD_PRELOAD env: {d['ld_preload']}") + L.append(" open fds:") + for fd, tgt in d["fds"].items(): + L.append(f" {fd} -> {tgt}") + L.append("") + else: + L.append(" (no specific pid in alerts)") + L.append("") + for title, body in extras.items(): + L.append(f"## {title}") + L.append(body.rstrip("\n")) + L.append("") + return "\n".join(L) + "\n" + + +def capture(alerts: list[Alert], state: SystemState, cfg: Config) -> Path: + """Write text + JSON snapshot, append events.log, and notify. Returns path.""" + cfg.log_dir.mkdir(parents=True, exist_ok=True) + now = datetime.now().astimezone() + stamp = now.strftime("%Y%m%d-%H%M%S") + base = cfg.log_dir / f"alert-{stamp}" + + severity = max((a.severity for a in alerts), default=Severity.HIGH) + pids: list[int] = sorted({p for a in alerts for p in a.pids}) + + report = { + "time": now.isoformat(), + "host": os.uname().nodename, + "severity": str(severity), + "alerts": [a.to_dict() for a in alerts], + "processes": [_pid_detail(state, p) for p in pids], + } + + cutoff = int(time.time()) - 3600 + recent_files = [] + for root in cfg.watch_persistence: + try: + if os.path.isfile(root) and os.lstat(root).st_mtime > cutoff: + recent_files.append(root) + except OSError: + pass + + ps_out = _run(["ps", "-eo", "pid,ppid,user,etimes,pcpu,pmem,stat,comm", + "--sort=-pcpu"]) + extras = { + "Process tree (top by CPU)": "\n".join(ps_out.splitlines()[:40]), + "Sockets (ss -tanp)": "\n".join(_run(["ss", "-tanp"]).splitlines()[:80]), + "/etc/ld.so.preload": ( + Path("/etc/ld.so.preload").read_text() + if Path("/etc/ld.so.preload").is_file() + and Path("/etc/ld.so.preload").stat().st_size + else " (empty — good)" + ), + "Recently modified watched files (1h)": "\n".join(recent_files) or " (none)", + "Loaded kernel modules (head)": "\n".join(_run(["lsmod"]).splitlines()[:15]), + "Logins": _run(["who"]) + "--\n" + "\n".join(_run(["last", "-n", "5"]).splitlines()[:5]), + "Auth events (authpriv, 10m)": "\n".join(_run( + ["journalctl", "--since", "10 minutes ago", + "--facility=authpriv", "--no-pager"]).splitlines()[-20:]), + } + if cfg.capture_execve_bpftrace: + extras["bpftrace execve (3s)"] = _run([ + "bpftrace", "-e", + "tracepoint:syscalls:sys_enter_execve { " + "printf(\"%d %s %s\\n\", pid, comm, str(args->filename)); } " + "interval:s:3 { exit(); }", + ], timeout=6) + + text = _format_text(report, extras) + base.with_suffix(".log").write_text(text) + base.with_suffix(".json").write_text(json.dumps(report, indent=2)) + try: + base.with_suffix(".log").chmod(0o640) + base.with_suffix(".json").chmod(0o640) + except OSError: + pass + + sigs = ", ".join(dict.fromkeys(a.signature for a in alerts)) + with open(cfg.events_log, "a") as fh: + fh.write(f"{now.isoformat()} [{severity}] captured " + f"{base.with_suffix('.log')} — signatures: {sigs}\n") + + _notify(cfg, f"{severity}: {sigs}") + return base.with_suffix(".log") + + +def _notify(cfg: Config, body: str) -> None: + for user in cfg.notify_users: + try: + uid = int(subprocess.run(["id", "-u", user], capture_output=True, + text=True, timeout=3).stdout.strip()) + except (OSError, subprocess.SubprocessError, ValueError): + continue + env = dict(os.environ, + DBUS_SESSION_BUS_ADDRESS=f"unix:path=/run/user/{uid}/bus") + try: + subprocess.Popen( + ["sudo", "-u", user, "notify-send", "-u", cfg.notify_urgency, + "-a", "enodia-sentinel", "⚠ Enodia Sentinel alert", body], + env=env) + except OSError: + pass + + +def prune(cfg: Config) -> None: + snaps = sorted(cfg.log_dir.glob("alert-*.log"), + key=lambda p: p.stat().st_mtime, reverse=True) + if cfg.max_snapshot_age_days > 0: + cutoff = time.time() - cfg.max_snapshot_age_days * 86400 + for p in list(snaps): + if p.stat().st_mtime < cutoff: + _remove_pair(p) + snaps.remove(p) + if cfg.max_snapshots > 0: + for p in snaps[cfg.max_snapshots:]: + _remove_pair(p) + + +def _remove_pair(log_path: Path) -> None: + for p in (log_path, log_path.with_suffix(".json")): + try: + p.unlink() + except OSError: + pass diff --git a/enodia_sentinel/system.py b/enodia_sentinel/system.py new file mode 100644 index 0000000..61cf519 --- /dev/null +++ b/enodia_sentinel/system.py @@ -0,0 +1,270 @@ +"""A snapshot of live system state, gathered once per detector sweep. + +Detectors read from a single ``SystemState`` instance instead of each shelling +out, which is what keeps a sweep cheap. Everything here is also injectable: +tests construct a ``SystemState`` with fake processes/sockets and never touch +the real ``/proc``. +""" +from __future__ import annotations + +import os +import re +import stat +import subprocess +from dataclasses import dataclass, field +from functools import cached_property +from pathlib import Path + +_INO_RE = re.compile(r"\bino:(\d+)") +_USER_RE = re.compile(r'users:\(\("([^"]+)",pid=(\d+),fd=\d+') + + +@dataclass +class Process: + """Lazily-read view of a process via ``/proc/``.""" + + pid: int + + def _read(self, name: str) -> str: + try: + with open(f"/proc/{self.pid}/{name}", "rb") as fh: + return fh.read().decode("utf-8", "replace") + except OSError: + return "" + + @cached_property + def comm(self) -> str: + return self._read("comm").strip() + + @cached_property + def cmdline(self) -> str: + return self._read("cmdline").replace("\0", " ").strip() + + @cached_property + def exe(self) -> str: + try: + return os.readlink(f"/proc/{self.pid}/exe") + except OSError: + return "" + + @cached_property + def cwd(self) -> str: + try: + return os.readlink(f"/proc/{self.pid}/cwd") + except OSError: + return "" + + @cached_property + def environ(self) -> dict[str, str]: + raw = self._read("environ") + out: dict[str, str] = {} + for item in raw.split("\0"): + if "=" in item: + k, _, v = item.partition("=") + out[k] = v + return out + + @cached_property + def status(self) -> dict[str, str]: + out: dict[str, str] = {} + for line in self._read("status").splitlines(): + k, _, v = line.partition(":") + out[k] = v.strip() + return out + + @property + def ppid(self) -> int: + try: + return int(self.status.get("PPid", "0")) + except ValueError: + return 0 + + @property + def uid(self) -> int: + field0 = self.status.get("Uid", "0").split() + return int(field0[0]) if field0 else 0 + + def stdio_socket_inode(self) -> int | None: + """Socket inode if fd 0, 1, or 2 is a socket — the reverse-shell tell.""" + for fd in (0, 1, 2): + try: + target = os.readlink(f"/proc/{self.pid}/fd/{fd}") + except OSError: + continue + if target.startswith("socket:["): + return int(target[8:-1]) + return None + + +@dataclass(frozen=True) +class Socket: + state: str + local: str + peer: str + inode: int | None + comm: str + pid: int | None + + +class SystemState: + """One sweep's worth of cached system state. + + Construct empty for the real system; pass ``processes``/``sockets`` to + inject fakes in tests. + """ + + def __init__( + self, + processes: list[Process] | None = None, + sockets: list[Socket] | None = None, + *, + listener_baseline: set[str] | None = None, + suid_baseline: set[str] | None = None, + suid_binaries: list[str] | None = None, + persist_since: float | None = None, + ) -> None: + self._injected_procs = processes + self._injected_socks = sockets + # Context the daemon supplies for baseline-diffing detectors. Left as + # None when not applicable (e.g. before the grace window, or in tests), + # in which case those detectors yield nothing. + self.listener_baseline = listener_baseline + self.suid_baseline = suid_baseline + self.suid_binaries = suid_binaries + self.persist_since = persist_since + + def listener_keys(self) -> set[str]: + """Current listening sockets as ``port/comm`` identity keys.""" + keys = set() + for s in self.listening_sockets(): + _host, port = s.local.rpartition(":")[0], s.local.rpartition(":")[2] + keys.add(f"{port}/{s.comm or '?'}") + return keys + + # -- processes --------------------------------------------------------- + @cached_property + def processes(self) -> list[Process]: + if self._injected_procs is not None: + return self._injected_procs + procs = [] + for name in os.listdir("/proc"): + if name.isdigit(): + procs.append(Process(int(name))) + return procs + + @cached_property + def by_pid(self) -> dict[int, Process]: + return {p.pid: p for p in self.processes} + + def process(self, pid: int) -> Process | None: + return self.by_pid.get(pid) + + # -- sockets ----------------------------------------------------------- + @cached_property + def sockets(self) -> list[Socket]: + if self._injected_socks is not None: + return self._injected_socks + out: list[Socket] = [] + for args in (["-tanep"], ["-uanep"]): + out.extend(self._parse_ss(self._run_ss(args))) + return out + + @staticmethod + def _run_ss(extra: list[str]) -> str: + try: + res = subprocess.run( + ["ss", "-H", *extra], + capture_output=True, text=True, timeout=10, + ) + return res.stdout + except (OSError, subprocess.SubprocessError): + return "" + + @staticmethod + def _parse_ss(text: str) -> list[Socket]: + socks: list[Socket] = [] + for line in text.splitlines(): + parts = line.split() + if len(parts) < 5: + continue + state, _rq, _sq, local, peer = parts[:5] + rest = line + ino_m = _INO_RE.search(rest) + usr_m = _USER_RE.search(rest) + socks.append(Socket( + state=state, + local=local, + peer=peer, + inode=int(ino_m.group(1)) if ino_m else None, + comm=usr_m.group(1) if usr_m else "", + pid=int(usr_m.group(2)) if usr_m else None, + )) + return socks + + @cached_property + def net_peer_by_inode(self) -> dict[int, str]: + """Map socket inode -> "local -> peer" for every network socket.""" + out: dict[int, str] = {} + for s in self.sockets: + if s.inode is not None: + out[s.inode] = f"{s.local} -> {s.peer}" + return out + + def listening_sockets(self) -> list[Socket]: + return [s for s in self.sockets if s.state == "LISTEN"] + + def established_sockets(self) -> list[Socket]: + return [s for s in self.sockets if s.state == "ESTAB"] + + +# -- filesystem-wide SUID scan (expensive; the daemon gates how often) ------ + +_SETID = stat.S_ISUID | stat.S_ISGID + + +def _walk_suid(root: str, *, stay_on_device: bool) -> list[str]: + # os.scandir is much faster than os.walk + per-file lstat because DirEntry + # caches the stat from the directory read and is_dir() often avoids a stat. + try: + root_dev = os.stat(root).st_dev + except OSError: + return [] + found: list[str] = [] + stack = [root] + while stack: + try: + it = os.scandir(stack.pop()) + except OSError: + continue + with it: + for entry in it: + try: + if entry.is_dir(follow_symlinks=False): + if stay_on_device: + if entry.stat(follow_symlinks=False).st_dev != root_dev: + continue + stack.append(entry.path) + elif entry.is_file(follow_symlinks=False): + if entry.stat(follow_symlinks=False).st_mode & _SETID: + found.append(entry.path) + except OSError: + continue + return found + + +def scan_suid_binaries( + root: str = "/", extra_dirs: tuple[str, ...] = () +) -> list[str]: + """Find SUID/SGID regular files. + + Walks ``root`` staying on its device (the ``find -xdev`` equivalent, so it + doesn't wander into mounted media), then *also* walks each of ``extra_dirs`` + regardless of device — those are the small writable mounts (``/tmp``, + ``/dev/shm`` …) which are usually a separate tmpfs yet are exactly where a + malicious SUID binary gets dropped. + """ + found = set(_walk_suid(root, stay_on_device=True)) + for d in extra_dirs: + if os.path.isdir(d): + found.update(_walk_suid(d, stay_on_device=False)) + return sorted(found) diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD index f4521c6..a1d14cf 100644 --- a/packaging/PKGBUILD +++ b/packaging/PKGBUILD @@ -1,24 +1,27 @@ # Maintainer: Enodia pkgname=enodia-sentinel -pkgver=0.1.0 +pkgver=0.2.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=('MIT') -depends=('bash' 'iproute2' 'procps-ng') +depends=('python>=3.11' 'iproute2' 'procps-ng') optdepends=('bpftrace: execve tracing of short-lived processes in snapshots' 'libnotify: desktop notifications on alert') -backup=('etc/enodia-sentinel.conf') +backup=('etc/enodia-sentinel.toml') source=() sha256sums=() package() { cd "$startdir/.." - install -Dm755 src/sentinel.sh "$pkgdir/usr/bin/sentinel.sh" + # Install the stdlib-only package as a directory + launcher under /usr. + install -d "$pkgdir/usr/lib/enodia-sentinel" + cp -r enodia_sentinel "$pkgdir/usr/lib/enodia-sentinel/" + install -Dm755 packaging/enodia-sentinel.wrapper "$pkgdir/usr/bin/enodia-sentinel" install -Dm755 src/sentinel-redteam "$pkgdir/usr/bin/sentinel-redteam" install -Dm644 systemd/enodia-sentinel.service "$pkgdir/usr/lib/systemd/system/enodia-sentinel.service" - install -Dm644 config/enodia-sentinel.conf "$pkgdir/etc/enodia-sentinel.conf" + install -Dm644 config/enodia-sentinel.toml "$pkgdir/etc/enodia-sentinel.toml" install -Dm644 LICENSE "$pkgdir/usr/share/licenses/$pkgname/LICENSE" install -dm750 "$pkgdir/var/log/enodia-sentinel" } diff --git a/packaging/enodia-sentinel.wrapper b/packaging/enodia-sentinel.wrapper new file mode 100755 index 0000000..58b0e85 --- /dev/null +++ b/packaging/enodia-sentinel.wrapper @@ -0,0 +1,8 @@ +#!/bin/sh +# Launcher for the Enodia Sentinel Python package. +# Computes the install prefix from its own location so it works whether +# installed under /usr or /usr/local, with no pip/site-packages involvement. +here=$(dirname "$(readlink -f "$0")") # .../bin +prefix=${here%/bin} +export PYTHONPATH="${prefix}/lib/enodia-sentinel:${PYTHONPATH}" +exec python3 -B -m enodia_sentinel.cli "$@" diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..0248e28 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "enodia-sentinel" +version = "0.2.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" +license = { text = "MIT" } +authors = [{ name = "Enodia" }] +keywords = ["security", "ids", "hids", "detection", "linux", "forensics"] +# Zero runtime dependencies — stdlib only, by design (a system security +# daemon should not pull in a dependency tree it has to trust and update). +dependencies = [] + +[project.scripts] +enodia-sentinel = "enodia_sentinel.cli:main" + +[project.optional-dependencies] +dev = ["pytest"] + +[tool.setuptools.packages.find] +include = ["enodia_sentinel*"] diff --git a/src/sentinel-redteam b/src/sentinel-redteam index f5a6ce9..a85225a 100755 --- a/src/sentinel-redteam +++ b/src/sentinel-redteam @@ -35,14 +35,25 @@ trap cleanup EXIT INT TERM note() { echo -e "\n=== DRILL: $1 ===\n $2"; } +# A tiny TCP listener that accepts one connection and holds it open (no EOF, no +# data) for $HOLD seconds. Python is always present and, unlike some nc/ncat +# builds, won't half-close and tear the connection down before detection. +drill_listener() { # $1 = port + python3 -c 'import socket,sys,time +s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) +s.bind(("127.0.0.1",int(sys.argv[1]))); s.listen(1) +try: + c,_=s.accept(); time.sleep(int(sys.argv[2])) +except Exception: pass' "$1" "$HOLD" >/dev/null 2>&1 & + PIDS+=("$!") +} + drill_reverse_shell() { note reverse_shell "a bash with a TCP socket wired to its stdin/stdout (classic reverse shell)" # Local listener stands in for the attacker's box. No remote traffic leaves. - ( exec -a enodia-drill-listener nc -l 127.0.0.1 "$LISTEN_PORT" >/dev/null 2>&1 ) & - PIDS+=("$!") - sleep 1 # let the listener bind (single-shot nc; don't probe-connect it) + drill_listener "$LISTEN_PORT" + sleep 1 # bash whose fd 0/1/2 are the socket — exactly what `nc -e /bin/bash` produces. - # The shell retries the connect itself so it wins the listener's accept slot. # It then BLOCKS reading the socket (like a real reverse shell awaiting # commands) so the process holding the socket stays a 'bash', not a sleep. ( exec -a enodia-drill-rsh bash -c \ @@ -73,9 +84,8 @@ drill_deleted_exe() { drill_new_listener() { note new_listener "a new TCP listening port that wasn't in the baseline" - ( exec -a enodia-drill-bind nc -l 127.0.0.1 $((LISTEN_PORT+1)) >/dev/null 2>&1 ) & - PIDS+=("$!") - echo " -> opened listener on 127.0.0.1:$((LISTEN_PORT+1)) (pid $!)" + drill_listener $((LISTEN_PORT+1)) + echo " -> opened listener on 127.0.0.1:$((LISTEN_PORT+1)) (pid ${PIDS[-1]})" } drill_new_suid() { diff --git a/systemd/enodia-sentinel.service b/systemd/enodia-sentinel.service index b513389..7943170 100644 --- a/systemd/enodia-sentinel.service +++ b/systemd/enodia-sentinel.service @@ -5,24 +5,29 @@ Documentation=https://github.com/Enodia/enodia-sentinel [Service] Type=simple -ExecStart=/usr/local/bin/sentinel.sh +# /usr/bin/env resolves the launcher from PATH, so this works whether it was +# installed under /usr (package) or /usr/local (make install). +ExecStart=/usr/bin/env enodia-sentinel run Restart=on-failure RestartSec=5 -# Needs to read every process's /proc, sockets, and SUID inventory, so it runs -# as root — but with the surface area locked down. It only ever READS the system -# and WRITES to its own log dir. +# It only ever READS the system and WRITES to its own log dir. Lock the rest +# of the surface down: needs root to read every /proc and root-owned files, +# but with a minimal capability set and no ability to modify the system. ProtectSystem=strict ReadWritePaths=/var/log/enodia-sentinel ProtectHome=read-only NoNewPrivileges=yes ProtectKernelTunables=yes +ProtectKernelModules=yes ProtectControlGroups=yes RestrictSUIDSGID=yes MemoryDenyWriteExecute=yes LockPersonality=yes +RestrictNamespaces=yes # CAP_SYS_PTRACE: read other processes' /proc; CAP_DAC_READ_SEARCH: read # root-owned files (authorized_keys, ld.so.preload) regardless of perms. +CapabilityBoundingSet=CAP_SYS_PTRACE CAP_DAC_READ_SEARCH AmbientCapabilities=CAP_SYS_PTRACE CAP_DAC_READ_SEARCH [Install] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_detectors.py b/tests/test_detectors.py new file mode 100644 index 0000000..a3c035d --- /dev/null +++ b/tests/test_detectors.py @@ -0,0 +1,152 @@ +"""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() diff --git a/tests/test_netutil.py b/tests/test_netutil.py new file mode 100644 index 0000000..ecec404 --- /dev/null +++ b/tests/test_netutil.py @@ -0,0 +1,51 @@ +import unittest + +from enodia_sentinel import netutil + + +class TestPublicIP(unittest.TestCase): + def test_public(self): + self.assertTrue(netutil.is_public_ip("8.8.8.8")) + self.assertTrue(netutil.is_public_ip("1.1.1.1")) + + def test_private_and_special(self): + for addr in ("10.0.0.5", "192.168.1.1", "172.16.0.1", + "127.0.0.1", "169.254.1.1", "100.64.0.1", "::1"): + self.assertFalse(netutil.is_public_ip(addr), addr) + + def test_zone_and_brackets(self): + self.assertFalse(netutil.is_public_ip("10.2.0.2%proton0")) + self.assertFalse(netutil.is_public_ip("[::1]")) + + def test_garbage(self): + self.assertFalse(netutil.is_public_ip("not-an-ip")) + self.assertFalse(netutil.is_public_ip("")) + + +class TestCidr(unittest.TestCase): + def test_in_and_out(self): + cidrs = ["203.0.113.0/24", "10.0.0.0/8"] + self.assertTrue(netutil.ip_in_cidrs("203.0.113.55", cidrs)) + self.assertTrue(netutil.ip_in_cidrs("10.9.9.9", cidrs)) + self.assertFalse(netutil.ip_in_cidrs("8.8.8.8", cidrs)) + + def test_empty_list(self): + self.assertFalse(netutil.ip_in_cidrs("8.8.8.8", [])) + + +class TestSplitHostPort(unittest.TestCase): + def test_ipv4(self): + self.assertEqual(netutil.split_host_port("1.2.3.4:443"), ("1.2.3.4", "443")) + + def test_ipv6_bracket(self): + self.assertEqual(netutil.split_host_port("[2001:db8::1]:22"), + ("2001:db8::1", "22")) + + def test_zone(self): + host, port = netutil.split_host_port("10.2.0.2%proton0:61877") + self.assertEqual(host, "10.2.0.2%proton0") + self.assertEqual(port, "61877") + + +if __name__ == "__main__": + unittest.main()