feat(sources): add execve, ptrace, sensitive_file_open BPF checks

sensitive_file_open uses compile-time unrolled byte comparison to
avoid BPF loops — compatible with all kernel versions.

Author: dfa1

sources/predefined_programs.py

@@ -182,6 +182,111 @@ def module_load() -> str:
 """
 
 
+def execve() -> str:
+    """Trace all execve syscalls (process execution)."""
+    return """
+#include <uapi/linux/ptrace.h>
+#include <linux/sched.h>
+
+#define TASK_COMM_LEN 16
+#define PAYLOAD_LEN   256
+
+struct event_t {
+    u64  ts;
+    u32  pid;
+    char comm[TASK_COMM_LEN];
+    char payload[PAYLOAD_LEN];
+};
+
+BPF_PERF_OUTPUT(events);
+
+TRACEPOINT_PROBE(syscalls, sys_enter_execve) {
+    struct event_t ev = {};
+    ev.ts  = bpf_ktime_get_ns();
+    ev.pid = bpf_get_current_pid_tgid() >> 32;
+    bpf_get_current_comm(&ev.comm, sizeof(ev.comm));
+    bpf_probe_read_user_str(ev.payload, sizeof(ev.payload), args->filename);
+    events.perf_submit(args, &ev, sizeof(ev));
+    return 0;
+}
+"""
+
+
+def ptrace() -> str:
+    """Trace ptrace calls via security_ptrace_access_check (process injection / debugging)."""
+    return """
+#include <uapi/linux/ptrace.h>
+#include <linux/sched.h>
+
+#define TASK_COMM_LEN 16
+#define PAYLOAD_LEN   256
+
+struct event_t {
+    u64  ts;
+    u32  pid;
+    char comm[TASK_COMM_LEN];
+    char payload[PAYLOAD_LEN];
+};
+
+BPF_PERF_OUTPUT(events);
+
+int kprobe__security_ptrace_access_check(struct pt_regs *ctx, struct task_struct *child, unsigned int mode) {
+    struct event_t ev = {};
+    ev.ts  = bpf_ktime_get_ns();
+    ev.pid = bpf_get_current_pid_tgid() >> 32;
+    bpf_get_current_comm(&ev.comm, sizeof(ev.comm));
+    bpf_probe_read_kernel_str(ev.payload, sizeof(ev.payload), child->comm);
+    events.perf_submit(ctx, &ev, sizeof(ev));
+    return 0;
+}
+"""
+
+
+def sensitive_file_open(path: str) -> str:
+    """Trace opens of a specific absolute file path (e.g. /etc/shadow).
+
+    Uses compile-time unrolled byte comparison — no BPF loops, works on all kernel versions.
+    """
+    if not path.startswith("/"):
+        raise ValueError(f"path must be absolute: {path!r}")
+    path_bytes = path.encode("utf-8")
+    if len(path_bytes) > 255:
+        raise ValueError(f"path too long (max 255 bytes): {path!r}")
+    comparisons = " ||\n        ".join(
+        f"fname[{i}] != {b}" for i, b in enumerate(path_bytes)
+    ) + f" ||\n        fname[{len(path_bytes)}] != 0"
+    return f"""
+#include <uapi/linux/ptrace.h>
+#include <linux/sched.h>
+
+#define TASK_COMM_LEN 16
+#define PAYLOAD_LEN   256
+
+struct event_t {{
+    u64  ts;
+    u32  pid;
+    char comm[TASK_COMM_LEN];
+    char payload[PAYLOAD_LEN];
+}};
+
+BPF_PERF_OUTPUT(events);
+
+TRACEPOINT_PROBE(syscalls, sys_enter_openat) {{
+    char fname[PAYLOAD_LEN];
+    bpf_probe_read_user_str(fname, sizeof(fname), args->filename);
+    if ({comparisons}) return 0;
+
+    struct event_t ev = {{}};
+    ev.ts  = bpf_ktime_get_ns();
+    ev.pid = bpf_get_current_pid_tgid() >> 32;
+    bpf_get_current_comm(&ev.comm, sizeof(ev.comm));
+    __builtin_memcpy(ev.payload, fname, sizeof(ev.payload));
+    events.perf_submit(args, &ev, sizeof(ev));
+    return 0;
+}}
+"""
+
+
 def ip_host(addr: str) -> str:
     """Trace TCP connections to or from the given IPv4 address (dotted decimal)."""
     octets = [int(o) for o in addr.split(".")]

tests/unit/test_predefined_programs.py

@@ -2,9 +2,12 @@
 
 from sources.predefined_programs import (
     commit_creds,
+    execve,
     icmp,
     ip_host,
     module_load,
+    ptrace,
+    sensitive_file_open,
     suid_exec,
     tcp_port,
     udp_port,
@@ -75,3 +78,50 @@ def test_module_load_returns_string() -> None:
 
 def test_module_load_is_deterministic() -> None:
     assert module_load() == module_load()
+
+
+# --- execve ---
+
+
+def test_execve_returns_string() -> None:
+    assert isinstance(execve(), str)
+
+
+def test_execve_is_deterministic() -> None:
+    assert execve() == execve()
+
+
+# --- ptrace ---
+
+
+def test_ptrace_returns_string() -> None:
+    assert isinstance(ptrace(), str)
+
+
+def test_ptrace_is_deterministic() -> None:
+    assert ptrace() == ptrace()
+
+
+# --- sensitive_file_open ---
+
+
+def test_sensitive_file_open_returns_string() -> None:
+    assert isinstance(sensitive_file_open("/etc/shadow"), str)
+
+
+def test_sensitive_file_open_different_paths_differ() -> None:
+    assert sensitive_file_open("/etc/shadow") != sensitive_file_open("/etc/passwd")
+
+
+def test_sensitive_file_open_is_deterministic() -> None:
+    assert sensitive_file_open("/etc/shadow") == sensitive_file_open("/etc/shadow")
+
+
+def test_sensitive_file_open_rejects_relative_path() -> None:
+    with pytest.raises(ValueError):
+        sensitive_file_open("etc/shadow")
+
+
+def test_sensitive_file_open_rejects_too_long_path() -> None:
+    with pytest.raises(ValueError):
+        sensitive_file_open("/" + "a" * 255)