Add event-driven memory syscall telemetry

enodia_sentinel/events/syscall_rules.py

@@ -0,0 +1,128 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Default rules for security-relevant syscall telemetry."""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Iterator
+from dataclasses import dataclass
+
+from ..alert import Alert, Severity
+from .syscall_event import SyscallEvent
+
+PROT_WRITE = 0x2
+PROT_EXEC = 0x4
+PTRACE_TRACEME = 0
+PTRACE_ATTACH = 16
+PTRACE_SEIZE = 0x4206
+PR_SET_SECCOMP = 22
+
+
+Predicate = Callable[[SyscallEvent], bool]
+
+
+@dataclass(frozen=True)
+class SyscallRule:
+    sid: int
+    msg: str
+    severity: Severity
+    classtype: str
+    syscalls: frozenset[str]
+    predicate: Predicate
+
+    def matches(self, ev: SyscallEvent) -> bool:
+        return ev.syscall in self.syscalls and self.predicate(ev)
+
+    def to_alert(self, ev: SyscallEvent) -> Alert:
+        detail = (
+            f"sid={self.sid} {self.msg} — pid={ev.pid} ppid={ev.ppid} "
+            f"comm={ev.comm} syscall={ev.syscall} "
+            f"args=[{ev.arg0:#x}, {ev.arg1:#x}, {ev.arg2:#x}, {ev.arg3:#x}]"
+        )
+        if ev.text:
+            detail += f" text=[{ev.text[:80]}]"
+        return Alert(
+            severity=self.severity,
+            signature=f"syscall_rule.{self.classtype}",
+            key=f"syscall:{self.sid}:{ev.pid}:{ev.syscall}:{ev.arg0:x}:{ev.arg2:x}",
+            detail=detail,
+            pids=(ev.pid,),
+            sid=self.sid,
+            classtype=self.classtype,
+        )
+
+
+def _prot_wx(prot: int) -> bool:
+    return bool((prot & PROT_WRITE) and (prot & PROT_EXEC))
+
+
+def _ptrace_sensitive(request: int) -> bool:
+    return request in {PTRACE_TRACEME, PTRACE_ATTACH, PTRACE_SEIZE}
+
+
+DEFAULT_SYSCALL_RULES: tuple[SyscallRule, ...] = (
+    SyscallRule(
+        sid=100060,
+        msg="mprotect made memory writable and executable",
+        severity=Severity.CRITICAL,
+        classtype="memory-obfuscation",
+        syscalls=frozenset({"mprotect"}),
+        predicate=lambda ev: _prot_wx(ev.arg2),
+    ),
+    SyscallRule(
+        sid=100061,
+        msg="mmap requested writable and executable memory",
+        severity=Severity.CRITICAL,
+        classtype="memory-obfuscation",
+        syscalls=frozenset({"mmap"}),
+        predicate=lambda ev: _prot_wx(ev.arg2),
+    ),
+    SyscallRule(
+        sid=100062,
+        msg="memfd_create used for anonymous in-memory file staging",
+        severity=Severity.MEDIUM,
+        classtype="fileless-execution",
+        syscalls=frozenset({"memfd_create"}),
+        predicate=lambda _ev: True,
+    ),
+    SyscallRule(
+        sid=100063,
+        msg="ptrace anti-debug or attach operation observed",
+        severity=Severity.HIGH,
+        classtype="anti-analysis",
+        syscalls=frozenset({"ptrace"}),
+        predicate=lambda ev: _ptrace_sensitive(ev.arg0),
+    ),
+    SyscallRule(
+        sid=100064,
+        msg="seccomp sandboxing call observed (possible anti-analysis hardening)",
+        severity=Severity.MEDIUM,
+        classtype="anti-analysis",
+        syscalls=frozenset({"prctl", "seccomp"}),
+        predicate=lambda ev: ev.syscall == "seccomp" or ev.arg0 == PR_SET_SECCOMP,
+    ),
+    SyscallRule(
+        sid=100065,
+        msg="cross-process memory read/write syscall observed",
+        severity=Severity.HIGH,
+        classtype="credential-access",
+        syscalls=frozenset({"process_vm_readv", "process_vm_writev"}),
+        predicate=lambda _ev: True,
+    ),
+    SyscallRule(
+        sid=100066,
+        msg="memory locking syscall observed (possible protected in-memory payload)",
+        severity=Severity.MEDIUM,
+        classtype="memory-obfuscation",
+        syscalls=frozenset({"mlock", "mlock2", "mlockall"}),
+        predicate=lambda _ev: True,
+    ),
+)
+
+
+class SyscallRuleEngine:
+    def __init__(self, rules: Iterable[SyscallRule] | None = None) -> None:
+        self.rules = list(rules if rules is not None else DEFAULT_SYSCALL_RULES)
+
+    def match(self, ev: SyscallEvent) -> Iterator[Alert]:
+        for rule in self.rules:
+            if rule.matches(ev):
+                yield rule.to_alert(ev)