enodia-sentinal/enodia_sentinel/events/bcc_source.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""eBPF execve event source, built on bcc.

Loads a small eBPF program that fires on every ``execve``, capturing the
executed path, the first couple of arguments, the uid, and the parent. Events
are delivered to user space over a perf buffer and decoded into ``ExecEvent``.

This is intentionally fail-safe: if bcc isn't installed, the kernel lacks BPF,
or we aren't root, ``available()`` returns False and the daemon simply runs with
its poll-based detectors. A broken probe must never take the daemon down.
"""
from __future__ import annotations

import os
from collections.abc import Callable

from .exec_event import ExecEvent

# eBPF C. bcc rewrites pointer dereferences (e.g. task->real_parent->tgid) into
# bpf_probe_read calls, so this stays close to bcc's own execsnoop example.
_BPF_PROGRAM = r"""
#include <uapi/linux/ptrace.h>
#include <linux/sched.h>

#define ARGLEN 128

struct data_t {
    u32 pid;
    u32 ppid;
    u32 uid;
    char parent_comm[TASK_COMM_LEN];
    char filename[ARGLEN];
    char arg1[ARGLEN];
    char arg2[ARGLEN];
};
BPF_PERF_OUTPUT(events);

int syscall__execve(struct pt_regs *ctx,
                    const char __user *filename,
                    const char __user *const __user *__argv,
                    const char __user *const __user *__envp)
{
    struct data_t data = {};
    struct task_struct *task = (struct task_struct *)bpf_get_current_task();

    data.pid = bpf_get_current_pid_tgid() >> 32;
    data.ppid = task->real_parent->tgid;
    data.uid = bpf_get_current_uid_gid() & 0xffffffff;
    bpf_get_current_comm(&data.parent_comm, sizeof(data.parent_comm));
    bpf_probe_read_user_str(&data.filename, sizeof(data.filename), filename);

    const char __user *argp = NULL;
    bpf_probe_read_user(&argp, sizeof(argp), &__argv[1]);
    if (argp)
        bpf_probe_read_user_str(&data.arg1, sizeof(data.arg1), argp);
    argp = NULL;
    bpf_probe_read_user(&argp, sizeof(argp), &__argv[2]);
    if (argp)
        bpf_probe_read_user_str(&data.arg2, sizeof(data.arg2), argp);

    events.perf_submit(ctx, &data, sizeof(data));
    return 0;
}
"""


def attach_kprobe_candidates(bpf, syscall: str, fn_name: str,
                             extra: tuple[str, ...] = ()) -> str:
    """Attach a kprobe to the first syscall symbol that exists on this kernel."""
    candidates = []
    try:
        candidates.append(bpf.get_syscall_fnname(syscall))
    except Exception:
        pass
    candidates.extend(extra)
    candidates.extend([
        f"__x64_sys_{syscall}",
        f"__x64_sys_{syscall}at",
        f"__arm64_sys_{syscall}",
        f"__arm64_sys_{syscall}at",
        f"__ia32_sys_{syscall}",
        f"__ia32_sys_{syscall}at",
        f"__sys_{syscall}",
        f"sys_{syscall}",
    ])
    seen: set[str] = set()
    last_exc: Exception | None = None
    for fnname in candidates:
        if not fnname or fnname in seen:
            continue
        seen.add(fnname)
        try:
            bpf.attach_kprobe(event=fnname, fn_name=fn_name)
            return fnname
        except Exception as exc:
            last_exc = exc
            continue
    raise RuntimeError(
        f"could not attach {syscall} kprobe (tried {', '.join(sorted(seen))})"
    ) from last_exc


def available() -> tuple[bool, str]:
    """Return (ok, reason). ok=True means the probe can be loaded."""
    if os.geteuid() != 0:
        return False, "not running as root (BPF requires CAP_BPF/root)"
    try:
        import bcc  # noqa: F401
    except ImportError:
        return False, "python bcc (python-bpfcc) not installed"
    if not os.path.exists("/sys/kernel/btf/vmlinux") and not os.path.isdir(
        "/lib/modules/%s/build" % os.uname().release
    ):
        return False, "no kernel BTF or headers for BPF compilation"
    return True, "ok"


class BccExecSource:
    """Loads the eBPF probe and pumps decoded ExecEvents to a callback."""

    def __init__(self, on_event: Callable[[ExecEvent], None]) -> None:
        self._on_event = on_event
        self._bpf = None
        self._running = False

    def start(self) -> None:
        from bcc import BPF

        self._bpf = BPF(text=_BPF_PROGRAM)
        fnname = self._attach_execve_probe()
        self._bpf["events"].open_perf_buffer(self._handle, page_cnt=64)
        self._running = True

    def _attach_execve_probe(self) -> str:
        assert self._bpf is not None
        return attach_kprobe_candidates(self._bpf, "execve", "syscall__execve")

    def poll(self, timeout_ms: int = 200) -> None:
        if self._bpf is not None:
            self._bpf.perf_buffer_poll(timeout=timeout_ms)

    def stop(self) -> None:
        self._running = False
        if self._bpf is not None:
            try:
                self._bpf.cleanup()
            except Exception:
                pass
            self._bpf = None

    @property
    def running(self) -> bool:
        return self._running

    def _handle(self, cpu, data, size) -> None:
        event = self._bpf["events"].event(data)
        argv = tuple(
            a for a in (_decode(event.arg1), _decode(event.arg2)) if a
        )
        ev = ExecEvent(
            pid=event.pid,
            ppid=event.ppid,
            uid=event.uid,
            parent_comm=_decode(event.parent_comm),
            filename=_decode(event.filename),
            argv=argv,
        )
        try:
            self._on_event(ev)
        except Exception:
            # one bad event must not kill the poll loop
            pass


def _decode(raw) -> str:
    if isinstance(raw, bytes):
        return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
    return str(raw)