145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
# 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)
|