# 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)