From 3b037646d2321fd264d9ad127a7f484efe5e74dc Mon Sep 17 00:00:00 2001 From: Luna Date: Thu, 9 Jul 2026 06:04:30 -0700 Subject: [PATCH] Add typed host event egress rule --- CLAUDE.md | 12 ++- README.md | 10 +- docs/COMMAND_REFERENCE.md | 28 +++-- docs/ROADMAP.md | 14 +-- docs/RULES.md | 22 +++- docs/SPECIFICATION.md | 8 +- enodia_sentinel/events/host_event.py | 41 +++++++ enodia_sentinel/events/host_rules.py | 102 ++++++++++++++++++ enodia_sentinel/ruleops.py | 72 ++++++++++++- enodia_sentinel/sids.py | 3 + ...00067-interpreter-unusual-public-port.json | 14 +++ tests/test_rule_compat.py | 14 ++- tests/test_ruleops.py | 32 ++++++ tests/test_sid_coverage.py | 12 ++- tests/test_sids.py | 7 +- 15 files changed, 360 insertions(+), 31 deletions(-) create mode 100644 enodia_sentinel/events/host_event.py create mode 100644 enodia_sentinel/events/host_rules.py create mode 100644 tests/fixtures/sids/100067-interpreter-unusual-public-port.json diff --git a/CLAUDE.md b/CLAUDE.md index 1760599..13b0b35 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,7 +13,8 @@ network blocking or automatic host mutation. Primary functions: -- Detect: poll detectors plus optional eBPF exec/syscall rules. +- Detect: poll detectors, optional eBPF exec/syscall rules, and typed + host-event rule fixtures. - Prevent: posture findings plus reviewed containment/recovery plans. - Verify: FIM, package DB anchor, signed-package verification, rootcheck. - Investigate: text/JSON snapshots, incidents, HTTPS management console. @@ -39,6 +40,9 @@ Implemented detection coverage: web/database service spawning shells, and `curl|wget|sh` style staging. - eBPF syscall rules: RWX `mprotect`/`mmap`, `memfd_create`, sensitive `ptrace`, seccomp changes, cross-process memory access, and memory locking. +- Typed host-event rules: `tcp_connect` JSON fixtures can already exercise + interpreter egress to unusual public ports; live event sources are still + future work. - Anti-rootkit checks: hidden PIDs, `/proc` vs `ps` mismatches, hidden modules, hidden TCP/UDP/raw/special-protocol sockets, raw ICMP/SCTP-style channels, promiscuous interfaces, known LKM rootkit names, module taint, and kernel @@ -59,7 +63,8 @@ Implemented investigation/response state: 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. + operators test exec, syscall, and typed host-event JSON fixtures against the + rule engines. - `rules docs` emits generated Markdown rule docs; `docs/RULES.md` is the checked-in built-in rule reference with match fields, false positives, and drill guidance. @@ -125,7 +130,8 @@ Start with: - `daemon.py` runs sweep/cooldown/background tasks. - `system.py` builds one cached injectable view of processes and sockets. - `detectors/` contains pure poll detectors. -- `events/` contains optional eBPF exec/syscall monitoring and declarative rules. +- `events/` contains optional eBPF exec/syscall monitoring plus declarative + exec, syscall, and typed host-event rules. - `snapshot.py` writes forensic `.log` and `.json` evidence. - `incident.py` groups snapshots by process lineage and time window. - `respond.py` builds read-only response plans from incident evidence; the CLI diff --git a/README.md b/README.md index 7d54375..2505185 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ behind reviewed response workflows rather than silent remediation. | Function | Current capability | Direction | |---|---|---| -| Detect | Poll detectors + eBPF exec/syscall rules | More event sources and correlation | +| Detect | Poll detectors + eBPF exec/syscall rules + typed host-event fixtures | More live event sources and correlation | | Prevent | Posture checks + dry-run containment plans | Audited, explicit `--apply` workflows | | Verify | FIM, package DB anchor, signed-package checks | External anchors and signed evidence | | Investigate | Text/JSON snapshots, incidents, dashboard, triage | Richer timelines and evidence export | @@ -103,9 +103,11 @@ spawning a shell — webshell/RCE (`100003`), and `curl|sh`-style ingress tool transfer (`100004`). The syscall stream adds short-lived memory/anti-analysis 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: +cross-process memory access (`100065`), and memory locking (`100066`). Typed +host-event fixtures now cover `tcp_connect`, starting with interpreter egress +to unusual public ports (`100067`) while live network event capture remains a +v1.1 roadmap item. 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 diff --git a/docs/COMMAND_REFERENCE.md b/docs/COMMAND_REFERENCE.md index c943c0d..832f528 100644 --- a/docs/COMMAND_REFERENCE.md +++ b/docs/COMMAND_REFERENCE.md @@ -117,13 +117,13 @@ enodia-sentinel rules docs ``` 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. `rules docs` emits Markdown -rule documentation from the active rule metadata; the built-in snapshot lives in -[RULES.md](RULES.md). +`rules list` includes built-in exec, syscall, and typed host-event rules plus +configured exec rules from `exec_rules_file`. `rules show ` prints the rule +metadata and match conditions. `rules test ` loads an exec, syscall, +or host-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. +`rules docs` emits Markdown rule documentation from the active rule metadata; +the built-in snapshot lives in [RULES.md](RULES.md). Exec event JSON shape: @@ -151,6 +151,20 @@ Syscall event JSON shape: } ``` +Host event JSON shape: + +```json +{ + "event": "tcp_connect", + "pid": 42, + "ppid": 1, + "uid": 1000, + "comm": "python3", + "peer_ip": "8.8.8.8", + "peer_port": 4444 +} +``` + Exit code: - `0`: list/show succeeded, or `rules test` matched at least one rule. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 050bf06..5961bd6 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -128,9 +128,12 @@ Exit criteria: Purpose: reduce blind spots and connect related signals. -- Generalize the exec-rule engine into a typed host-event rule engine: - `exec`, `tcp_connect`, `bind`, `listen`, `accept`, `file_write`, - `chmod/chown`, `setuid/setgid`, `capset`, and module-load events. +- ✅ Add the first typed host-event rule family beside exec/syscall rules: + `rules test` accepts `tcp_connect` event JSON and ships SID `100067` for + interpreter egress to unusual public ports, with a replayable fixture. +- Continue generalizing the rule engine across more typed host events: + `bind`, `listen`, `accept`, `file_write`, `chmod/chown`, `setuid/setgid`, + `capset`, and module-load events. - Keep rules declarative and Snort-like: `sid`, `msg`, `severity`, `classtype`, `event`, and match fields such as process name, parent process, argv regex, path prefix, peer IP/port, UID/GID, @@ -151,11 +154,10 @@ Purpose: reduce blind spots and connect related signals. suspected trojaned binary with checksum cover-up. - web/database service spawning shell + tool transfer (`curl|wget|sh`) → CRITICAL likely RCE payload staging. -- Add high-signal living-off-the-land detections: +- Add additional high-signal living-off-the-land detections: `curl|wget | sh`, Python/perl/bash reverse-shell argv variants, `systemctl enable` from unusual ancestry, `chmod +s` outside package - transactions, interpreter egress to unusual public ports, and execution from - writable directories. + transactions, and execution from writable directories. - Add first-seen and rarity tracking for local network behavior: first public destination per process name, first listening port per binary, interpreter connections to non-standard ports, and long-lived low-byte diff --git a/docs/RULES.md b/docs/RULES.md index 78a007c..6961442 100644 --- a/docs/RULES.md +++ b/docs/RULES.md @@ -1,6 +1,6 @@ # Enodia Sentinel Event Rule Reference -Generated from the active exec/syscall rule defaults plus any configured `exec_rules_file` entries. +Generated from the active exec/syscall/host-event rule defaults plus any configured `exec_rules_file` entries. Use `enodia-sentinel rules list/show/test` to inspect rules and validate event fixtures locally. @@ -203,3 +203,23 @@ Expected false positives: - Realtime or performance-sensitive services using `mlock` intentionally. Drill or fixture: Use `rules test` with syscall `mlock`, `mlock2`, or `mlockall`. + +## SID 100067: Interpreter connected to an unusual public port + +- Event: `tcp_connect` +- Signature: `host_rule.suspicious-egress` +- Classtype: `suspicious-egress` +- Severity: `HIGH` +- Origin: `builtin` + +Match fields: +- `events`: `tcp_connect` +- `comm`: `ash`, `bash`, `curl`, `dash`, `fetch`, `ksh`, `lua`, `nc`, `ncat`, `netcat`, `node`, `nodejs`, `perl`, `php`, `python`, `python2`, `python3`, `ruby`, `sh`, `socat`, `wget`, `zsh` +- `peer_public`: `True` +- `peer_port_exclude`: `22`, `53`, `80`, `123`, `443`, `853` + +Expected false positives: +- Interactive admin scripts that intentionally connect to a non-standard public service. +- Developer tooling using interpreters for custom APIs on high ports. + +Drill or fixture: Use `rules test` with a `tcp_connect` event whose `comm` is an interpreter and whose public `peer_port` is not a common service port. diff --git a/docs/SPECIFICATION.md b/docs/SPECIFICATION.md index 8cc08f0..f6330ee 100644 --- a/docs/SPECIFICATION.md +++ b/docs/SPECIFICATION.md @@ -70,7 +70,9 @@ Optional bcc eBPF monitors feed event rule engines. The `execve` stream closes the polling gap for short-lived commands and supports custom TOML rules without code changes. The syscall stream watches short-lived memory and anti-analysis behavior: RWX `mprotect`/`mmap`, `memfd_create`, sensitive `ptrace`, seccomp, -cross-process memory access, and memory locking. Failure is fail-safe: if a +cross-process memory access, and memory locking. Typed host-event rules can be +tested from JSON fixtures before live sources exist; `tcp_connect` currently +covers interpreter egress to unusual public ports. Failure is fail-safe: if a probe cannot load, polling continues. ### Integrity and Tamper-Evidence @@ -136,8 +138,8 @@ incident. ### Detect -- More event sources: `tcp_connect`, `bind`, `accept`, file-write events for - persistence paths, module loading, privilege transitions, and LSM hooks. +- More live event sources: `tcp_connect`, `bind`, `accept`, file-write events + for persistence paths, module loading, privilege transitions, and LSM hooks. - Correlation across signals: "web service spawned shell" + "new listener" + "persistence write" should become one incident, not three unrelated alerts. - Optional YARA-style and Sigma-like local rules if they can stay dependency diff --git a/enodia_sentinel/events/host_event.py b/enodia_sentinel/events/host_event.py new file mode 100644 index 0000000..1319636 --- /dev/null +++ b/enodia_sentinel/events/host_event.py @@ -0,0 +1,41 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Typed host event shared by non-exec, non-syscall event rules.""" +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class HostEvent: + event: str + pid: int + ppid: int + uid: int + comm: str + parent_comm: str = "" + path: str = "" + argv: tuple[str, ...] = field(default_factory=tuple) + peer_ip: str = "" + peer_port: int = 0 + local_ip: str = "" + local_port: int = 0 + + @property + def argv_str(self) -> str: + return " ".join((self.path, *self.argv)).strip() + + def to_dict(self) -> dict: + return { + "event": self.event, + "pid": self.pid, + "ppid": self.ppid, + "uid": self.uid, + "comm": self.comm, + "parent_comm": self.parent_comm, + "path": self.path, + "argv": list(self.argv), + "peer_ip": self.peer_ip, + "peer_port": self.peer_port, + "local_ip": self.local_ip, + "local_port": self.local_port, + } diff --git a/enodia_sentinel/events/host_rules.py b/enodia_sentinel/events/host_rules.py new file mode 100644 index 0000000..7129e2d --- /dev/null +++ b/enodia_sentinel/events/host_rules.py @@ -0,0 +1,102 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +"""Declarative rules for typed host events beyond exec/syscall telemetry.""" +from __future__ import annotations + +import re +from collections.abc import Iterable, Iterator +from dataclasses import dataclass + +from ..alert import Alert, Severity +from ..netutil import is_public_ip +from .host_event import HostEvent + + +_INTERPRETERS = frozenset( + "sh bash dash zsh ksh ash python python2 python3 perl ruby php lua " + "node nodejs nc ncat netcat socat curl wget fetch".split() +) +_COMMON_PUBLIC_PORTS = frozenset({22, 53, 80, 123, 443, 853}) + + +@dataclass(frozen=True) +class HostRule: + sid: int + msg: str + severity: Severity + classtype: str + events: frozenset[str] + comm: frozenset[str] = frozenset() + parent_comm: frozenset[str] = frozenset() + peer_public: bool | None = None + peer_ports: frozenset[int] = frozenset() + peer_port_exclude: frozenset[int] = frozenset() + argv_regex: str | None = None + + def __post_init__(self) -> None: + if not self.events: + raise ValueError(f"rule sid={self.sid} has no event types") + if not any(( + self.comm, self.parent_comm, self.peer_public is not None, + self.peer_ports, self.peer_port_exclude, self.argv_regex, + )): + raise ValueError(f"rule sid={self.sid} has no match conditions") + if self.argv_regex is not None: + object.__setattr__(self, "_argv_re", + re.compile(self.argv_regex, re.IGNORECASE)) + else: + object.__setattr__(self, "_argv_re", None) + + def matches(self, ev: HostEvent) -> bool: + if ev.event not in self.events: + return False + if self.comm and ev.comm not in self.comm: + return False + if self.parent_comm and ev.parent_comm not in self.parent_comm: + return False + if self.peer_public is not None and is_public_ip(ev.peer_ip) != self.peer_public: + return False + if self.peer_ports and ev.peer_port not in self.peer_ports: + return False + if self.peer_port_exclude and ev.peer_port in self.peer_port_exclude: + return False + if self._argv_re is not None and not self._argv_re.search(ev.argv_str): + return False + return True + + def to_alert(self, ev: HostEvent) -> Alert: + return Alert( + severity=self.severity, + signature=f"host_rule.{self.classtype}", + key=f"host:{self.sid}:{ev.event}:{ev.pid}:{ev.peer_ip}:{ev.peer_port}", + detail=( + f"sid={self.sid} {self.msg} - pid={ev.pid} ppid={ev.ppid} " + f"comm={ev.comm} event={ev.event} peer={ev.peer_ip}:{ev.peer_port}" + ), + pids=(ev.pid,), + sid=self.sid, + classtype=self.classtype, + ) + + +DEFAULT_HOST_RULES: tuple[HostRule, ...] = ( + HostRule( + sid=100067, + msg="Interpreter connected to an unusual public port", + severity=Severity.HIGH, + classtype="suspicious-egress", + events=frozenset({"tcp_connect"}), + comm=_INTERPRETERS, + peer_public=True, + peer_port_exclude=_COMMON_PUBLIC_PORTS, + ), +) + + +class HostRuleEngine: + def __init__(self, rules: Iterable[HostRule] | None = None) -> None: + self.rules = list(rules if rules is not None else DEFAULT_HOST_RULES) + + def match(self, ev: HostEvent) -> Iterator[Alert]: + for rule in self.rules: + if rule.matches(ev): + yield rule.to_alert(ev) diff --git a/enodia_sentinel/ruleops.py b/enodia_sentinel/ruleops.py index e7863a3..98e9fd4 100644 --- a/enodia_sentinel/ruleops.py +++ b/enodia_sentinel/ruleops.py @@ -9,6 +9,8 @@ from typing import Any from .alert import Alert from .config import Config from .events.exec_event import ExecEvent +from .events.host_event import HostEvent +from .events.host_rules import DEFAULT_HOST_RULES, HostRule, HostRuleEngine from .events.rules import DEFAULT_EXEC_RULES, ExecRule, ExecRuleEngine from .events.syscall_event import SyscallEvent from .events.syscall_rules import ( @@ -22,6 +24,7 @@ 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} + host_builtin = {r.sid for r in DEFAULT_HOST_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 @@ -30,6 +33,10 @@ def list_rules(cfg: Config) -> list[dict[str, Any]]: _syscall_rule_record(r, "builtin" if r.sid in syscall_builtin else "configured") for r in SyscallRuleEngine().rules ) + records.extend( + _host_rule_record(r, "builtin" if r.sid in host_builtin else "configured") + for r in HostRuleEngine().rules + ) return sorted(records, key=lambda r: (int(r["sid"]), str(r["event"]))) @@ -45,8 +52,8 @@ def render_markdown(cfg: Config) -> str: lines = [ "# Enodia Sentinel Event Rule Reference", "", - "Generated from the active exec/syscall rule defaults plus any configured " - "`exec_rules_file` entries.", + "Generated from the active exec/syscall/host-event rule defaults plus " + "any configured `exec_rules_file` entries.", "", "Use `enodia-sentinel rules list/show/test` to inspect rules and validate " "event fixtures locally.", @@ -98,7 +105,9 @@ def test_event(cfg: Config, event: dict[str, Any]) -> list[Alert]: 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") + if kind == "host": + return list(HostRuleEngine().match(_host_event(event))) + raise ValueError("event JSON must be an exec, syscall, or host event") def _exec_rule_record(rule: ExecRule, origin: str) -> dict[str, Any]: @@ -141,6 +150,34 @@ def _syscall_rule_record(rule: SyscallRule, origin: str) -> dict[str, Any]: } +def _host_rule_record(rule: HostRule, origin: str) -> dict[str, Any]: + conditions: dict[str, Any] = { + "events": sorted(rule.events), + } + if rule.comm: + conditions["comm"] = sorted(rule.comm) + if rule.parent_comm: + conditions["parent_comm"] = sorted(rule.parent_comm) + if rule.peer_public is not None: + conditions["peer_public"] = rule.peer_public + if rule.peer_ports: + conditions["peer_ports"] = sorted(rule.peer_ports) + if rule.peer_port_exclude: + conditions["peer_port_exclude"] = sorted(rule.peer_port_exclude) + if rule.argv_regex: + conditions["argv_regex"] = rule.argv_regex + return { + "sid": rule.sid, + "event": ",".join(sorted(rule.events)), + "origin": origin, + "severity": str(rule.severity), + "classtype": rule.classtype, + "signature": f"host_rule.{rule.classtype}", + "msg": rule.msg, + "conditions": conditions, + } + + _RULE_DOCS: dict[int, dict[str, Any]] = { 100001: { "false_positives": [ @@ -219,6 +256,13 @@ _RULE_DOCS: dict[int, dict[str, Any]] = { ], "drill": "Use `rules test` with syscall `mlock`, `mlock2`, or `mlockall`.", }, + 100067: { + "false_positives": [ + "Interactive admin scripts that intentionally connect to a non-standard public service.", + "Developer tooling using interpreters for custom APIs on high ports.", + ], + "drill": "Use `rules test` with a `tcp_connect` event whose `comm` is an interpreter and whose public `peer_port` is not a common service port.", + }, } @@ -246,6 +290,11 @@ def _event_kind(event: dict[str, Any]) -> str: return "exec" if explicit in {"syscall", "sys"}: return "syscall" + if explicit in { + "tcp_connect", "bind", "listen", "accept", "file_write", + "chmod", "chown", "setuid", "setgid", "capset", "module_load", + }: + return "host" if "filename" in event or "argv" in event or "parent_comm" in event: return "exec" if "syscall" in event or "args" in event: @@ -283,6 +332,23 @@ def _syscall_event(event: dict[str, Any]) -> SyscallEvent: ) +def _host_event(event: dict[str, Any]) -> HostEvent: + return HostEvent( + event=str(event.get("event") or event.get("type") or ""), + 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", "")), + parent_comm=str(event.get("parent_comm", "")), + path=str(event.get("path", "")), + argv=tuple(str(a) for a in event.get("argv", ())), + peer_ip=str(event.get("peer_ip", event.get("remote_ip", ""))), + peer_port=_to_int(event.get("peer_port", event.get("remote_port", 0))), + local_ip=str(event.get("local_ip", "")), + local_port=_to_int(event.get("local_port", 0)), + ) + + def _to_int(value: Any) -> int: if isinstance(value, str): return int(value, 0) diff --git a/enodia_sentinel/sids.py b/enodia_sentinel/sids.py index 26003d5..f79b228 100644 --- a/enodia_sentinel/sids.py +++ b/enodia_sentinel/sids.py @@ -17,6 +17,7 @@ from .detectors import ( stealth_network, ) from .events.rules import DEFAULT_EXEC_RULES +from .events.host_rules import DEFAULT_HOST_RULES from .events.syscall_rules import DEFAULT_SYSCALL_RULES @@ -42,6 +43,8 @@ BUILTIN_SIDS: tuple[SidInfo, ...] = ( "enodia_sentinel.events.rules"), *_rule_rows(DEFAULT_SYSCALL_RULES, "syscall_rule", "enodia_sentinel.events.syscall_rules"), + *_rule_rows(DEFAULT_HOST_RULES, "host_rule", + "enodia_sentinel.events.host_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"), diff --git a/tests/fixtures/sids/100067-interpreter-unusual-public-port.json b/tests/fixtures/sids/100067-interpreter-unusual-public-port.json new file mode 100644 index 0000000..81bd620 --- /dev/null +++ b/tests/fixtures/sids/100067-interpreter-unusual-public-port.json @@ -0,0 +1,14 @@ +{ + "event": "tcp_connect", + "pid": 4242, + "ppid": 1, + "uid": 1000, + "comm": "python3", + "parent_comm": "bash", + "path": "/usr/bin/python3", + "argv": ["-c", "connect back"], + "peer_ip": "8.8.8.8", + "peer_port": 4444, + "local_ip": "10.0.0.5", + "local_port": 51510 +} diff --git a/tests/test_rule_compat.py b/tests/test_rule_compat.py index 793341d..ad6f38a 100644 --- a/tests/test_rule_compat.py +++ b/tests/test_rule_compat.py @@ -71,9 +71,21 @@ class TestRuleEventCompatibility(unittest.TestCase): } self.assertIn(100061, self._sids(event)) + def test_host_event_accepts_tcp_connect_shape(self): + event = { + "event": "tcp_connect", + "pid": "4242", + "ppid": "1", + "uid": "1000", + "comm": "python3", + "remote_ip": "8.8.8.8", + "remote_port": "4444", + } + self.assertIn(100067, self._sids(event)) + def test_unknown_event_shape_still_errors(self): with self.assertRaisesRegex( - ValueError, "event JSON must be an exec or syscall event" + ValueError, "event JSON must be an exec, syscall, or host event" ): ruleops.test_event(self.cfg, {"pid": 4242}) diff --git a/tests/test_ruleops.py b/tests/test_ruleops.py index fab8729..b1648fe 100644 --- a/tests/test_ruleops.py +++ b/tests/test_ruleops.py @@ -18,9 +18,13 @@ class TestRuleOps(unittest.TestCase): sids = {r["sid"] for r in rules} self.assertIn(100001, sids) self.assertIn(100060, sids) + self.assertIn(100067, 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"]) + host_rule = next(r for r in rules if r["sid"] == 100067) + self.assertEqual(host_rule["event"], "tcp_connect") + self.assertIn("peer_port_exclude", host_rule["conditions"]) def test_configured_exec_rule_is_listed(self): with tempfile.TemporaryDirectory() as d: @@ -66,6 +70,33 @@ exec_comm = ["id"] alerts = ruleops.test_event(cfg, event) self.assertIn(100060, {a.sid for a in alerts}) + def test_matches_host_event_json(self): + cfg = Config() + event = { + "event": "tcp_connect", + "pid": 42, + "ppid": 1, + "uid": 1000, + "comm": "python3", + "peer_ip": "8.8.8.8", + "peer_port": 4444, + } + alerts = ruleops.test_event(cfg, event) + self.assertIn(100067, {a.sid for a in alerts}) + + def test_host_event_common_public_port_no_match(self): + cfg = Config() + event = { + "event": "tcp_connect", + "pid": 42, + "ppid": 1, + "uid": 1000, + "comm": "python3", + "peer_ip": "8.8.8.8", + "peer_port": 443, + } + self.assertEqual(ruleops.test_event(cfg, event), []) + def test_render_markdown_documents_rules(self): text = ruleops.render_markdown(Config()) self.assertIn("# Enodia Sentinel Event Rule Reference", text) @@ -136,6 +167,7 @@ class TestRulesCli(unittest.TestCase): self.assertEqual(code, 0) self.assertIn("Enodia Sentinel Event Rule Reference", out) self.assertIn("SID 100060", out) + self.assertIn("SID 100067", out) self.assertIn("Expected false positives", out) diff --git a/tests/test_sid_coverage.py b/tests/test_sid_coverage.py index fee6755..e8d7feb 100644 --- a/tests/test_sid_coverage.py +++ b/tests/test_sid_coverage.py @@ -32,7 +32,11 @@ def fixture_sids() -> dict[int, set[int]]: 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") + wanted = ( + sids.engine_sids("exec_rule") + | sids.engine_sids("syscall_rule") + | sids.engine_sids("host_rule") + ) for sid in sorted(wanted): self.assertIn(sid, covered, f"no fixture file for sid {sid} in {FIXTURE_DIR}") @@ -54,7 +58,11 @@ def drill_sids() -> dict[int, set[int]]: 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") + event_sids = ( + sids.engine_sids("exec_rule") + | sids.engine_sids("syscall_rule") + | sids.engine_sids("host_rule") + ) wanted = sids.all_sids() - event_sids self.assertEqual(set(covered), wanted) for sid in sorted(wanted): diff --git a/tests/test_sids.py b/tests/test_sids.py index 6ea201a..ce7ed1a 100644 --- a/tests/test_sids.py +++ b/tests/test_sids.py @@ -10,6 +10,7 @@ from enodia_sentinel.detectors import ( stealth_network, ) from enodia_sentinel.events.rules import DEFAULT_EXEC_RULES +from enodia_sentinel.events.host_rules import DEFAULT_HOST_RULES from enodia_sentinel.events.syscall_rules import DEFAULT_SYSCALL_RULES @@ -19,7 +20,7 @@ class TestRegistry(unittest.TestCase): self.assertEqual(len(nums), len(set(nums))) def test_count_and_helpers(self): - self.assertEqual(len(sids.BUILTIN_SIDS), 52) + self.assertEqual(len(sids.BUILTIN_SIDS), 53) self.assertEqual( sids.all_sids(), frozenset(s.sid for s in sids.BUILTIN_SIDS) ) @@ -37,6 +38,10 @@ class TestRegistry(unittest.TestCase): sids.engine_sids("syscall_rule"), frozenset(r.sid for r in DEFAULT_SYSCALL_RULES), ) + self.assertEqual( + sids.engine_sids("host_rule"), + frozenset(r.sid for r in DEFAULT_HOST_RULES), + ) def test_registry_uses_the_source_constants(self): by_sid = {s.sid: s for s in sids.BUILTIN_SIDS}