From 7eed2feef2893e5ba44f06bfb9586411c87819ec Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 04:07:23 +0000 Subject: [PATCH] feat(runtime): probe<->sensor protocol, schema and conformance kit (#73) docs/RUNTIME_PROTOCOL.md specifies activation, the SECFOO_SENSOR_ADDR transports, newline-framed hello/auth/welcome/slow plus #53 lines, privacy, fail-open and versioning; protocol.schema.json and protocol.py validate it. secfoo sensor conformance -- runs a probe's sample app against a fake loopback sensor and checks inert, call/nocall, privacy (sentinel in URL and header), fail-open (sensor killed mid-run) and marginal per-call overhead. CI runs it against the reference Python probe. Fixes the kit found in the probe: submodule symbol resolution, per-call app-code check, per-flush recomputation, auth after hello, and honouring welcome/slow. Closes #73 Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 3 + docs/RUNTIME_PROTOCOL.md | 144 ++++++++++++ src/secfoo/cli.py | 46 ++++ src/secfoo/runtime/conformance/__init__.py | 1 + src/secfoo/runtime/conformance/kit.py | 223 +++++++++++++++++++ src/secfoo/runtime/conformance/sample_app.py | 62 ++++++ src/secfoo/runtime/conformance/vulnerable.py | 10 + src/secfoo/runtime/protocol.py | 95 ++++++++ src/secfoo/runtime/protocol.schema.json | 60 +++++ src/secfoo/sensor/probe.py | 59 ++++- src/secfoo/sensor/transport.py | 47 +++- tests/test_runtime/test_protocol.py | 193 ++++++++++++++++ 12 files changed, 931 insertions(+), 12 deletions(-) create mode 100644 docs/RUNTIME_PROTOCOL.md create mode 100644 src/secfoo/runtime/conformance/__init__.py create mode 100644 src/secfoo/runtime/conformance/kit.py create mode 100644 src/secfoo/runtime/conformance/sample_app.py create mode 100644 src/secfoo/runtime/conformance/vulnerable.py create mode 100644 src/secfoo/runtime/protocol.py create mode 100644 src/secfoo/runtime/protocol.schema.json create mode 100644 tests/test_runtime/test_protocol.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1f1f8dc..4443490 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -49,6 +49,9 @@ jobs: - name: Test (secfoo core) run: pytest tests + - name: Runtime probe conformance (reference Python probe, #73) + run: secfoo sensor conformance -- python -m secfoo.sensor -m secfoo.runtime.conformance.sample_app + - name: Validate example custom skills # Same check a team's own skills repo would run (docs/SKILLS.md). run: secfoo skills validate examples/skills diff --git a/docs/RUNTIME_PROTOCOL.md b/docs/RUNTIME_PROTOCOL.md new file mode 100644 index 0000000..bcde296 --- /dev/null +++ b/docs/RUNTIME_PROTOCOL.md @@ -0,0 +1,144 @@ +# Runtime probe ↔ sensor protocol (v1) + +Every runtime gets its own in-process **probe**: Python, Node.js, JVM, .NET, +Go, Ruby, PHP and so on. All probes talk to the same **sensor** (`secfoo +sensor`, #67) in the same way. This document is that contract. A probe for a +new runtime can be built from this page and checked with the conformance kit, +without reading another probe's code. + +- **Machine-readable schemas:** + [`src/secfoo/runtime/protocol.schema.json`](../src/secfoo/runtime/protocol.schema.json) + for control messages, and + [`src/secfoo/runtime/evidence.schema.json`](../src/secfoo/runtime/evidence.schema.json) + for event lines. +- **Reference probe:** `secfoo.sensor` (Python). +- **Design rules:** see #51. They apply to every probe. + +## 1. Activation + +A probe is **inert unless `SECFOO_SENSOR=1` is set**: no hooks, no threads, +no connections, no files. Installing secfoo or a probe package never activates +anything. The conformance kit checks this. + +## 2. Configuration + +| Variable | Meaning | +|---|---| +| `SECFOO_SENSOR=1` | Required to do anything at all. | +| `SECFOO_SENSOR_ADDR` | Where to send events (§3). | +| `SECFOO_EVENTS_OUT` | A file to write when `SECFOO_SENSOR_ADDR` is unset. `{pid}` expands to the process id. | +| `SECFOO_WATCHLIST` | The watchlist file. Re-read when its mtime changes, checked at most every 30 s. | +| `SECFOO_FLUSH_INTERVAL` | Seconds between flushes, until a sensor's `welcome` says otherwise (default 10). | +| `SECFOO_SENSOR_KEY` | An optional shared key for `tcp://` (§4). | +| `SECFOO_API_KEY_FILE` / `SECFOO_API_KEY`, `SECFOO_PROJECT`, `SECFOO_APP`, `SECFOO_ENV` | Direct mode only (§3). | + +## 3. Transport + +One variable, `SECFOO_SENSOR_ADDR`, picks the transport: + +| Value | Use it for | +|---|---| +| `unix:///var/run/secfoo/sensor.sock` (default on Linux/macOS) | A host daemon, or a sidecar sharing a volume. | +| `tcp://127.0.0.1:7461` | Sidecars that share a network namespace but not a filesystem: a K8s pod, an ECS/Fargate task, Cloud Run, Azure Container Apps, a Nomad group. **Loopback only.** A probe must refuse a routable host. | +| `npipe:////./pipe/secfoo-sensor` (default on Windows) | Windows hosts and Windows containers. | +| `https://secfoo.internal` | **Direct mode**, when no second process can run (serverless, single-process PaaS). The probe POSTs gzip NDJSON batches to `POST /api/runtime/v1/events?project=&app=&env=` with `Authorization: Bearer `, using the same API key as the CLI. | +| `file:///path/events.ndjson` | Offline. Ingest the file later with `secfoo runtime ingest`. | + +The sensor listens only on loopback or local IPC, never on a routable address. +A probe opens **no connection other than the one `SECFOO_SENSOR_ADDR` names**. + +## 4. Messages + +A connection is a stream of **newline-framed JSON objects**, one per line, +UTF-8, with no line longer than 64 KiB. + +``` +probe → sensor {"type":"hello", ...} +probe → sensor {"type":"auth","key":"..."} (tcp:// with SECFOO_SENSOR_KEY only) +sensor → probe {"type":"welcome", ...} +probe → sensor #53 event lines: session, load, hit (repeated on every flush) +sensor → probe {"type":"slow","sample_every":4} (any time, optional) +``` + +| Message | Direction | Fields | +|---|---|---| +| `hello` | probe → sensor, the first line on every connection | `protocol` (1), `runtime` (`python` `node` `jvm` `dotnet` `go` `rust` `ruby` `php` `native`), `runtime_version`, `probe_version`, `pid`, `host` (host or pod name) | +| `auth` | probe → sensor, straight after `hello`, `tcp://` only | `key`: the value of `SECFOO_SENSOR_KEY`. Other containers in a pod share loopback, so the sensor may require it. | +| `welcome` | sensor → probe | `protocol` (the accepted version), `flush_interval` (seconds; probes adopt it); optional `watchlist_version` and `watchlist_path` | +| `slow` | sensor → probe | `sample_every` (≥ 1): flush less often and sample more coarsely until reconnecting | +| event lines | probe → sensor | Exactly the `session` / `load` / `hit` lines of the runtime evidence format (`docs/REFERENCE.md` → *Runtime evidence*) | + +**Running totals.** Each flush resends the session line and the session's +running totals. A hit's `count` is the total so far for (fingerprint, symbol, +route). The sensor and server **replace** values, never add them, so a +resent batch is harmless. + +**Privacy (#51 rule 3).** Lines carry function identity, counts, timestamps, +`file:line` frames of app code (plus the vulnerable function itself), a +route **template** and an anonymous/authenticated flag. They never carry +argument or return values, local variables, request bodies, concrete URLs, +query strings, header values or user identities. + +## 5. Failure behaviour + +- **Fail-open.** An error inside the probe disables the probe and never + reaches the app. The app must never crash, block or slow down noticeably. + Target: at most 1 ms p99 added per request. +- **No sensor.** If the sensor is gone, the probe drops that flush's events, + counts the drops, and reconnects on the next flush. It never blocks or + retries on a request path. All I/O happens on a background flush thread or + timer. +- **Bounded.** Memory is bounded by the watchlist and the size of the app, not + by traffic: hits are aggregated in place, and an outgoing queue that fills + up drops. +- **Direct mode** flushes at the end of each invocation (serverless) or on the + interval, with a hard cap of 200 ms per flush, so a cold start or shutdown + never waits on the network. +- **Workers.** Each process (forked or spawned worker) is its own session, + with its own `session.id`. + +## 6. Conformance kit + +```bash +secfoo sensor conformance -- +# the reference Python probe: +secfoo sensor conformance -- python -m secfoo.sensor -m secfoo.runtime.conformance.sample_app +``` + +The kit writes a watchlist containing `--symbol` (default +`secfoo.runtime.conformance.vulnerable.parse`; pass your runtime's equivalent +in its #52 notation, with `--ecosystem`). It then starts a fake sensor on +`tcp://127.0.0.1:` and runs the command once per step. The step is +passed in `SECFOO_CONFORMANCE_STEP`. + +**Sample app contract.** Your sample app reads the step, does exactly this, +prints anything, and exits 0: + +| Step | The app must… | +|---|---| +| `call` | call the watched function 3 times | +| `nocall` | call only an unwatched function | +| `request` | handle one request to the concrete URL `/users/8675309?token=` (route template `GET /users/{id}`) with header `Authorization: Bearer `, passing both to the watched function, where `` is `SECFOO_CONFORMANCE_SENTINEL` | +| `slow` | keep calling the watched function for ~1.5 s (the kit kills its sensor meanwhile) | +| `loop` | call the watched function `SECFOO_CONFORMANCE_N` times | + +**Checks.** Every line in every step must validate against the schemas. + +| Check | Passes when | +|---|---| +| `inert` | With `SECFOO_SENSOR` unset, the probe makes no connection and the app exits 0. | +| `call` | The first line is `hello`, there is a `session` line, and the watched symbol has a total `count` of 3. | +| `nocall` | There is a session but no hit. | +| `privacy` | No sentinel value, concrete URL path, query string or `Authorization` value appears on the wire. | +| `fail-open` | The app finishes and exits 0 although the sensor died mid-run. | +| `overhead` | The marginal cost of a watched call (measured from two loop sizes, so start-up doesn't count) is at most `--max-call-overhead-us` (default 50 µs), and the probe's one-time start-up and exit work is under `--max-startup-seconds` (default 5 s). | + +A runtime probe issue is **done** only when its probe passes this kit. CI runs +it against the reference Python probe on every push. + +## 7. Versioning + +`hello.protocol` and `welcome.protocol` are `1`. A future incompatible +change bumps the version. A sensor that doesn't support a probe's version +closes the connection, and the probe then behaves as if the sensor were gone +(§5). diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index 0d2cb88..ec48f84 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -1816,6 +1816,52 @@ def skills_show( sys.stdout.write(render_prompt(skill, context, depth=depth.value) + "\n") +sensor_app = typer.Typer(name="sensor", help="Runtime sensor tools: the probe conformance kit (#73).") +app.add_typer(sensor_app, name="sensor") + + +@sensor_app.command( + name="conformance", + context_settings={"allow_extra_args": True, "ignore_unknown_options": True}, +) +def sensor_conformance( + ctx: typer.Context, + symbol: Annotated[str, typer.Option( + "--symbol", help="Watched symbol the sample app calls, in the #52 notation for its ecosystem.", + )] = "secfoo.runtime.conformance.vulnerable.parse", + ecosystem: Annotated[str, typer.Option("--ecosystem", help="OSV ecosystem of --symbol.")] = "PyPI", + max_call_overhead_us: Annotated[float, typer.Option( + "--max-call-overhead-us", help="Allowed extra microseconds per watched call (#51 budget: 1 ms p99 per request).", + )] = 50.0, + max_startup_seconds: Annotated[float, typer.Option( + "--max-startup-seconds", help="Allowed one-time probe start-up plus exit work.", + )] = 5.0, +) -> None: + """Check a runtime probe against the #73 contract: secfoo sensor conformance -- + + The command must start the sample app with the probe loaded (see + docs/RUNTIME_PROTOCOL.md). For the reference Python probe: + secfoo sensor conformance -- python -m secfoo.sensor -m secfoo.runtime.conformance.sample_app + """ + from secfoo.runtime.conformance.kit import run_kit + + command = list(ctx.args) + if command[:1] == ["--"]: + command = command[1:] + if not command: + err_console.print("[red]Error:[/] give the sample app's start command after --") + raise typer.Exit(code=2) + results = run_kit(command, symbol=symbol, ecosystem=ecosystem, max_call_overhead_us=max_call_overhead_us, + max_startup_seconds=max_startup_seconds) + for result in results: + mark = "[green]PASS[/]" if result.passed else "[red]FAIL[/]" + console.print(f"{mark} {result.name:<10} {result.detail}") + failed = [r.name for r in results if not r.passed] + console.print(f"{len(results) - len(failed)}/{len(results)} checks passed.") + if failed: + raise typer.Exit(code=1) + + _SITECUSTOMIZE = """\ # Written by `secfoo instrument` (#55): starts the secfoo runtime probe in # every Python process of the wrapped command. Inert unless SECFOO_SENSOR=1. diff --git a/src/secfoo/runtime/conformance/__init__.py b/src/secfoo/runtime/conformance/__init__.py new file mode 100644 index 0000000..e42fc3b --- /dev/null +++ b/src/secfoo/runtime/conformance/__init__.py @@ -0,0 +1 @@ +"""Conformance kit for runtime probes (#73). See kit.py and docs/RUNTIME_PROTOCOL.md.""" diff --git a/src/secfoo/runtime/conformance/kit.py b/src/secfoo/runtime/conformance/kit.py new file mode 100644 index 0000000..4f03fd8 --- /dev/null +++ b/src/secfoo/runtime/conformance/kit.py @@ -0,0 +1,223 @@ +"""`secfoo sensor conformance -- `: checks a runtime probe against #73. + +The kit starts a fake sensor on loopback TCP, runs the command once per +step of the sample-app contract (see sample_app.py) with the probe pointed +at it, and checks: + + inert with SECFOO_SENSOR unset, the probe connects to nothing + call a hello, then valid #53 lines, and a hit for the watched symbol (count 3) + nocall a session but no hit + privacy no argument values, concrete URL paths, query strings, header values + fail-open the app keeps working, and exits 0, when the sensor dies mid-run + overhead a watched call costs at most --max-call-overhead-us more (marginal, from two loop sizes), + and the probe's one-time start/exit work stays under --max-startup-seconds + +Every line in every step must validate as a protocol message or #53 event. +""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess # nosec B404 -- runs the probe author's own sample app +import tempfile +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path + +from secfoo.runtime.protocol import ( + PROTOCOL_VERSION, + SENTINEL, + ProtocolError, + forbidden_data, + validate_message, +) + +DEFAULT_SYMBOL = "secfoo.runtime.conformance.vulnerable.parse" +FINGERPRINT = "conformance-fp" +STEP_TIMEOUT_SECONDS = 60 + + +class FakeSensor: + """Loopback TCP listener that records every line and answers hello with + welcome. `kill_after_first_batch` closes everything after the first + batch, the way a crashed sidecar would.""" + + def __init__(self, *, kill_after_first_batch: bool = False) -> None: + self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._server.bind(("127.0.0.1", 0)) + self._server.listen(16) + self._server.settimeout(0.2) + self.port = self._server.getsockname()[1] + self.kill = kill_after_first_batch + self.lines: list[str] = [] + self.connections = 0 + self._stop = threading.Event() + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + + @property + def address(self) -> str: + return f"tcp://127.0.0.1:{self.port}" + + def _serve(self) -> None: + while not self._stop.is_set(): + try: + conn, _ = self._server.accept() + except (TimeoutError, OSError): + continue + self.connections += 1 + threading.Thread(target=self._read, args=(conn,), daemon=True).start() + + def _read(self, conn: socket.socket) -> None: + conn.settimeout(STEP_TIMEOUT_SECONDS) + buffer = b"" + try: + while True: + chunk = conn.recv(65536) + if not chunk: + break + buffer += chunk + *complete, buffer = buffer.split(b"\n") + for raw in complete: + line = raw.decode("utf-8", errors="replace") + self.lines.append(line) + if line.startswith('{"type":"hello"'): + welcome = {"type": "welcome", "protocol": PROTOCOL_VERSION, "flush_interval": 0.1} + conn.sendall((json.dumps(welcome) + "\n").encode()) + if self.kill and len(self.lines) > 1: + self.close() + break + except OSError: + pass + finally: + conn.close() + + def close(self) -> None: + self._stop.set() + try: + self._server.close() + except OSError: + pass + + +@dataclass +class StepResult: + name: str + passed: bool + detail: str = "" + lines: list[str] = field(default_factory=list) + + +def _watchlist(path: Path, symbol: str, ecosystem: str) -> Path: + path.write_text(json.dumps({"version": 1, "generated_at": "2026-01-01T00:00:00Z", "entries": [{ + "finding_fingerprint": FINGERPRINT, "advisory": None, "ecosystem": ecosystem, "package": "conformance", + "version": "1.0", "granularity": "symbol", "symbols": [symbol], "source": "osv"}]}), encoding="utf-8") + return path + + +def _run(command: list[str], env: dict[str, str]) -> tuple[subprocess.CompletedProcess, float]: + started = time.perf_counter() + result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=STEP_TIMEOUT_SECONDS, # nosec B603 + check=False) + return result, time.perf_counter() - started + + +def _parse(lines: list[str]) -> tuple[list[dict], list[str]]: + messages, problems = [], [] + for number, line in enumerate(lines, start=1): + try: + messages.append(validate_message(json.loads(line))) + except (json.JSONDecodeError, ProtocolError) as exc: + problems.append(f"line {number}: {exc}") + return messages, problems + + +def _hit_count(messages: list[dict], symbol: str) -> int: + latest: dict = {} + for message in messages: + if message.get("type") == "hit" and message.get("symbol") == symbol: + latest[(message["session"], message.get("route", ""))] = message["count"] # running totals + return sum(latest.values()) + + +def run_kit(command: list[str], *, symbol: str = DEFAULT_SYMBOL, ecosystem: str = "PyPI", + max_call_overhead_us: float = 50.0, loop_calls: int = 200000, + max_startup_seconds: float = 5.0) -> list[StepResult]: + results: list[StepResult] = [] + with tempfile.TemporaryDirectory(prefix="secfoo-conformance-") as tmp: + watchlist = _watchlist(Path(tmp) / "watchlist.json", symbol, ecosystem) + base = {k: v for k, v in os.environ.items() if not k.startswith("SECFOO_")} + base.update(SECFOO_WATCHLIST=str(watchlist), SECFOO_FLUSH_INTERVAL="0.1", + SECFOO_CONFORMANCE_SENTINEL=SENTINEL, SECFOO_APP_ROOT=tmp) + + def probe_env(sensor: FakeSensor, step: str, **extra: str) -> dict[str, str]: + return {**base, "SECFOO_SENSOR": "1", "SECFOO_SENSOR_ADDR": sensor.address, + "SECFOO_CONFORMANCE_STEP": step, **extra} + + # inert: the probe must not even connect without SECFOO_SENSOR=1 + sensor = FakeSensor() + env = {**base, "SECFOO_SENSOR_ADDR": sensor.address, "SECFOO_CONFORMANCE_STEP": "call"} + proc, _ = _run(command, env) + sensor.close() + results.append(StepResult( + "inert", proc.returncode == 0 and sensor.connections == 0, + f"exit {proc.returncode}, {sensor.connections} connection(s) with SECFOO_SENSOR unset", sensor.lines)) + + for step, check in (("call", "hit"), ("nocall", "nohit"), ("request", "privacy")): + sensor = FakeSensor() + proc, _ = _run(command, probe_env(sensor, step)) + time.sleep(0.2) + sensor.close() + messages, problems = _parse(sensor.lines) + if proc.returncode != 0: + problems.append(f"app exited {proc.returncode}: {proc.stderr.strip()[-300:]}") + if not messages or messages[0].get("type") != "hello": + problems.append("the first line must be hello") + if not any(m.get("type") == "session" for m in messages): + problems.append("no session line") + count = _hit_count(messages, symbol) + if check == "hit" and count != 3: + problems.append(f"expected 3 calls of {symbol}, got {count}") + if check == "nohit" and count: + problems.append(f"expected no hit, got {count}") + if check == "privacy": + problems += forbidden_data(sensor.lines) + name = {"hit": "call", "nohit": "nocall", "privacy": "privacy"}[check] + results.append(StepResult(name, not problems, "; ".join(problems) or f"{len(sensor.lines)} valid line(s)", + sensor.lines)) + + # fail-open: the sensor dies after the first batch; the app must not care + sensor = FakeSensor(kill_after_first_batch=True) + proc, elapsed = _run(command, probe_env(sensor, "slow")) + ok = proc.returncode == 0 and "ok" in proc.stdout + results.append(StepResult("fail-open", ok, f"exit {proc.returncode} after {elapsed:.1f}s with the sensor killed")) + + # overhead: the MARGINAL cost of a watched call, from two loop sizes, + # so interpreter start-up and the probe's one-time start/exit work + # (reported separately) don't swamp the per-call number. + def timed(n: int, probed: bool) -> float: + extra = {"SECFOO_CONFORMANCE_N": str(n)} + if not probed: + return _run(command, {**base, "SECFOO_CONFORMANCE_STEP": "loop", **extra})[1] + sensor = FakeSensor() + try: + return _run(command, probe_env(sensor, "loop", **extra))[1] + finally: + sensor.close() + + small, large = loop_calls, loop_calls * 5 + bare_small, bare_large = timed(small, False), timed(large, False) + probe_small, probe_large = timed(small, True), timed(large, True) + bare_per_call = max(bare_large - bare_small, 1e-6) / (large - small) + probe_per_call = max(probe_large - probe_small, 0.0) / (large - small) + extra_us = (probe_per_call - bare_per_call) * 1e6 + fixed = max(probe_small - bare_small - probe_per_call * small, 0.0) + results.append(StepResult( + "overhead", extra_us <= max_call_overhead_us and fixed <= max_startup_seconds, + f"watched call {bare_per_call * 1e6:.2f}us bare -> {probe_per_call * 1e6:.2f}us probed " + f"(+{extra_us:.2f}us per call, budget {max_call_overhead_us:.0f}us); " + f"one-time start/exit {fixed:.2f}s (budget {max_startup_seconds:.0f}s)")) + return results diff --git a/src/secfoo/runtime/conformance/sample_app.py b/src/secfoo/runtime/conformance/sample_app.py new file mode 100644 index 0000000..eed6c8b --- /dev/null +++ b/src/secfoo/runtime/conformance/sample_app.py @@ -0,0 +1,62 @@ +"""Reference sample app for the #73 conformance kit (Python). + +A probe for another runtime ships its own sample app that follows the same +contract: read SECFOO_CONFORMANCE_STEP and do exactly this, then exit 0. + + call call the watched function 3 times + nocall call only the safe function + request handle one request to the concrete URL + /users/8675309?token= (route template "GET /users/{id}") + with an "Authorization: Bearer " header, passing both + to the watched function as arguments; none of it may leave the app + slow call the watched function, then keep working for ~1.5 s (the + kit kills its fake sensor meanwhile; the app must not notice) + loop call the watched function SECFOO_CONFORMANCE_N times (overhead) + +Run it through the probe, e.g.: + secfoo sensor conformance -- python -m secfoo.sensor -m secfoo.runtime.conformance.sample_app +""" + +from __future__ import annotations + +import os +import sys +import time + +from secfoo.runtime.conformance import vulnerable + +SENTINEL = os.environ.get("SECFOO_CONFORMANCE_SENTINEL", "") + + +def handle_request(path: str, headers: dict[str, str]) -> str: + # A real app would route "GET /users/{id}" here. + return vulnerable.parse(path + headers.get("Authorization", "")) + + +def main() -> int: + step = os.environ.get("SECFOO_CONFORMANCE_STEP", "call") + if step == "call": + for _ in range(3): + vulnerable.parse("payload") + elif step == "nocall": + for _ in range(3): + vulnerable.safe("payload") + elif step == "request": + handle_request(f"/users/8675309?token={SENTINEL}", {"Authorization": f"Bearer {SENTINEL}"}) + elif step == "slow": + deadline = time.monotonic() + 1.5 + while time.monotonic() < deadline: + vulnerable.parse("payload") + time.sleep(0.05) + elif step == "loop": + for _ in range(int(os.environ.get("SECFOO_CONFORMANCE_N", "100000"))): + vulnerable.parse("payload") + else: + print(f"unknown step {step!r}", file=sys.stderr) + return 2 + print(f"conformance step {step} ok") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/secfoo/runtime/conformance/vulnerable.py b/src/secfoo/runtime/conformance/vulnerable.py new file mode 100644 index 0000000..5f16f67 --- /dev/null +++ b/src/secfoo/runtime/conformance/vulnerable.py @@ -0,0 +1,10 @@ +"""Stand-in for a vulnerable dependency, used by the #73 conformance kit.""" + + +def parse(text): + """The "vulnerable" function the kit's watchlist names.""" + return text[::-1] + + +def safe(text): + return text diff --git a/src/secfoo/runtime/protocol.py b/src/secfoo/runtime/protocol.py new file mode 100644 index 0000000..eefca44 --- /dev/null +++ b/src/secfoo/runtime/protocol.py @@ -0,0 +1,95 @@ +"""Probe <-> sensor protocol (#73): one contract for every runtime. + +Spec: docs/RUNTIME_PROTOCOL.md. Schema: protocol.schema.json (next to this +module). A stream is newline-framed JSON: the probe sends `hello` (and, for +TCP with SECFOO_SENSOR_KEY set, `auth`) followed by #53 event lines; the +sensor may answer `welcome` and later `slow`. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable + +from secfoo.runtime.evidence import LANGUAGES, EvidenceError, validate_event + +PROTOCOL_VERSION = 1 +CONTROL_TYPES = frozenset({"hello", "welcome", "slow", "auth"}) + +# Values a probe must never send (#51 rule 3): argument values, concrete URL +# paths, query strings, header values, user ids. The conformance kit plants +# this sentinel in all of them and then looks for it, plus these patterns. +SENTINEL = "s3cr3t-SENTINEL-73" +FORBIDDEN_PATTERNS = ( + re.compile(re.escape(SENTINEL)), + re.compile(r"\?[\w.~-]+="), # a query string + re.compile(r"(?i)\b(bearer|basic)\s+[A-Za-z0-9._~+/-]{6,}"), # an Authorization header value + re.compile(r"/users/8675309"), # the kit's concrete URL path +) + + +class ProtocolError(ValueError): + pass + + +def _text(message: dict, key: str) -> str: + value = message.get(key) + if not isinstance(value, str) or not value.strip() or len(value) > 256: + raise ProtocolError(f"{message.get('type')}: {key!r} must be a non-empty string") + return value + + +def validate_control(message: dict) -> dict: + kind = message.get("type") + if kind == "hello": + if message.get("protocol") != PROTOCOL_VERSION: + raise ProtocolError(f"hello: protocol must be {PROTOCOL_VERSION}") + runtime = _text(message, "runtime") + if runtime not in LANGUAGES: + raise ProtocolError(f"hello: runtime must be one of {', '.join(sorted(LANGUAGES))}") + _text(message, "runtime_version") + _text(message, "probe_version") + pid = message.get("pid") + if isinstance(pid, bool) or not isinstance(pid, int) or pid <= 0: + raise ProtocolError("hello: pid must be a positive integer") + _text(message, "host") + elif kind == "welcome": + if message.get("protocol") != PROTOCOL_VERSION: + raise ProtocolError(f"welcome: protocol must be {PROTOCOL_VERSION}") + interval = message.get("flush_interval") + if not isinstance(interval, (int, float)) or isinstance(interval, bool) or interval <= 0: + raise ProtocolError("welcome: flush_interval must be a positive number of seconds") + elif kind == "slow": + factor = message.get("sample_every") + if isinstance(factor, bool) or not isinstance(factor, int) or factor < 1: + raise ProtocolError("slow: sample_every must be an integer >= 1") + elif kind == "auth": + _text(message, "key") + else: + raise ProtocolError(f"unknown control message type {kind!r}") + return message + + +def validate_message(message: object) -> dict: + """A control message or a #53 event line; raises ProtocolError otherwise.""" + if not isinstance(message, dict): + raise ProtocolError("each line must be a JSON object") + if message.get("type") in CONTROL_TYPES: + return validate_control(message) + try: + return validate_event(message) + except EvidenceError as exc: + raise ProtocolError(str(exc)) from None + + +def forbidden_data(lines: Iterable[str]) -> list[str]: + """Lines that carry data a probe must never send, with the reason.""" + problems = [] + for number, line in enumerate(lines, start=1): + if '"type":"auth"' in line.replace(" ", ""): + continue # the shared key is the one secret that belongs on the wire + for pattern in FORBIDDEN_PATTERNS: + if pattern.search(line): + problems.append(f"line {number}: matches forbidden pattern {pattern.pattern!r}") + break + return problems diff --git a/src/secfoo/runtime/protocol.schema.json b/src/secfoo/runtime/protocol.schema.json new file mode 100644 index 0000000..a3bf272 --- /dev/null +++ b/src/secfoo/runtime/protocol.schema.json @@ -0,0 +1,60 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/secfoo-com/secfoo/blob/main/src/secfoo/runtime/protocol.schema.json", + "title": "Secfoo probe <-> sensor protocol message (one newline-framed line)", + "description": "#73. Control messages are defined here; every other line is a #53 event (evidence.schema.json). secfoo/runtime/protocol.py:validate_message enforces the same rules.", + "oneOf": [ + {"$ref": "#/$defs/hello"}, + {"$ref": "#/$defs/welcome"}, + {"$ref": "#/$defs/slow"}, + {"$ref": "#/$defs/auth"}, + {"$ref": "evidence.schema.json"} + ], + "$defs": { + "text": {"type": "string", "minLength": 1, "maxLength": 256, "pattern": "\\S"}, + "hello": { + "description": "Probe -> sensor, first line on every connection.", + "type": "object", + "required": ["type", "protocol", "runtime", "runtime_version", "probe_version", "pid", "host"], + "properties": { + "type": {"const": "hello"}, + "protocol": {"const": 1}, + "runtime": {"enum": ["python", "node", "jvm", "dotnet", "go", "rust", "ruby", "php", "native"]}, + "runtime_version": {"$ref": "#/$defs/text"}, + "probe_version": {"$ref": "#/$defs/text"}, + "pid": {"type": "integer", "minimum": 1}, + "host": {"$ref": "#/$defs/text", "description": "Host or pod name."} + } + }, + "welcome": { + "description": "Sensor -> probe, the answer to hello. Probes honour flush_interval.", + "type": "object", + "required": ["type", "protocol", "flush_interval"], + "properties": { + "type": {"const": "welcome"}, + "protocol": {"const": 1}, + "flush_interval": {"type": "number", "exclusiveMinimum": 0, "description": "Seconds between flushes."}, + "watchlist_version": {"$ref": "#/$defs/text"}, + "watchlist_path": {"$ref": "#/$defs/text"} + } + }, + "slow": { + "description": "Sensor -> probe backpressure: flush less often / sample more coarsely until further notice.", + "type": "object", + "required": ["type", "sample_every"], + "properties": { + "type": {"const": "slow"}, + "sample_every": {"type": "integer", "minimum": 1} + } + }, + "auth": { + "description": "Probe -> sensor, right after hello on tcp:// when SECFOO_SENSOR_KEY is set (other containers in a pod share loopback).", + "type": "object", + "required": ["type", "key"], + "properties": { + "type": {"const": "auth"}, + "key": {"$ref": "#/$defs/text"} + } + } + } +} diff --git a/src/secfoo/sensor/probe.py b/src/secfoo/sensor/probe.py index 587d8c7..e0134c4 100644 --- a/src/secfoo/sensor/probe.py +++ b/src/secfoo/sensor/probe.py @@ -74,6 +74,7 @@ def __init__(self, *, watchlist_path: str | None, transport, app_root: str | Non self._watched: dict = {} # code object -> (symbol, [fingerprints]) self._pending_symbols: dict = {} # symbol -> [fingerprints], not resolved yet self._seen_app_code: set = set() + self._app_code_cache: dict = {} # code object -> is app code self._tool_id: int | None = None self._stop = threading.Event() self._thread: threading.Thread | None = None @@ -212,10 +213,10 @@ def _on_py_start(self, code, _offset): 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): + if code not in self._seen_app_code and self._is_app_code_obj(code): self._seen_app_code.add(code) return None # keep counting watched functions - if code.co_name != "" and self._is_app_code(code.co_filename): + if code.co_name != "" and self._is_app_code_obj(code): self._seen_app_code.add(code) return sys.monitoring.DISABLE except Exception as exc: # noqa: BLE001 -- never raise into the app @@ -232,7 +233,7 @@ def profiler(frame, event, _arg): 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)): + and self._is_app_code_obj(code)): self._seen_app_code.add(code) except Exception as exc: # noqa: BLE001 self._disable(exc) @@ -242,6 +243,14 @@ def profiler(frame, event, _arg): # ---------- recording ---------- + def _is_app_code_obj(self, code) -> bool: + """_is_app_code for a code object, cached: path normalisation is far + too slow for the per-call path.""" + known = self._app_code_cache.get(code) + if known is None: + known = self._app_code_cache[code] = self._is_app_code(code.co_filename) + return known + def _is_app_code(self, filename: str) -> bool: if not filename or filename.startswith("<"): return False @@ -313,10 +322,15 @@ def _loads(self) -> list[dict]: from importlib.metadata import packages_distributions, version except ImportError: return [] + global _DISTRIBUTIONS + if _DISTRIBUTIONS is None: + # Installed packages don't change while the app runs: scan their + # metadata once, not on every flush (it's the slowest part). + _DISTRIBUTIONS = {top: list(dists) for top, dists in packages_distributions().items()} imported = {name.split(".")[0] for name in list(sys.modules)} seen: set[str] = set() loads = [] - for top, dists in packages_distributions().items(): + for top, dists in _DISTRIBUTIONS.items(): if top not in imported: continue for dist in dists: @@ -324,8 +338,10 @@ def _loads(self) -> list[dict]: continue seen.add(dist.lower()) try: + if dist not in _VERSIONS: + _VERSIONS[dist] = version(dist) loads.append({"type": "load", "session": self.session_id, "ecosystem": "PyPI", - "package": dist, "version": version(dist), "first_seen": self.started_at}) + "package": dist, "version": _VERSIONS[dist], "first_seen": self.started_at}) except Exception: # noqa: BLE001, S112 -- broken package metadata: skip it continue return loads @@ -341,9 +357,15 @@ def flush(self) -> None: except Exception as exc: # noqa: BLE001 self._disable(exc) + def _interval(self) -> float: + """The sensor's `welcome` interval if it sent one, doubled while it + says `slow` (#73 backpressure); otherwise SECFOO_FLUSH_INTERVAL.""" + interval = getattr(self.transport, "flush_interval", None) or self.flush_seconds + return interval * 2 if getattr(self.transport, "slow", False) else interval + def _start_flusher(self) -> None: def loop(): - while not self._stop.wait(self.flush_seconds): + while not self._stop.wait(self._interval()): self.flush() self._thread = threading.Thread(target=loop, name="secfoo-sensor-flush", daemon=True) @@ -381,22 +403,38 @@ def _count_app_functions(probe: Probe) -> int: 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): + if not probe._is_app_code(filename): + continue + try: + mtime = os.path.getmtime(filename) + except OSError: + continue + cached = _FUNCTION_COUNTS.get(filename) + if cached is not None and cached[0] == mtime: # compile each file once, not every flush + total += cached[1] continue try: source = Path(filename).read_text(encoding="utf-8") stack = [compile(source, filename, "exec")] except (OSError, SyntaxError, ValueError, UnicodeDecodeError): continue + file_total = 0 while stack: code = stack.pop() for const in code.co_consts: if hasattr(const, "co_code"): - total += 1 + file_total += 1 stack.append(const) + _FUNCTION_COUNTS[filename] = (mtime, file_total) + total += file_total return total +_FUNCTION_COUNTS: dict[str, tuple[float, int]] = {} # app file -> (mtime, function count) +_DISTRIBUTIONS: dict[str, list[str]] | None = None # top-level module -> distributions +_VERSIONS: dict[str, str] = {} + + 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.""" @@ -412,7 +450,9 @@ def find_spec(self, fullname, path=None, target=None): return None self._busy.on = True try: - spec = importlib.util.find_spec(fullname) if path is None else None + # Submodules (pkg.mod) arrive with the package's __path__. + spec = (importlib.util.find_spec(fullname) if path is None + else importlib.machinery.PathFinder.find_spec(fullname, path)) except (ImportError, ValueError): spec = None finally: @@ -433,4 +473,5 @@ def exec_module(module): return spec +import importlib.machinery import importlib.util diff --git a/src/secfoo/sensor/transport.py b/src/secfoo/sensor/transport.py index 6cba198..2aa60a1 100644 --- a/src/secfoo/sensor/transport.py +++ b/src/secfoo/sensor/transport.py @@ -79,6 +79,9 @@ def __init__(self, address: str) -> None: def _connect(self): raise NotImplementedError + def _after_hello(self) -> list[dict]: + return [] + def _write(self, data: bytes) -> None: raise NotImplementedError @@ -86,7 +89,7 @@ def send(self, events: list[dict]) -> bool: try: if self._conn is None: self._connect() - self._write(_lines([hello()])) + self._write(_lines([hello(), *self._after_hello()])) self._write(_lines(events)) return True except OSError: @@ -122,6 +125,41 @@ def __init__(self, address: str) -> None: else: self.target = parsed.path + def send(self, events: list[dict]) -> bool: + ok = super().send(events) + if ok: + self._read_control() + return ok + + def _read_control(self) -> None: + """Non-blocking read of what the sensor said back (#73): `welcome` + sets the flush interval, `slow` asks for coarser flushing.""" + conn = self._conn + if conn is None: + return + try: + conn.setblocking(False) + data = conn.recv(65536) + except (BlockingIOError, InterruptedError): + return + except OSError: + self.close() + return + finally: + if self._conn is not None: + self._conn.settimeout(SOCKET_TIMEOUT_SECONDS) + for raw in data.splitlines(): + try: + message = json.loads(raw) + except ValueError: + continue + if message.get("type") == "welcome" and isinstance(message.get("flush_interval"), (int, float)): + self.flush_interval = max(float(message["flush_interval"]), 0.05) + elif message.get("type") == "slow": + self.slow = True + + flush_interval: float | None = None + def _connect(self): family = socket.AF_INET6 if self.scheme == "tcp" and ":" in self.target[0] else socket.AF_INET if self.scheme == "unix": @@ -134,9 +172,12 @@ def _connect(self): conn.close() raise self._conn = conn + + def _after_hello(self) -> list[dict]: + # #73: on tcp://, other containers in the pod share loopback, so an + # optional shared key follows hello. key = os.environ.get("SECFOO_SENSOR_KEY") - if key and self.scheme == "tcp": - self._write(_lines([{"type": "auth", "key": key}])) + return [{"type": "auth", "key": key}] if key and self.scheme == "tcp" else [] def _write(self, data: bytes) -> None: self._conn.sendall(data) diff --git a/tests/test_runtime/test_protocol.py b/tests/test_runtime/test_protocol.py new file mode 100644 index 0000000..5d0dfb0 --- /dev/null +++ b/tests/test_runtime/test_protocol.py @@ -0,0 +1,193 @@ +"""#73: probe <-> sensor protocol, schemas and the conformance kit.""" +from __future__ import annotations + +import json +import socket +import sys +import textwrap +import threading +from pathlib import Path + +import pytest + +import secfoo +from secfoo.runtime import protocol +from secfoo.runtime.conformance.kit import run_kit +from secfoo.runtime.protocol import ( + SENTINEL, + ProtocolError, + forbidden_data, + validate_message, +) +from secfoo.sensor import transport as transport_mod + +RUNTIME_DIR = Path(protocol.__file__).parent +REFERENCE = [sys.executable, "-m", "secfoo.sensor", "-m", "secfoo.runtime.conformance.sample_app"] + + +def _hello(**overrides): + return {**transport_mod.hello(), **overrides} + + +# ---------- messages ---------- + + +def test_the_reference_probe_hello_is_valid(): + validate_message(_hello()) + + +@pytest.mark.parametrize("message, field", [ + (_hello(protocol=2), "protocol"), + (_hello(runtime="cobol"), "runtime"), + (_hello(pid=0), "pid"), + ({"type": "welcome", "protocol": 1, "flush_interval": 0}, "flush_interval"), + ({"type": "slow", "sample_every": 0}, "sample_every"), + ({"type": "auth", "key": ""}, "key"), + ({"type": "nonsense"}, "type"), +]) +def test_invalid_control_messages_name_the_field(message, field): + with pytest.raises(ProtocolError, match=field): + validate_message(message) + + +def test_event_lines_are_validated_with_the_53_rules(): + validate_message({"type": "session", "id": "s", "language": "go", "source": "sensor", + "started_at": "2026-11-12T00:00:00Z"}) + with pytest.raises(ProtocolError, match="template"): + validate_message({"type": "hit", "session": "s", "finding_fingerprint": "f", "symbol": "x.y", + "count": 1, "route": "GET /a?b=c"}) + + +def test_forbidden_data_detection(): + clean = json.dumps({"type": "hit", "route": "GET /users/{id}", "symbol": "x.y", "count": 1}) + leaky = [json.dumps({"type": "hit", "arg": SENTINEL}), + json.dumps({"type": "hit", "route": "/search?q=1"}), + json.dumps({"type": "hit", "header": "Bearer abcdef123456"}), + json.dumps({"type": "hit", "route": "GET /users/8675309"})] + assert forbidden_data([clean]) == [] + assert len(forbidden_data(leaky)) == 4 + assert forbidden_data([json.dumps({"type": "auth", "key": "Bearer abcdefgh"})]) == [] # the one allowed secret + + +def test_json_schema_agrees_with_the_validator(): + jsonschema = pytest.importorskip("jsonschema") + from referencing import Registry, Resource + + schemas = {name: json.loads((RUNTIME_DIR / name).read_text(encoding="utf-8")) + for name in ("protocol.schema.json", "evidence.schema.json")} + registry = Registry().with_resources( + (name, Resource.from_contents(body)) for name, body in schemas.items() + ).with_resources((body["$id"], Resource.from_contents(body)) for body in schemas.values()) + jsonschema.Draft202012Validator.check_schema(schemas["protocol.schema.json"]) + validator = jsonschema.Draft202012Validator(schemas["protocol.schema.json"], registry=registry) + good = [_hello(), {"type": "welcome", "protocol": 1, "flush_interval": 0.5}, + {"type": "slow", "sample_every": 4}, {"type": "auth", "key": "k"}, + {"type": "session", "id": "s", "language": "python", "source": "sensor", "started_at": "2026-11-12T00:00:00Z"}] + for message in good: + validate_message(message) + assert validator.is_valid(message), list(validator.iter_errors(message)) + for message in (_hello(protocol=2), {"type": "slow", "sample_every": 0}, {"type": "welcome", "protocol": 1}): + assert not validator.is_valid(message) + with pytest.raises(ProtocolError): + validate_message(message) + + +# ---------- the reference probe's transport follows the protocol ---------- + + +def _listen(): + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen() + return server, server.getsockname()[1] + + +def test_hello_then_auth_then_events_and_welcome_is_honoured(monkeypatch): + monkeypatch.setenv("SECFOO_SENSOR_KEY", "pod-shared-key") + server, port = _listen() + received = [] + + def sensor(): + conn, _ = server.accept() + data = b"" + while data.count(b"\n") < 3: + data += conn.recv(65536) + conn.sendall(b'{"type":"welcome","protocol":1,"flush_interval":0.25}\n{"type":"slow","sample_every":2}\n') + received.extend(json.loads(line) for line in data.splitlines()) + threading.Event().wait(0.5) + conn.close() + server.close() + + thread = threading.Thread(target=sensor, daemon=True) + thread.start() + tcp = transport_mod.SocketTransport(f"tcp://127.0.0.1:{port}") + assert tcp.send([{"type": "session", "id": "s"}]) + thread.join(0.3) + tcp.send([{"type": "session", "id": "s"}]) # picks up the sensor's answers + assert [m["type"] for m in received[:3]] == ["hello", "auth", "session"] + assert received[1]["key"] == "pod-shared-key" + assert tcp.flush_interval == 0.25 and tcp.slow + + +def test_probe_flush_interval_follows_welcome_and_slow(): + from secfoo.sensor.probe import Probe + + class Stub(transport_mod.NullTransport): + flush_interval = 2.0 + slow = False + + probe = Probe(watchlist_path=None, transport=Stub(), flush_seconds=10) + assert probe._interval() == 2.0 + probe.transport.slow = True + assert probe._interval() == 4.0 + + +# ---------- the kit ---------- + + +def test_reference_python_probe_passes_the_kit(): + results = {r.name: r for r in run_kit(REFERENCE, loop_calls=50000)} + failed = {name: r.detail for name, r in results.items() if not r.passed} + assert not failed, failed + assert set(results) == {"inert", "call", "nocall", "privacy", "fail-open", "overhead"} + + +def test_the_kit_catches_a_misbehaving_probe(tmp_path): + """A 'probe' that ignores SECFOO_SENSOR, sends a concrete URL and leaks the + request's secret must fail inert, call and privacy.""" + bad = tmp_path / "bad_probe.py" + bad.write_text(textwrap.dedent("""\ + import json, os, socket, sys + addr = os.environ["SECFOO_SENSOR_ADDR"].split("//")[1] + host, port = addr.split(":") + s = socket.create_connection((host, int(port))) + hello = {"type": "hello", "protocol": 1, "runtime": "python", "runtime_version": "3", "probe_version": "0", + "pid": os.getpid(), "host": "h"} + hit = {"type": "hit", "session": "s", "finding_fingerprint": "f", "symbol": "x.y", "count": 1, + "route": "GET /users/8675309?token=" + os.environ.get("SECFOO_CONFORMANCE_SENTINEL", "")} + s.sendall((json.dumps(hello) + "\\n" + json.dumps(hit) + "\\n").encode()) + s.close() + print("ok") + """), encoding="utf-8") + results = {r.name: r for r in run_kit([sys.executable, str(bad)], loop_calls=1000)} + assert not results["inert"].passed + assert not results["call"].passed + assert not results["privacy"].passed + assert "template" in results["privacy"].detail or "forbidden" in results["privacy"].detail + + +def test_cli_reports_and_exits_nonzero_on_failure(tmp_path): + from typer.testing import CliRunner + + from secfoo.cli import app + + noop = tmp_path / "noop.py" + noop.write_text("print('ok')\n", encoding="utf-8") + result = CliRunner(env={"COLUMNS": "200"}).invoke(app, ["sensor", "conformance", "--", sys.executable, str(noop)]) + assert result.exit_code == 1 + assert "PASS inert" in result.output and "FAIL call" in result.output + + +def test_sample_app_ships_with_the_package(): + package = Path(secfoo.__file__).parent / "runtime" / "conformance" + assert {"sample_app.py", "vulnerable.py", "kit.py"} <= {p.name for p in package.iterdir()}