diff --git a/benchmarks/sensor_overhead.py b/benchmarks/sensor_overhead.py new file mode 100644 index 0000000..36b381b --- /dev/null +++ b/benchmarks/sensor_overhead.py @@ -0,0 +1,87 @@ +"""Overhead of the Python runtime probe (#55): requests/sec with and without it. + + python benchmarks/sensor_overhead.py # all modes, ~10 s + python benchmarks/sensor_overhead.py --requests 200000 + +Each mode runs in a fresh interpreter. A "request" is a small handler that +parses a query string, builds a response dict through a few app functions, +and serializes it -- the shape of a typical web handler, without network +noise. "watched" calls the watched vulnerable function on every request +(the worst case); "unwatched" never calls it. +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import tempfile +import textwrap +from pathlib import Path + +APP = textwrap.dedent("""\ + import json, sys, time, urllib.parse + import vulnlib + + def parse(qs): + return dict(urllib.parse.parse_qsl(qs)) + + def build(params, call_vuln): + body = {"user": params.get("user"), "items": [i * 2 for i in range(10)]} + if call_vuln: + body["parsed"] = vulnlib.load(params.get("doc", "")) + return body + + def handle(qs, call_vuln): + return json.dumps(build(parse(qs), call_vuln)) + + n, call_vuln = int(sys.argv[1]), sys.argv[2] == "watched" + for _ in range(2000): + handle("user=a&doc=x", call_vuln) + start = time.perf_counter() + for _ in range(n): + handle("user=alice&doc=hello", call_vuln) + print(n / (time.perf_counter() - start)) + """) +VULNLIB = "def load(text):\n return text[::-1]\n" + + +def run(workdir: Path, requests: int, traffic: str, env_extra: dict[str, str] | None) -> float: + env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1] / "src")} + for key in ("SECFOO_SENSOR", "SECFOO_SENSOR_ADDR", "SECFOO_EVENTS_OUT"): + env.pop(key, None) + env.update(env_extra or {}) + cmd = [sys.executable, "-m", "secfoo.sensor", "app.py", str(requests), traffic] + out = subprocess.run(cmd, cwd=workdir, env=env, capture_output=True, text=True, check=True) # nosec B603 + return float(out.stdout.strip().splitlines()[-1]) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--requests", type=int, default=100_000) + args = parser.parse_args() + with tempfile.TemporaryDirectory() as tmp: + workdir = Path(tmp) + (workdir / "app.py").write_text(APP, encoding="utf-8") + (workdir / "vulnlib.py").write_text(VULNLIB, encoding="utf-8") + (workdir / "watchlist.json").write_text(json.dumps({"version": 1, "generated_at": "x", "entries": [{ + "finding_fingerprint": "fp", "advisory": None, "ecosystem": "PyPI", "package": "vulnlib", + "version": "1", "granularity": "symbol", "symbols": ["vulnlib.load"], "source": "osv"}]}), encoding="utf-8") + probe_env = {"SECFOO_SENSOR": "1", "SECFOO_WATCHLIST": "watchlist.json", + "SECFOO_EVENTS_OUT": str(workdir / "events.ndjson"), "SECFOO_FLUSH_INTERVAL": "5"} + backends = ["profile"] + (["monitoring"] if hasattr(sys, "monitoring") else []) + print(f"Python {sys.version.split()[0]}, {args.requests:,} requests per mode\n") + print(f"{'mode':<34}{'req/s':>12}{'overhead':>10}") + for traffic in ("unwatched", "watched"): + base = run(workdir, args.requests, traffic, None) + print(f"{'no probe, ' + traffic:<34}{base:>12,.0f}{'':>10}") + for backend in backends: + rate = run(workdir, args.requests, traffic, {**probe_env, "SECFOO_SENSOR_BACKEND": backend}) + label = f"probe ({backend}), {traffic}" + print(f"{label:<34}{rate:>12,.0f}{(base / rate - 1) * 100:>9.1f}%") + + +if __name__ == "__main__": + main() diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index c4f68e4..c028715 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -473,6 +473,73 @@ runtime-confirmed findings, but only once there is runtime data. ![SCA vulnerabilities list with the Runtime column and a highlighted conflict](assets/sca-runtime.jpg) +### Python probe + +The probe runs inside a Python app and records which watched vulnerable +functions actually run. It writes the event format above. + +```bash +secfoo runtime watchlist --project payments -o watchlist.json +secfoo instrument --watchlist watchlist.json -o events.ndjson -- python app.py # or gunicorn ..., uvicorn ... +secfoo runtime ingest events.ndjson --project payments +``` + +`secfoo instrument` starts the probe in every Python process the command +launches, workers included, without changing the app's start command. It +does this through a temporary `sitecustomize` on `PYTHONPATH`. When you can't +wrap the command, set the variables yourself: + +```bash +SECFOO_SENSOR=1 SECFOO_WATCHLIST=watchlist.json SECFOO_EVENTS_OUT=events-{pid}.ndjson python -m secfoo.sensor app.py +``` + +| Variable | Meaning | +|---|---| +| `SECFOO_SENSOR=1` | Required. Without it the probe installs nothing: no hooks, threads or connections. | +| `SECFOO_WATCHLIST` | The watchlist file. It is re-read when its mtime changes (checked at most every 30 s). | +| `SECFOO_EVENTS_OUT` | The events file. `{pid}` gives each process its own file; otherwise processes append to one file safely. | +| `SECFOO_SENSOR_ADDR` | Send to a sensor instead: `unix://…`, `tcp://127.0.0.1:port` (loopback only), `npipe://…`, `https://…` (direct mode, using the runtime ingest API with `SECFOO_API_KEY_FILE` / `SECFOO_PROJECT`), or `file://…`. | +| `SECFOO_FLUSH_INTERVAL` | Seconds between flushes (default 10). It also flushes at exit. | +| `SECFOO_APP_ROOT` | The app's own code, for coverage and stack frames (default: the working directory). | + +**How it works.** + +- **Backends.** On Python 3.12+ it uses `sys.monitoring` (PEP 669), with + `PY_START` events only on the watched functions' code objects, so unwatched + code costs nothing. On 3.10/3.11 it falls back to `sys.setprofile` and + `threading.setprofile`. +- **Symbols** are resolved as their modules are imported. +- **Coverage.** Each app function is counted the first time it runs, for + `app_functions_executed`. On 3.12+ that event is then disabled for the + function. +- **Inventory.** It emits `load` lines for installed distributions whose + top-level module was imported. +- **Workers.** Each process, including forked or spawned workers (gunicorn, + uvicorn, multiprocessing), is its own session. With gunicorn `--preload`, + the probe starts in the master and each forked worker starts a new session + automatically. + +**Guarantees.** + +- It never reads `frame.f_locals`, arguments or return values. It records + only function identity, counts, timestamps and `file:line` app frames. +- An internal error disables the probe, with one stderr line, and never + reaches the app. +- Transports run on a background thread and drop data rather than block + when the destination is gone. +- It opens no connection other than `SECFOO_SENSOR_ADDR`. + +**Overhead** (`python benchmarks/sensor_overhead.py`, CPython 3.14, a ~7 µs +request-like handler): + +| Mode | Unwatched code | Watched function called on every request | +|---|---|---| +| `sys.monitoring` (3.12+) | ≈ 0% | ~20% (≈ 1.4 µs per watched call) | +| `setprofile` fallback (3.10/3.11) | ~10× slower | ~10× slower | + +The fallback fires on every Python call, so prefer 3.12+ in production or +keep the fallback to staging. + ### Runtime ingest API Live sensors send evidence to the same server your CLI results go to. They diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index e3b39a9..0d2cb88 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import shutil import sys from datetime import date, datetime, timezone @@ -1815,6 +1816,67 @@ def skills_show( sys.stdout.write(render_prompt(skill, context, depth=depth.value) + "\n") +_SITECUSTOMIZE = """\ +# Written by `secfoo instrument` (#55): starts the secfoo runtime probe in +# every Python process of the wrapped command. Inert unless SECFOO_SENSOR=1. +try: + import secfoo.sensor as _secfoo_sensor + _secfoo_sensor.start_from_env() +except Exception: # never break the app's interpreter start-up + pass +""" + + +@app.command( + name="instrument", + context_settings={"allow_extra_args": True, "ignore_unknown_options": True}, +) +def instrument( + ctx: typer.Context, + watchlist: Annotated[Path | None, typer.Option( + "--watchlist", exists=True, dir_okay=False, help="Watchlist from `secfoo runtime watchlist`.", + )] = None, + output: Annotated[Path | None, typer.Option( + "--output", "-o", + help="Write events here ({pid} is replaced per process). Ignored when SECFOO_SENSOR_ADDR is set.", + )] = None, + flush_interval: Annotated[float, typer.Option("--flush-interval", min=0.05)] = 10.0, +) -> None: + """Run a Python command with the secfoo runtime probe: secfoo instrument -o events.ndjson -- python app.py + + The probe starts in every Python process the command launches (workers + included), records which watched vulnerable functions run, and writes + #53 events. Then: `secfoo runtime ingest events.ndjson --project

