enodia-sentinal/enodia_sentinel/events/rules.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""A declarative, Snort-style rule engine for execve events.

Detection logic lives in *data*, not code. Each ``ExecRule`` is the host-event
analogue of a Snort rule: a stable ``sid``, a human ``msg``, a ``classtype`` and
``severity``, plus match conditions on the executed path, the program, the
parent process, and the argument vector. The engine just walks the rules.

A rule fires when **every condition it specifies** matches (AND); unspecified
conditions are wildcards. A rule with no conditions is rejected at load time so
it can't match everything by accident.

Default rules ship below; operators can add more via a TOML file
(``[[exec_rules]]`` tables) referenced by ``exec_rules_file`` in the config.
"""
from __future__ import annotations

import re
import tomllib
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from pathlib import Path

from ..alert import Alert, Severity
from .exec_event import ExecEvent

_SEVERITY = {
    "MEDIUM": Severity.MEDIUM,
    "HIGH": Severity.HIGH,
    "CRITICAL": Severity.CRITICAL,
}


@dataclass(frozen=True)
class ExecRule:
    sid: int
    msg: str
    severity: Severity
    classtype: str
    # Conditions (all that are set must match):
    path_prefixes: tuple[str, ...] = ()       # filename startswith any
    exec_comm: frozenset[str] = frozenset()   # basename(filename) in set
    parent_comm: frozenset[str] = frozenset() # caller comm in set
    argv_regex: str | None = None             # re.search over argv_str
    parent_exclude: frozenset[str] = frozenset()  # never fire for these parents

    def __post_init__(self) -> None:
        if not any((self.path_prefixes, self.exec_comm,
                    self.parent_comm, self.argv_regex)):
            raise ValueError(f"rule sid={self.sid} has no match conditions")
        if self.argv_regex is not None:
            # compile once; store on the frozen instance via object.__setattr__
            object.__setattr__(self, "_argv_re", re.compile(self.argv_regex,
                                                            re.IGNORECASE))
        else:
            object.__setattr__(self, "_argv_re", None)

    def matches(self, ev: ExecEvent) -> bool:
        if ev.parent_comm in self.parent_exclude:
            return False
        if self.path_prefixes and not ev.filename.startswith(self.path_prefixes):
            return False
        if self.exec_comm and ev.exec_comm not in self.exec_comm:
            return False
        if self.parent_comm and ev.parent_comm not in self.parent_comm:
            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: ExecEvent) -> Alert:
        return Alert(
            severity=self.severity,
            signature=f"exec_rule.{self.classtype}",
            key=f"exec:{self.sid}:{ev.pid}",
            detail=(
                f"sid={self.sid} {self.msg} — pid={ev.pid} ppid={ev.ppid} "
                f"parent={ev.parent_comm} exec={ev.filename} "
                f"argv=[{ev.argv_str[:120]}]"
            ),
            pids=(ev.pid,),
            sid=self.sid,
            classtype=self.classtype,
        )


# --- shipped default rules -------------------------------------------------
# SIDs 100001–100999 are reserved for Enodia's built-in exec rules.

_INTERP = frozenset(
    "sh bash dash zsh ksh ash python python2 python3 perl ruby php lua "
    "nc ncat netcat socat".split()
)
_WEB_DB_SERVERS = frozenset(
    "nginx apache apache2 httpd php-fpm php php7 php8 lighttpd caddy "
    "node nodejs tomcat catalina mysqld mariadbd postgres redis-server".split()
)

DEFAULT_EXEC_RULES: tuple[ExecRule, ...] = (
    ExecRule(
        sid=100001,
        msg="Program executed from a world-writable directory",
        severity=Severity.CRITICAL,
        classtype="fileless-execution",
        path_prefixes=("/tmp/", "/dev/shm/", "/var/tmp/"),
    ),
    ExecRule(
        sid=100002,
        msg="Reverse-shell command pattern in execve arguments",
        severity=Severity.CRITICAL,
        classtype="c2-reverse-shell",
        argv_regex=(
            r"/dev/(tcp|udp)/"
            r"|\b(ba|da|z)?sh\b[^|]*\s-i\b"
            r"|\bn(c|cat|etcat)\b.*\s-e\b"
            r"|\bsocat\b.*\bexec"
            r"|python[0-9.]*\b.*(pty\.spawn|socket\.socket)"
            r"|perl\b.*\bSocket\b"
        ),
    ),
    ExecRule(
        sid=100003,
        msg="Web/DB service spawned a shell or interpreter (possible RCE/webshell)",
        severity=Severity.CRITICAL,
        classtype="web-rce",
        parent_comm=_WEB_DB_SERVERS,
        exec_comm=_INTERP,
    ),
    ExecRule(
        sid=100004,
        msg="Download piped directly to a shell (ingress tool transfer)",
        severity=Severity.HIGH,
        classtype="ingress-tool-transfer",
        argv_regex=r"\b(curl|wget|fetch)\b.*\|\s*(ba|da|z)?sh\b",
    ),
)


class ExecRuleEngine:
    """Holds a rule set and matches events against it."""

    def __init__(self, rules: Iterable[ExecRule] | None = None) -> None:
        self.rules: list[ExecRule] = list(
            rules if rules is not None else DEFAULT_EXEC_RULES
        )

    def match(self, ev: ExecEvent) -> Iterator[Alert]:
        for rule in self.rules:
            if rule.matches(ev):
                yield rule.to_alert(ev)

    @classmethod
    def load(cls, path: str | Path | None) -> "ExecRuleEngine":
        rules = list(DEFAULT_EXEC_RULES)
        if path and Path(path).is_file():
            rules.extend(_load_toml_rules(Path(path)))
        return cls(rules)


def _load_toml_rules(path: Path) -> list[ExecRule]:
    with open(path, "rb") as fh:
        data = tomllib.load(fh)
    out: list[ExecRule] = []
    for r in data.get("exec_rules", []):
        out.append(ExecRule(
            sid=int(r["sid"]),
            msg=r.get("msg", ""),
            severity=_SEVERITY.get(str(r.get("severity", "HIGH")).upper(),
                                   Severity.HIGH),
            classtype=r.get("classtype", "uncategorized"),
            path_prefixes=tuple(r.get("path_prefixes", ())),
            exec_comm=frozenset(r.get("exec_comm", ())),
            parent_comm=frozenset(r.get("parent_comm", ())),
            argv_regex=r.get("argv_regex"),
            parent_exclude=frozenset(r.get("parent_exclude", ())),
        ))
    return out