# SPDX-License-Identifier: GPL-3.0-or-later """Runs the eBPF exec source on a thread and routes events through the engine.""" from __future__ import annotations import threading from collections.abc import Callable from ..alert import Alert from ..config import Config from . import bcc_source from .exec_event import ExecEvent from .rules import ExecRuleEngine class ExecMonitor: """Owns the event source + rule engine; emits Alerts via ``on_alert``. ``start()`` is best-effort: it returns a (started, reason) pair and never raises, so the daemon can log the reason and carry on with polling. """ def __init__(self, cfg: Config, on_alert: Callable[[Alert], None]) -> None: self.cfg = cfg self._on_alert = on_alert self.engine = ExecRuleEngine.load(cfg.exec_rules_file) self._source = bcc_source.BccExecSource(self._on_event) self._thread: threading.Thread | None = None self._stop = threading.Event() def start(self) -> tuple[bool, str]: ok, reason = bcc_source.available() if not ok: return False, reason try: self._source.start() except Exception as exc: # compilation/attach/permission failure return False, f"probe load failed: {exc!r}" self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() return True, "ok" def _run(self) -> None: while not self._stop.is_set(): try: self._source.poll(timeout_ms=200) except Exception: break def _on_event(self, ev: ExecEvent) -> None: for alert in self.engine.match(ev): self._on_alert(alert) def stop(self) -> None: self._stop.set() self._source.stop()