`. + """ + import subprocess # nosec B404 -- runs the user's own command, as given + import tempfile + + import secfoo + + command = list(ctx.args) + if command[:1] == ["--"]: + command = command[1:] + if not command: + err_console.print("[red]Error:[/] give the command to run after --, e.g. -- python app.py") + raise typer.Exit(code=2) + env = dict(os.environ) + if output is None and not env.get("SECFOO_SENSOR_ADDR") and not env.get("SECFOO_EVENTS_OUT"): + err_console.print("[red]Error:[/] pass -o events.ndjson (or set SECFOO_SENSOR_ADDR) so events go somewhere.") + raise typer.Exit(code=2) + with tempfile.TemporaryDirectory(prefix="secfoo-instrument-") as hook_dir: + Path(hook_dir, "sitecustomize.py").write_text(_SITECUSTOMIZE, encoding="utf-8") + package_root = str(Path(secfoo.__file__).resolve().parent.parent) + env["PYTHONPATH"] = os.pathsep.join(p for p in (hook_dir, package_root, env.get("PYTHONPATH", "")) if p) + env["SECFOO_SENSOR"] = "1" + env["SECFOO_FLUSH_INTERVAL"] = str(flush_interval) + if watchlist is not None: + env["SECFOO_WATCHLIST"] = str(watchlist.resolve()) + if output is not None: + env["SECFOO_EVENTS_OUT"] = str(output.resolve()) + result = subprocess.run(command, env=env, check=False) # nosec B603 -- the user's command + raise typer.Exit(code=result.returncode) + + MERMAID_CDN_URL = "https://cdn.jsdelivr.net/npm/mermaid@11/dist/mermaid.min.js" diff --git a/src/secfoo/sensor/__init__.py b/src/secfoo/sensor/__init__.py new file mode 100644 index 0000000..8d17d00 --- /dev/null +++ b/src/secfoo/sensor/__init__.py @@ -0,0 +1,49 @@ +"""secfoo runtime sensor: the in-process Python probe (#55). + +Opt-in only (#51 rule 0). `start_from_env()` does nothing unless +`SECFOO_SENSOR=1` is set, so importing this package -- or having it on +PYTHONPATH -- never changes an app's behavior. + + SECFOO_SENSOR=1 SECFOO_WATCHLIST=watchlist.json SECFOO_EVENTS_OUT=events.ndjson python -m secfoo.sensor app.py + secfoo instrument --watchlist watchlist.json -o events.ndjson -- python app.py +""" + +from __future__ import annotations + +import os +import sys + +_probe = None + + +def enabled(environ=None) -> bool: + return (os.environ if environ is None else environ).get("SECFOO_SENSOR", "").strip() == "1" + + +def start_from_env(): + """Start the probe from SECFOO_* variables. Returns it, or None when the + sensor isn't enabled. Never raises into the app.""" + global _probe + if _probe is not None or not enabled(): + return _probe + try: + from secfoo.sensor.probe import DEFAULT_FLUSH_SECONDS, Probe + from secfoo.sensor.transport import from_env + + flush = float(os.environ.get("SECFOO_FLUSH_INTERVAL", DEFAULT_FLUSH_SECONDS)) + _probe = Probe( + watchlist_path=os.environ.get("SECFOO_WATCHLIST") or None, + transport=from_env(), + app_root=os.environ.get("SECFOO_APP_ROOT") or None, + flush_seconds=max(flush, 0.05), + backend=os.environ.get("SECFOO_SENSOR_BACKEND") or None, + ) + _probe.start() + except Exception as exc: # noqa: BLE001 -- fail-open: a bad config never stops the app + print(f"secfoo sensor: not started ({exc})", file=sys.stderr) + _probe = None + return _probe + + +def current_probe(): + return _probe diff --git a/src/secfoo/sensor/__main__.py b/src/secfoo/sensor/__main__.py new file mode 100644 index 0000000..769edaa --- /dev/null +++ b/src/secfoo/sensor/__main__.py @@ -0,0 +1,29 @@ +"""`python -m secfoo.sensor app.py [args]` or `python -m secfoo.sensor -m module [args]`. + +Starts the probe (only if SECFOO_SENSOR=1), then runs the app exactly as +`python app.py` would. +""" + +from __future__ import annotations + +import runpy +import sys + +from secfoo.sensor import start_from_env + + +def main(argv: list[str]) -> None: + if not argv: + print("usage: python -m secfoo.sensor app.py [args] | -m module [args]", file=sys.stderr) + raise SystemExit(2) + start_from_env() + if argv[0] == "-m" and len(argv) > 1: + sys.argv = [argv[1], *argv[2:]] + runpy.run_module(argv[1], run_name="__main__", alter_sys=True) + else: + sys.argv = list(argv) + runpy.run_path(argv[0], run_name="__main__") + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/src/secfoo/sensor/probe.py b/src/secfoo/sensor/probe.py new file mode 100644 index 0000000..587d8c7 --- /dev/null +++ b/src/secfoo/sensor/probe.py @@ -0,0 +1,436 @@ +"""In-process Python probe (#55): records which watched vulnerable functions +run, which packages are loaded, and how much app code ran, as #53 events. + +Design rules (#51), all enforced here: + +- **Inert unless `SECFOO_SENSOR=1`.** `start()` returns without installing a + hook, thread or connection otherwise. +- **Observe-only, fail-open.** Callbacks only count. Any internal error turns + the probe off (one stderr line) and the app carries on. +- **No argument values.** Only code identity, counts, timestamps and + `file:line` frames are recorded. `frame.f_locals`, arguments and return + values are never read. +- **Never blocks the app.** Aggregation is in memory (bounded by the size of + the watchlist and the app); a daemon thread flushes running totals, so a + slow or missing transport only ever drops data. + +Backends: `sys.monitoring` (PEP 669, Python 3.12+) with PY_START on the +watched code objects only, or `sys.setprofile` filtered to them on 3.10/3.11 +(`SECFOO_SENSOR_BACKEND=profile` forces it, for tests and comparisons). +""" + +from __future__ import annotations + +import atexit +import importlib.abc +import json +import os +import sys +import sysconfig +import threading +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path + +PROBE_VERSION = "0.1.0" +DEFAULT_FLUSH_SECONDS = 10.0 +WATCHLIST_RECHECK_SECONDS = 30.0 +MAX_STACK_FRAMES = 10 +_TOOL_NAME = "secfoo-sensor" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _iso(epoch: float) -> str: + return datetime.fromtimestamp(epoch, timezone.utc).isoformat() + + +def _stdlib_and_site_dirs() -> tuple[str, ...]: + paths = sysconfig.get_paths() + dirs = {paths.get(k) for k in ("stdlib", "platstdlib", "purelib", "platlib")} + dirs |= {p for p in sys.path if p and ("site-packages" in p or "dist-packages" in p)} + return tuple(sorted(os.path.normcase(os.path.abspath(d)) for d in dirs if d)) + + +class Probe: + def __init__(self, *, watchlist_path: str | None, transport, app_root: str | None = None, + flush_seconds: float = DEFAULT_FLUSH_SECONDS, backend: str | None = None) -> None: + self.watchlist_path = watchlist_path + self.transport = transport + self.flush_seconds = flush_seconds + self.app_root = os.path.normcase(os.path.abspath(app_root or os.getcwd())) + self.backend = backend or ("monitoring" if hasattr(sys, "monitoring") else "profile") + self._excluded_dirs = _stdlib_and_site_dirs() + self._lock = threading.Lock() + self._active = False + self._broken = False + self._reset_session() + self._watch_entries: list[dict] = [] + self._watchlist_mtime: float | None = None + self._watchlist_checked = 0.0 + self._watched: dict = {} # code object -> (symbol, [fingerprints]) + self._pending_symbols: dict = {} # symbol -> [fingerprints], not resolved yet + self._seen_app_code: set = set() + self._tool_id: int | None = None + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + # ---------- lifecycle ---------- + + def _reset_session(self) -> None: + self.session_id = str(uuid.uuid4()) + self.started_at = _now() + self._hits: dict[tuple, dict] = {} # (fingerprint, symbol) -> aggregate + self._seen_app_code = set() + self.dropped = 0 + + def start(self) -> None: + try: + self._load_watchlist(force=True) + if self.backend == "monitoring": + self._start_monitoring() + else: + self._start_profile() + sys.meta_path.insert(0, _ImportWatcher(self)) + self._active = True + self._resolve_pending() + self._start_flusher() + atexit.register(self.stop) + # multiprocessing workers leave through os._exit(), which skips + # atexit; its own exit hook still runs, so flush there too. + import multiprocessing.util + + multiprocessing.util.Finalize(None, self.stop, exitpriority=100) + if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=self._after_fork) + except Exception as exc: # noqa: BLE001 -- fail-open + self._disable(exc) + + def stop(self) -> None: + if not self._active: + return + self._stop.set() + self.flush() + self._active = False + try: + if self._tool_id is not None: + sys.monitoring.register_callback(self._tool_id, sys.monitoring.events.PY_START, None) + sys.monitoring.set_events(self._tool_id, 0) + sys.monitoring.free_tool_id(self._tool_id) + self._tool_id = None + elif self.backend == "profile": + sys.setprofile(None) + threading.setprofile(None) # type: ignore[arg-type] + self.transport.close() + except Exception: # noqa: BLE001, S110 -- shutting down; never raise into the app + pass + + def _after_fork(self) -> None: + """A forked worker (gunicorn, multiprocessing) is its own session.""" + try: + self._lock = threading.Lock() + self._reset_session() + self._stop = threading.Event() + self.transport = self.transport.reopen() + self._start_flusher() + except Exception as exc: # noqa: BLE001 -- fail-open + self._disable(exc) + + def _disable(self, exc: BaseException) -> None: + if self._broken: + return + self._broken = True + self._active = False + try: + if self._tool_id is not None: + sys.monitoring.set_events(self._tool_id, 0) + for code in list(self._watched): + sys.monitoring.set_local_events(self._tool_id, code, 0) + elif self.backend == "profile": + sys.setprofile(None) + print(f"secfoo sensor: disabled after an internal error ({exc.__class__.__name__}); " + "the app is unaffected", file=sys.stderr) + except Exception: # noqa: BLE001, S110 -- never raise into the app + pass + + # ---------- watchlist ---------- + + def _load_watchlist(self, *, force: bool = False) -> None: + if not self.watchlist_path: + return + now = time.monotonic() + if not force and now - self._watchlist_checked < WATCHLIST_RECHECK_SECONDS: + return + self._watchlist_checked = now + try: + mtime = os.path.getmtime(self.watchlist_path) + except OSError: + return + if mtime == self._watchlist_mtime: + return + self._watchlist_mtime = mtime + with open(self.watchlist_path, encoding="utf-8") as handle: + document = json.load(handle) + entries = [e for e in document.get("entries", []) if e.get("ecosystem") == "PyPI"] + self._watch_entries = entries + for entry in entries: + for symbol in entry.get("symbols") or []: + self._pending_symbols.setdefault(symbol, []).append(entry["finding_fingerprint"]) + + def _resolve_pending(self) -> None: + for symbol in list(self._pending_symbols): + code = _resolve(symbol) + if code is None: + continue + fingerprints = self._pending_symbols.pop(symbol) + known = self._watched.get(code) + self._watched[code] = (symbol, sorted(set(fingerprints + (known[1] if known else [])))) + if self._tool_id is not None: + sys.monitoring.set_local_events(self._tool_id, code, sys.monitoring.events.PY_START) + + # ---------- backends ---------- + + def _start_monitoring(self) -> None: + mon = sys.monitoring + for tool_id in range(mon.PROFILER_ID, 6): # first free tool id; 0-1 are debugger/coverage + if mon.get_tool(tool_id) is None: + mon.use_tool_id(tool_id, _TOOL_NAME) + self._tool_id = tool_id + break + else: + raise RuntimeError("no free sys.monitoring tool id") + mon.register_callback(self._tool_id, mon.events.PY_START, self._on_py_start) + # Global PY_START for coverage of app code; each location is disabled + # after its first event, so steady-state cost is ~zero. + mon.set_events(self._tool_id, mon.events.PY_START) + + def _on_py_start(self, code, _offset): + try: + watched = self._watched.get(code) + if watched is not None: + self._record_hit(code, watched) + if code not in self._seen_app_code and self._is_app_code(code.co_filename): + self._seen_app_code.add(code) + return None # keep counting watched functions + if code.co_name != "" and self._is_app_code(code.co_filename): + self._seen_app_code.add(code) + return sys.monitoring.DISABLE + except Exception as exc: # noqa: BLE001 -- never raise into the app + self._disable(exc) + return sys.monitoring.DISABLE + + def _start_profile(self) -> None: + def profiler(frame, event, _arg): + if event != "call": + return + try: + code = frame.f_code + watched = self._watched.get(code) + if watched is not None: + self._record_hit(code, watched) + if (code not in self._seen_app_code and code.co_name != "" + and self._is_app_code(code.co_filename)): + self._seen_app_code.add(code) + except Exception as exc: # noqa: BLE001 + self._disable(exc) + + sys.setprofile(profiler) + threading.setprofile(profiler) + + # ---------- recording ---------- + + def _is_app_code(self, filename: str) -> bool: + if not filename or filename.startswith("<"): + return False + path = os.path.normcase(os.path.abspath(filename)) + return path.startswith(self.app_root) and not path.startswith(self._excluded_dirs) + + def _record_hit(self, code, watched) -> None: + """The hot path: one dict lookup and two stores per watched call. + Timestamps stay raw floats until flush. No lock: under the GIL a + lost increment between threads is possible but rare, and a running + total that's off by one is an acceptable price for not serialising + the app's threads on every call.""" + symbol, fingerprints = watched + now = time.time() + hits = self._hits + for fingerprint in fingerprints: + agg = hits.get((fingerprint, symbol)) + if agg is not None: + agg[0] += 1 + agg[2] = now + continue + stack = self._stack_sample() + # stack[0] is the watched function itself; the entry frame is the + # app code that called it. [count, first_seen, last_seen, stack, entry] + with self._lock: + hits.setdefault((fingerprint, symbol), [1, now, now, stack, stack[1] if len(stack) > 1 else None]) + + def _stack_sample(self) -> list[str]: + """file:line of app frames only (plus nothing else): no locals, no args.""" + frames: list[str] = [] + frame = sys._getframe(3) + while frame is not None and len(frames) < MAX_STACK_FRAMES: + filename = frame.f_code.co_filename + if self._is_app_code(filename): + rel = os.path.relpath(filename, self.app_root).replace(os.sep, "/") + frames.append(f"{rel}:{frame.f_lineno}") + frame = frame.f_back + return frames + + # ---------- output ---------- + + def snapshot(self) -> list[dict]: + """The session's running totals as #53 events.""" + with self._lock: + hits = {k: list(v) for k, v in self._hits.items()} + executed = len(self._seen_app_code) + session = { + "type": "session", "id": self.session_id, "language": "python", + "runtime": ".".join(map(str, sys.version_info[:3])), "source": "sensor", + "started_at": self.started_at, "ended_at": _now(), "sensor_version": PROBE_VERSION, + "instance": os.environ.get("HOSTNAME") or os.environ.get("COMPUTERNAME") or None, + } + if self.backend == "monitoring": + total = _count_app_functions(self) + session["app_functions_total"] = max(total, executed) + session["app_functions_executed"] = executed + session = {k: v for k, v in session.items() if v is not None} + events = [session, *self._loads()] + for (fingerprint, symbol), (count, first, last, stack, entry) in sorted(hits.items()): + hit = {"type": "hit", "session": self.session_id, "finding_fingerprint": fingerprint, + "symbol": symbol, "route": "unknown", "count": count, "first_seen": _iso(first), + "last_seen": _iso(last), "stack_sample": stack, "entry_frame": entry} + events.append({k: v for k, v in hit.items() if v not in (None, [])}) + return events + + def _loads(self) -> list[dict]: + """`load` lines for installed distributions with an imported top-level module.""" + try: + from importlib.metadata import packages_distributions, version + except ImportError: + return [] + imported = {name.split(".")[0] for name in list(sys.modules)} + seen: set[str] = set() + loads = [] + for top, dists in packages_distributions().items(): + if top not in imported: + continue + for dist in dists: + if dist.lower() in seen: + continue + seen.add(dist.lower()) + try: + loads.append({"type": "load", "session": self.session_id, "ecosystem": "PyPI", + "package": dist, "version": version(dist), "first_seen": self.started_at}) + except Exception: # noqa: BLE001, S112 -- broken package metadata: skip it + continue + return loads + + def flush(self) -> None: + if self._broken: + return + try: + self._load_watchlist() + self._resolve_pending() + if not self.transport.send(self.snapshot()): + self.dropped += 1 + except Exception as exc: # noqa: BLE001 + self._disable(exc) + + def _start_flusher(self) -> None: + def loop(): + while not self._stop.wait(self.flush_seconds): + self.flush() + + self._thread = threading.Thread(target=loop, name="secfoo-sensor-flush", daemon=True) + self._thread.start() + + +def _resolve(symbol: str): + """The code object for `module.qualname`, once the module is imported.""" + parts = symbol.split(".") + for split in range(len(parts) - 1, 0, -1): + module = sys.modules.get(".".join(parts[:split])) + if module is None: + continue + obj = module + try: + for attr in parts[split:]: + obj = getattr(obj, attr) + except AttributeError: + return None + obj = getattr(obj, "__func__", obj) + while hasattr(obj, "__wrapped__"): + obj = obj.__wrapped__ + return getattr(obj, "__code__", None) + return None + + +def _count_app_functions(probe: Probe) -> int: + """Function code objects defined in loaded app modules (the coverage + denominator). Walks co_consts, never module or frame state.""" + total = 0 + files = {getattr(module, "__file__", None) for module in list(sys.modules.values())} + # __main__ may already be gone from sys.modules at exit; the files of app + # code that ran, and the main script, still name it. + files |= {code.co_filename for code in list(probe._seen_app_code)} + if sys.argv and sys.argv[0]: + files.add(os.path.abspath(sys.argv[0])) + for filename in {os.path.normcase(os.path.abspath(f)) for f in files if f}: + if not os.path.isfile(filename) or not probe._is_app_code(filename): + continue + try: + source = Path(filename).read_text(encoding="utf-8") + stack = [compile(source, filename, "exec")] + except (OSError, SyntaxError, ValueError, UnicodeDecodeError): + continue + while stack: + code = stack.pop() + for const in code.co_consts: + if hasattr(const, "co_code"): + total += 1 + stack.append(const) + return total + + +class _ImportWatcher(importlib.abc.MetaPathFinder): + """Resolves watched symbols right after their module is imported. Never + finds modules itself (find_spec returns None), so imports are unchanged.""" + + def __init__(self, probe: Probe) -> None: + self.probe = probe + self._busy = threading.local() + + def find_spec(self, fullname, path=None, target=None): + if getattr(self._busy, "on", False) or not self.probe._pending_symbols: + return None + if not any(s == fullname or s.startswith(fullname + ".") for s in self.probe._pending_symbols): + return None + self._busy.on = True + try: + spec = importlib.util.find_spec(fullname) if path is None else None + except (ImportError, ValueError): + spec = None + finally: + self._busy.on = False + if spec is None or spec.loader is None or not hasattr(spec.loader, "exec_module"): + return None + original = spec.loader.exec_module + probe = self.probe + + def exec_module(module): + original(module) + try: + probe._resolve_pending() + except Exception as exc: # noqa: BLE001 + probe._disable(exc) + + spec.loader.exec_module = exec_module # type: ignore[method-assign] + return spec + + +import importlib.util diff --git a/src/secfoo/sensor/transport.py b/src/secfoo/sensor/transport.py new file mode 100644 index 0000000..6cba198 --- /dev/null +++ b/src/secfoo/sensor/transport.py @@ -0,0 +1,214 @@ +"""Where a probe's events go (#73 transports), chosen by one env var. + + SECFOO_SENSOR_ADDR=unix:///var/run/secfoo/sensor.sock host daemon / sidecar sharing a volume + SECFOO_SENSOR_ADDR=tcp://127.0.0.1:7461 sidecar sharing a network namespace (loopback only) + SECFOO_SENSOR_ADDR=npipe:////./pipe/secfoo-sensor Windows + SECFOO_SENSOR_ADDR=https://secfoo.internal direct mode (serverless): POST to the #66 API + SECFOO_SENSOR_ADDR=file:///path/events.ndjson offline; read later with `secfoo runtime ingest` + +With no SECFOO_SENSOR_ADDR, SECFOO_EVENTS_OUT (a file path) is used. Every +`send()` runs on the probe's flush thread, never on a request path, and +returns False instead of raising or retrying when the destination is gone. +The probe opens no connection other than the one named here. +""" + +from __future__ import annotations + +import gzip +import ipaddress +import json +import os +import socket +import sys +import urllib.parse +import urllib.request + +PROTOCOL_VERSION = 1 +DIRECT_TIMEOUT_SECONDS = 0.2 # #73: a cold start or shutdown never waits on the network +SOCKET_TIMEOUT_SECONDS = 1.0 + + +class TransportError(ValueError): + """The address is malformed or not allowed (e.g. a routable TCP host).""" + + +def _lines(events: list[dict]) -> bytes: + return "".join(json.dumps(e, separators=(",", ":")) + "\n" for e in events).encode("utf-8") + + +def hello(runtime: str = "python") -> dict: + return { + "type": "hello", "protocol": PROTOCOL_VERSION, "runtime": runtime, + "runtime_version": ".".join(map(str, sys.version_info[:3])), + "probe_version": "0.1.0", "pid": os.getpid(), + "host": os.environ.get("HOSTNAME") or socket.gethostname(), + } + + +class FileTransport: + def __init__(self, path: str) -> None: + self.path = path.replace("{pid}", str(os.getpid())) + self.template = path + + def send(self, events: list[dict]) -> bool: + try: + fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) + try: + os.write(fd, _lines(events)) # one write per flush: lines from forked workers never interleave + finally: + os.close(fd) + return True + except OSError: + return False + + def reopen(self) -> FileTransport: + return FileTransport(self.template) + + def close(self) -> None: + pass + + +class _StreamTransport: + """unix / tcp / npipe: newline-framed #53 lines after a `hello`.""" + + def __init__(self, address: str) -> None: + self.address = address + self._conn = None + self.slow = False + + def _connect(self): + raise NotImplementedError + + def _write(self, data: bytes) -> None: + raise NotImplementedError + + def send(self, events: list[dict]) -> bool: + try: + if self._conn is None: + self._connect() + self._write(_lines([hello()])) + self._write(_lines(events)) + return True + except OSError: + self.close() # sensor gone: drop this batch, reconnect next flush + return False + + def reopen(self): + return type(self)(self.address) + + def close(self) -> None: + conn, self._conn = self._conn, None + if conn is not None: + try: + conn.close() + except OSError: + pass + + +class SocketTransport(_StreamTransport): + def __init__(self, address: str) -> None: + super().__init__(address) + parsed = urllib.parse.urlsplit(address) + self.scheme = parsed.scheme + if parsed.scheme == "tcp": + host = parsed.hostname or "" + try: + loopback = host == "localhost" or ipaddress.ip_address(host).is_loopback + except ValueError: + loopback = False + if not loopback or not parsed.port: + raise TransportError("tcp:// sensor addresses must be loopback with a port, e.g. tcp://127.0.0.1:7461") + self.target = (host, parsed.port) + else: + self.target = parsed.path + + def _connect(self): + family = socket.AF_INET6 if self.scheme == "tcp" and ":" in self.target[0] else socket.AF_INET + if self.scheme == "unix": + family = socket.AF_UNIX + conn = socket.socket(family, socket.SOCK_STREAM) + conn.settimeout(SOCKET_TIMEOUT_SECONDS) + try: + conn.connect(self.target) + except OSError: + conn.close() + raise + self._conn = conn + key = os.environ.get("SECFOO_SENSOR_KEY") + if key and self.scheme == "tcp": + self._write(_lines([{"type": "auth", "key": key}])) + + def _write(self, data: bytes) -> None: + self._conn.sendall(data) + + +class PipeTransport(_StreamTransport): + def __init__(self, address: str) -> None: + super().__init__(address) + # npipe:////./pipe/secfoo-sensor -> \\.\pipe\secfoo-sensor + self.path = "\\\\" + address[len("npipe:////"):].replace("/", "\\") + + def _connect(self): + self._conn = open(self.path, "wb", buffering=0) # noqa: SIM115 -- kept open between flushes + + def _write(self, data: bytes) -> None: + self._conn.write(data) + + +class DirectTransport: + """Direct mode: POST gzip NDJSON to the #66 API with the sensor's key, + capped at DIRECT_TIMEOUT_SECONDS so the app never waits on the network.""" + + def __init__(self, address: str) -> None: + from secfoo.runtime.client import load_sensor_config + + self.config = load_sensor_config({**os.environ, "SECFOO_SERVER": address}) + + def send(self, events: list[dict]) -> bool: + query = urllib.parse.urlencode({k: v for k, v in { + "project": self.config.project, "app": self.config.app, "env": self.config.env}.items() if v}) + request = urllib.request.Request( # nosec B310 -- https enforced by load_sensor_config + self.config.url("events") + "?" + query, data=gzip.compress(_lines(events)), method="POST", + headers={"Authorization": f"Bearer {self.config.api_key}", "Content-Encoding": "gzip", + "Content-Type": "application/x-ndjson"}, + ) + try: + with urllib.request.urlopen(request, timeout=DIRECT_TIMEOUT_SECONDS) as response: # nosec B310 + return 200 <= response.status < 300 + except OSError: + return False + + def reopen(self): + return self + + def close(self) -> None: + pass + + +class NullTransport: + def send(self, events: list[dict]) -> bool: + return False + + def reopen(self): + return self + + def close(self) -> None: + pass + + +def from_env(environ: dict[str, str] | None = None): + environ = os.environ if environ is None else environ + address = environ.get("SECFOO_SENSOR_ADDR", "").strip() + if not address: + out = environ.get("SECFOO_EVENTS_OUT", "").strip() + return FileTransport(out) if out else NullTransport() + scheme = urllib.parse.urlsplit(address).scheme + if scheme == "file": + return FileTransport(urllib.request.url2pathname(urllib.parse.urlsplit(address).path)) + if scheme in ("unix", "tcp"): + return SocketTransport(address) + if scheme == "npipe": + return PipeTransport(address) + if scheme == "https": + return DirectTransport(address) + raise TransportError(f"unsupported SECFOO_SENSOR_ADDR {address!r}; use unix://, tcp://, npipe://, https:// or file://") diff --git a/tests/test_sensor/test_python_probe.py b/tests/test_sensor/test_python_probe.py new file mode 100644 index 0000000..17bd9c3 --- /dev/null +++ b/tests/test_sensor/test_python_probe.py @@ -0,0 +1,276 @@ +"""#55: the in-process Python probe.""" +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import textwrap +import threading +from pathlib import Path + +import pytest + +import secfoo +from secfoo.runtime.evidence import ingest_events +from secfoo.sensor import transport as transport_mod +from secfoo.sensor.probe import Probe +from secfoo.storage.repository import RunRepository + +SRC = str(Path(secfoo.__file__).resolve().parent.parent) +BACKENDS = ["profile"] + (["monitoring"] if hasattr(sys, "monitoring") else []) + +VULN = "def parse(text):\n return text.upper()\n\ndef safe(text):\n return text\n" +APP = textwrap.dedent("""\ + import sys + import sample_vuln + + def handle(secret): + return sample_vuln.parse(secret) if "--call" in sys.argv else sample_vuln.safe(secret) + + if __name__ == "__main__": + for _ in range(5): + handle("password=hunter2") + print("served") + """) +WORKERS_APP = textwrap.dedent("""\ + import multiprocessing as mp + import sample_vuln + + def work(n): + return sample_vuln.parse("x" * n) + + if __name__ == "__main__": + pool = mp.get_context("spawn").Pool(2) + print(sum(len(r) for r in pool.map(work, range(1, 9), chunksize=4))) + pool.close() # workers exit normally, as gunicorn/uvicorn workers do + pool.join() + """) + + +def _watchlist(path: Path, symbols=("sample_vuln.parse",)) -> Path: + path.write_text(json.dumps({"version": 1, "generated_at": "2026-11-12T00:00:00Z", "entries": [ + {"finding_fingerprint": "fp-1", "advisory": None, "ecosystem": "PyPI", "package": "sample-vuln", + "version": "1.0", "granularity": "symbol", "symbols": list(symbols), "source": "osv"}]}), encoding="utf-8") + return path + + +@pytest.fixture +def app_dir(tmp_path): + (tmp_path / "sample_vuln.py").write_text(VULN, encoding="utf-8") + (tmp_path / "app.py").write_text(APP, encoding="utf-8") + (tmp_path / "workers.py").write_text(WORKERS_APP, encoding="utf-8") + _watchlist(tmp_path / "watchlist.json") + return tmp_path + + +def _run_instrumented(app_dir: Path, *args: str, backend: str, out: str = "events.ndjson", extra_env=None): + env = {**os.environ, "PYTHONPATH": SRC, "SECFOO_SENSOR_BACKEND": backend, **(extra_env or {})} + env.pop("SECFOO_SENSOR_ADDR", None) + cmd = [sys.executable, "-m", "secfoo.cli", "instrument", "--watchlist", "watchlist.json", "-o", out, + "--flush-interval", "0.1", "--", sys.executable, *args] + return subprocess.run(cmd, cwd=app_dir, env=env, capture_output=True, text=True, timeout=120, check=False) + + +def _events(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def _latest_hits(events: list[dict]) -> dict: + hits = {} + for event in events: + if event["type"] == "hit": + hits[(event["session"], event["symbol"])] = event + return hits + + +# ---------- calling / not calling the watched function ---------- + + +@pytest.mark.parametrize("backend", BACKENDS) +def test_calling_the_watched_function_is_recorded(app_dir, backend): + result = _run_instrumented(app_dir, "app.py", "--call", backend=backend) + assert result.returncode == 0 and "served" in result.stdout, result.stderr + events = _events(app_dir / "events.ndjson") + [hit] = _latest_hits(events).values() + assert hit["finding_fingerprint"] == "fp-1" and hit["symbol"] == "sample_vuln.parse" and hit["count"] == 5 + assert hit["entry_frame"] == "app.py:5" # the app line that called it + assert hit["stack_sample"][0].startswith("sample_vuln.py:") + session = [e for e in events if e["type"] == "session"][-1] # the final flush's running totals + assert session["language"] == "python" and session["source"] == "sensor" + if backend == "monitoring": + # handle() and parse() ran; safe() didn't. + assert session["app_functions_executed"] == 2 and session["app_functions_total"] >= 3 + + +@pytest.mark.parametrize("backend", BACKENDS) +def test_not_calling_it_records_a_session_without_hits(app_dir, backend): + result = _run_instrumented(app_dir, "app.py", backend=backend) + assert result.returncode == 0, result.stderr + events = _events(app_dir / "events.ndjson") + assert any(e["type"] == "session" for e in events) + assert not any(e["type"] == "hit" for e in events) + + +@pytest.mark.parametrize("backend", BACKENDS) +def test_argument_values_never_reach_the_events(app_dir, backend): + _run_instrumented(app_dir, "app.py", "--call", backend=backend) + assert "hunter2" not in (app_dir / "events.ndjson").read_text(encoding="utf-8") + + +def test_output_ingests_cleanly_with_the_53_format(app_dir, tmp_path): + _run_instrumented(app_dir, "app.py", "--call", backend=BACKENDS[-1]) + with RunRepository(db_path=tmp_path / "db.sqlite") as repo: + pid = repo.upsert_project("/app", "app", "local") + run_id = repo.get_run(repo.create_run(project_id=pid, skill_id="sca-reachability", skill_name="SCA", + agent_id="claude", confluence_urls=[])).id + repo.upsert_sca_finding(project_id=pid, fingerprint="fp-1", run_id=run_id, current_ref="D1", + package="sample-vuln", severity="High") + lines = (app_dir / "events.ndjson").read_text(encoding="utf-8").splitlines() + report = ingest_events(repo.connection, pid, lines) + assert report.skipped == [] and report.hits >= 1 + assert repo.connection.execute("SELECT count FROM runtime_evidence").fetchone()[0] == 5 + + +# ---------- opt-in only ---------- + + +def test_inert_without_secfoo_sensor(app_dir): + env = {**os.environ, "PYTHONPATH": SRC, "SECFOO_WATCHLIST": "watchlist.json", "SECFOO_EVENTS_OUT": "events.ndjson"} + env.pop("SECFOO_SENSOR", None) + result = subprocess.run([sys.executable, "-m", "secfoo.sensor", "app.py", "--call"], cwd=app_dir, env=env, + capture_output=True, text=True, timeout=60, check=False) + assert result.returncode == 0 and "served" in result.stdout + assert not (app_dir / "events.ndjson").exists() + + +def test_importing_the_package_does_nothing(monkeypatch): + monkeypatch.delenv("SECFOO_SENSOR", raising=False) + from secfoo import sensor + + assert sensor.start_from_env() is None and sensor.current_probe() is None + + +# ---------- fail-open ---------- + + +def test_a_broken_watchlist_never_stops_the_app(app_dir): + (app_dir / "watchlist.json").write_text("{not json", encoding="utf-8") + result = _run_instrumented(app_dir, "app.py", "--call", backend=BACKENDS[-1]) + assert result.returncode == 0 and "served" in result.stdout + assert "secfoo sensor" in result.stderr + + +@pytest.mark.parametrize("backend", BACKENDS) +def test_an_error_inside_the_probe_disables_it_and_the_app_keeps_running(tmp_path, backend, monkeypatch, capsys): + module = tmp_path / "probe_target_mod.py" + module.write_text(VULN, encoding="utf-8") + monkeypatch.syspath_prepend(str(tmp_path)) + import probe_target_mod + + probe = Probe(watchlist_path=str(_watchlist(tmp_path / "w.json", ("probe_target_mod.parse",))), + transport=transport_mod.NullTransport(), app_root=str(tmp_path), flush_seconds=60, backend=backend) + probe.start() + try: + def boom(*_a, **_k): + raise RuntimeError("probe bug") + + monkeypatch.setattr(probe, "_record_hit", boom) + assert probe_target_mod.parse("abc") == "ABC" # app code still works + assert probe_target_mod.parse("def") == "DEF" + assert probe._broken + assert "disabled after an internal error" in capsys.readouterr().err + finally: + probe.stop() + if backend == "profile": + sys.setprofile(None) + + +def test_sensor_dying_mid_run_drops_events_without_blocking(tmp_path): + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen() + port = server.getsockname()[1] + received = [] + + def accept_once(): + conn, _ = server.accept() + data = b"" + while b"\n" not in data or data.count(b"\n") < 2: + chunk = conn.recv(65536) + if not chunk: + break + data += chunk + received.append(data) + conn.close() + server.close() # the sensor goes away + + thread = threading.Thread(target=accept_once, daemon=True) + thread.start() + tcp = transport_mod.SocketTransport(f"tcp://127.0.0.1:{port}") + assert tcp.send([{"type": "session", "id": "s"}]) + thread.join(5) + hello = json.loads(received[0].splitlines()[0]) + assert hello["type"] == "hello" and hello["protocol"] == 1 and hello["runtime"] == "python" + # Sensor gone: sends now fail fast and return False instead of raising or retrying. + results = [tcp.send([{"type": "session", "id": "s"}]) for _ in range(3)] + assert results[-1] is False + + +# ---------- several processes ---------- + + +def test_each_worker_process_is_its_own_session(app_dir): + result = _run_instrumented(app_dir, "workers.py", backend=BACKENDS[-1]) + assert result.returncode == 0 and result.stdout.strip() == "36", result.stderr + events = _events(app_dir / "events.ndjson") + hits = _latest_hits(events) + sessions = {e["id"] for e in events if e["type"] == "session"} + assert len(sessions) >= 2 # the parent and its spawned workers are separate sessions + assert {session for session, _ in hits} < sessions # hits come from workers, not the parent + assert sum(h["count"] for h in hits.values()) == 8 # every call counted exactly once + + +def test_pid_placeholder_gives_each_process_its_own_file(app_dir): + result = _run_instrumented(app_dir, "workers.py", backend=BACKENDS[-1], out="events-{pid}.ndjson") + assert result.returncode == 0, result.stderr + assert len(list(app_dir.glob("events-*.ndjson"))) >= 2 + + +# ---------- transports ---------- + + +@pytest.mark.parametrize("address", ["tcp://10.0.0.5:7461", "tcp://example.com:7461", "tcp://127.0.0.1"]) +def test_tcp_sensor_address_must_be_loopback_with_a_port(address): + with pytest.raises(transport_mod.TransportError): + transport_mod.SocketTransport(address) + + +def test_transport_selection(tmp_path): + assert isinstance(transport_mod.from_env({}), transport_mod.NullTransport) + assert isinstance(transport_mod.from_env({"SECFOO_EVENTS_OUT": str(tmp_path / "e")}), transport_mod.FileTransport) + assert isinstance(transport_mod.from_env({"SECFOO_SENSOR_ADDR": "tcp://127.0.0.1:7461"}), + transport_mod.SocketTransport) + assert isinstance(transport_mod.from_env({"SECFOO_SENSOR_ADDR": "npipe:////./pipe/secfoo-sensor"}), + transport_mod.PipeTransport) + with pytest.raises(transport_mod.TransportError): + transport_mod.from_env({"SECFOO_SENSOR_ADDR": "ftp://x"}) + + +def test_direct_mode_needs_explicit_key_and_project(monkeypatch): + monkeypatch.delenv("SECFOO_API_KEY", raising=False) + monkeypatch.delenv("SECFOO_API_KEY_FILE", raising=False) + from secfoo.runtime.client import SensorConfigError + + with pytest.raises(SensorConfigError): + transport_mod.from_env({"SECFOO_SENSOR_ADDR": "https://secfoo.internal"}) + + +def test_instrument_requires_a_destination_and_a_command(app_dir): + env = {**os.environ, "PYTHONPATH": SRC} + env.pop("SECFOO_SENSOR_ADDR", None) + env.pop("SECFOO_EVENTS_OUT", None) + no_out = subprocess.run([sys.executable, "-m", "secfoo.cli", "instrument", "--", sys.executable, "app.py"], + cwd=app_dir, env=env, capture_output=True, text=True, timeout=60, check=False) + assert no_out.returncode == 2 and "-o events.ndjson" in no_out.stderr