From 8b1e6e6bcdb366614d4ab70f77c9dc15ccec40c2 Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 03:25:21 +0000 Subject: [PATCH] feat(runtime): ingest API on secfoo serve with the user's API key (#66) GET watchlist (ETag), POST events (gzip NDJSON through ingest_events, so replays never double-count), POST heartbeat and GET evidence under /api/runtime/v1, authenticated with Bearer keys issued by secfoo serve key create (stored hashed, project- and runtime-scopable, revocable). Unknown projects are rejected, bodies and line counts capped, requests rate-limited per key, and keys refused over plain HTTP on non-loopback binds unless --allow-insecure-runtime. load_sensor_config() reads SECFOO_API_KEY_FILE / SECFOO_SERVER / SECFOO_PROJECT with no fallback account. Closes #66 Co-Authored-By: Claude Opus 5.5 --- docs/REFERENCE.md | 56 ++++++ src/secfoo/cli.py | 78 +++++++- src/secfoo/runtime/api.py | 267 +++++++++++++++++++++++++ src/secfoo/runtime/client.py | 85 ++++++++ src/secfoo/runtime/keys.py | 97 +++++++++ src/secfoo/storage/schema.sql | 32 +++ src/secfoo/web/app.py | 8 +- tests/test_runtime/test_runtime_api.py | 258 ++++++++++++++++++++++++ 8 files changed, 876 insertions(+), 5 deletions(-) create mode 100644 src/secfoo/runtime/api.py create mode 100644 src/secfoo/runtime/client.py create mode 100644 src/secfoo/runtime/keys.py create mode 100644 tests/test_runtime/test_runtime_api.py diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index e601b4f..c4f68e4 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -473,6 +473,62 @@ 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) +### Runtime ingest API + +Live sensors send evidence to the same server your CLI results go to. They +authenticate with **the same Secfoo API key**, so one key gives one view of +static findings and runtime evidence. `secfoo serve` implements this API; the +hosted service implements the same contract. + +**On a self-hosted `secfoo serve`, issue a key:** + +```bash +secfoo serve key create --name payments-prod-sensor --project payments-api --runtime-only +secfoo serve key list +secfoo serve key revoke 3 +secfoo cloud login --api-key --portal-url https://secfoo.internal # the CLI shares the same key +``` + +The key is printed once. Only a SHA-256 of it is stored. `--project` limits a +key to one project, and `--runtime-only` limits it to these endpoints; both +are recommended for production clusters. + +**Sensor configuration**, set explicitly at deploy time: + +```bash +SECFOO_API_KEY_FILE=/var/run/secrets/secfoo/api-key # or SECFOO_API_KEY; mount it from a secret, never bake it into an image +SECFOO_SERVER=https://secfoo.internal # default: the portal URL `secfoo cloud login` saved +SECFOO_PROJECT=https://github.com/acme/payments-api # the identifier exactly as `secfoo run` recorded it +SECFOO_APP=payments-api +SECFOO_ENV=prod +``` + +A missing key, server or project is an error. There is no fallback account. +`SECFOO_SERVER` must be `https://`, or a loopback `http://` address. + +**Endpoints** (`Authorization: Bearer `; `project` is required on each): + +| Endpoint | Purpose | +|---|---| +| `GET /api/runtime/v1/watchlist?project=` | The watchlist above, with an `ETag`. Send `If-None-Match` to get a cheap `304`. | +| `POST /api/runtime/v1/events?project=&app=&env=` | An NDJSON batch in the evidence format (gzip allowed, `Content-Encoding: gzip`). It uses the same validation and idempotent upsert as `secfoo runtime ingest`, so a retried batch never double-counts. `app` / `env` fill session lines that don't set them. Returns the counts stored and any skipped lines. | +| `POST /api/runtime/v1/heartbeat?project=` | JSON `{sensor_id, version, app, env, instances, dropped}`. Registers the sensor and sums dropped events. | +| `GET /api/runtime/v1/evidence?project=` | The runtime state of each open SCA finding (state, routes, coverage, conflict), for the CLI and CI. | + +- **Errors:** `401` for a missing, unknown or revoked key; `403` for a key + without access to the project; `404` for an unknown project (it is not + auto-created; run `secfoo run` first); `413` for oversized batches; `429` + when rate-limited. +- **Limits:** 5 MiB request body, 20 MiB decompressed, 50,000 lines per + batch, and 120 requests per minute per key. +- **Logging:** keys and event payloads are never logged or echoed in errors. + +**TLS.** `secfoo serve` binds `127.0.0.1` by default. On any other `--host`, +it refuses API keys over plain HTTP. Put a TLS-terminating proxy in front +(e.g. Caddy, nginx, or a cloud load balancer forwarding `X-Forwarded-Proto`) +and run uvicorn behind it. `--allow-insecure-runtime` disables the check, for +a trusted private network only. + ## Install Pick whichever fits how you work — they're all the same tool underneath. diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index 38cdb59..e3b39a9 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -617,18 +617,88 @@ def show( console.print("[yellow]No report content available for this run.[/]") -@app.command() +serve_app = typer.Typer( + name="serve", invoke_without_command=True, + help="Launch the local web dashboard (and runtime ingest API). `secfoo serve key` manages its API keys.", +) +serve_key_app = typer.Typer(name="key", help="API keys for runtime sensors and the CLI on this server (#66).") +serve_app.add_typer(serve_key_app, name="key") +app.add_typer(serve_app, name="serve") + + +@serve_app.callback() def serve( - host: str = typer.Option("127.0.0.1", "--host"), - port: int = typer.Option(8787, "--port"), + ctx: typer.Context, + host: Annotated[str, typer.Option("--host")] = "127.0.0.1", + port: Annotated[int, typer.Option("--port")] = 8787, + allow_insecure_runtime: Annotated[bool, typer.Option( + "--allow-insecure-runtime", + help="Accept runtime API keys over plain HTTP on a non-loopback --host. Prefer TLS in front instead.", + )] = False, ) -> None: """Launch the local web dashboard.""" + if ctx.invoked_subcommand is not None: + return import uvicorn from secfoo.web.app import create_app console.print(f"Serving dashboard at [bold]http://{host}:{port}[/]") - uvicorn.run(create_app(), host=host, port=port) + uvicorn.run(create_app(bind_host=host, allow_insecure_runtime=allow_insecure_runtime), host=host, port=port) + + +@serve_key_app.command(name="create") +def serve_key_create( + name: Annotated[str, typer.Option("--name", help="A label, e.g. payments-prod-sensor.")], + project: Annotated[str | None, typer.Option( + "--project", help="Limit the key to one project (id, name or path/URL). Default: every project.", + )] = None, + runtime_only: Annotated[bool, typer.Option( + "--runtime-only", help="Only allow the runtime endpoints (recommended for production clusters).", + )] = False, +) -> None: + """Issue an API key. It is shown once: store it in a secret, never in an image.""" + from secfoo.runtime import keys as keys_module + + with RunRepository() as repo: + project_id = None + if project is not None: + project_id = repo.resolve_project_ref(project) + if project_id is None: + err_console.print(f"[red]Error:[/] no single project matches {project!r}.") + raise typer.Exit(code=1) + key, record = keys_module.create_key( + repo.connection, name=name, scope="runtime" if runtime_only else "all", project_id=project_id + ) + err_console.print(f"Created key {record.id} ({record.name}, scope {record.scope}). It won't be shown again:") + print(key) # stdout, so `KEY=$(secfoo serve key create ...)` captures only the key + + +@serve_key_app.command(name="list") +def serve_key_list() -> None: + """Keys on this server (never the keys themselves).""" + from secfoo.runtime import keys as keys_module + + table = Table(show_header=True, header_style="bold") + for column in ("id", "name", "prefix", "scope", "project", "created", "last used", "status"): + table.add_column(column) + with RunRepository() as repo: + for k in keys_module.list_keys(repo.connection): + table.add_row(str(k.id), k.name, k.key_prefix + "…", k.scope, str(k.project_id or "all"), + k.created_at[:19], (k.last_used_at or "-")[:19], "active" if k.active else "revoked") + console.print(table) + + +@serve_key_app.command(name="revoke") +def serve_key_revoke(key_id: Annotated[int, typer.Argument(help="Key id from `secfoo serve key list`.")]) -> None: + """Revoke a key. Sensors and CLIs using it are rejected from the next request.""" + from secfoo.runtime import keys as keys_module + + with RunRepository() as repo: + if not keys_module.revoke_key(repo.connection, key_id): + err_console.print(f"[red]Error:[/] no active key with id {key_id}.") + raise typer.Exit(code=1) + console.print(f"Revoked key {key_id}.") @app.command() diff --git a/src/secfoo/runtime/api.py b/src/secfoo/runtime/api.py new file mode 100644 index 0000000..d628cd4 --- /dev/null +++ b/src/secfoo/runtime/api.py @@ -0,0 +1,267 @@ +"""Runtime ingest API on `secfoo serve` (#66). + +The contract sensors (#67) and the CLI (#80) speak. All endpoints take +`Authorization: Bearer ` -- the same key the CLI uses -- and a +`project` that must already exist (it was scanned with `secfoo run`): + + GET /api/runtime/v1/watchlist?project= the #52 watchlist, with an ETag + POST /api/runtime/v1/events?project= a (gzip) NDJSON batch in the #53 format + POST /api/runtime/v1/heartbeat?project= sensor id, version, app, env, instances, dropped + GET /api/runtime/v1/evidence?project= runtime state per open SCA finding + +Events go through `secfoo.runtime.evidence.ingest_events`, the same code as +`secfoo runtime ingest`, so a retried batch never double-counts. Keys and +event payloads are never logged. +""" + +from __future__ import annotations + +import gzip +import hashlib +import json +import threading +import time +import zlib +from collections import defaultdict, deque +from datetime import datetime, timezone +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response + +from secfoo.runtime import keys as keys_module +from secfoo.runtime.evidence import ingest_events +from secfoo.runtime.watchlist import build_watchlist +from secfoo.storage.repository import RunRepository + +MAX_BODY_BYTES = 5 * 1024 * 1024 # compressed request body +MAX_DECOMPRESSED_BYTES = 20 * 1024 * 1024 +MAX_BATCH_LINES = 50_000 +RATE_LIMIT_REQUESTS = 120 # per key, per window +RATE_LIMIT_WINDOW_SECONDS = 60 +LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) + +router = APIRouter(prefix="/api/runtime/v1", tags=["runtime"]) + + +class _RateLimiter: + def __init__(self) -> None: + self._hits: dict[int, deque[float]] = defaultdict(deque) + self._lock = threading.Lock() + + def allow(self, key_id: int) -> bool: + now = time.monotonic() + with self._lock: + hits = self._hits[key_id] + while hits and now - hits[0] > RATE_LIMIT_WINDOW_SECONDS: + hits.popleft() + if len(hits) >= RATE_LIMIT_REQUESTS: + return False + hits.append(now) + return True + + def reset(self) -> None: + with self._lock: + self._hits.clear() + + +rate_limiter = _RateLimiter() + + +def _error(status: int, message: str) -> HTTPException: + return HTTPException(status_code=status, detail=message) + + +def _require_secure_transport(request: Request) -> None: + """`secfoo serve` binds loopback by default. On any other address, a key + must not cross the network in clear text unless the operator opted in.""" + state = request.app.state + bind_host = getattr(state, "bind_host", "127.0.0.1") + if bind_host in LOOPBACK_HOSTS or getattr(state, "allow_insecure_runtime", False): + return + if request.url.scheme != "https": + raise _error(403, "refusing an API key over plain HTTP on a non-loopback address; put TLS in front " + "(see docs/REFERENCE.md) or start secfoo serve with --allow-insecure-runtime") + + +class _Caller: + def __init__(self, repo: RunRepository, key: keys_module.ApiKey, project_id: int) -> None: + self.repo, self.key, self.project_id = repo, key, project_id + + +async def caller(request: Request, project: str = Query(..., description="Project identifier as `secfoo run` recorded it")): + # async, like every route here: sqlite connections are bound to the thread + # that opened them, and FastAPI runs sync dependencies in a worker pool. + _require_secure_transport(request) + header = request.headers.get("authorization", "") + scheme, _, token = header.partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + raise _error(401, "missing Authorization: Bearer ") + repo = RunRepository() + try: + key = keys_module.verify_key(repo.connection, token.strip()) + if key is None: + raise _error(401, "invalid or revoked API key") + if not rate_limiter.allow(key.id): + raise _error(429, "rate limit exceeded for this key; retry later") + project_id = repo.find_project_id(project) + if project_id is None: + # Not auto-created: evidence with no scan has nothing to attach to. + raise _error(404, f"unknown project {project!r}; run `secfoo run` against it first") + if not key.can_access(project_id): + raise _error(403, "this API key has no access to that project") + yield _Caller(repo, key, project_id) + finally: + repo.close() + + +def _watchlist_etag(document: dict) -> str: + stable = json.dumps(document["entries"], sort_keys=True).encode("utf-8") + return '"' + hashlib.sha256(stable).hexdigest()[:32] + '"' + + +@router.get("/watchlist") +async def get_watchlist(request: Request, who: Annotated[_Caller, Depends(caller)]): + document, _notes = build_watchlist(who.repo, who.project_id) + etag = _watchlist_etag(document) + if request.headers.get("if-none-match") == etag: + return Response(status_code=304, headers={"ETag": etag}) + return Response( + content=json.dumps(document), media_type="application/json", headers={"ETag": etag, "Cache-Control": "no-cache"} + ) + + +async def _read_body(request: Request) -> bytes: + declared = request.headers.get("content-length") + if declared and declared.isdigit() and int(declared) > MAX_BODY_BYTES: + raise _error(413, f"request body over {MAX_BODY_BYTES} bytes") + body = b"" + async for chunk in request.stream(): + body += chunk + if len(body) > MAX_BODY_BYTES: + raise _error(413, f"request body over {MAX_BODY_BYTES} bytes") + if request.headers.get("content-encoding", "").lower() == "gzip" or body[:2] == b"\x1f\x8b": + try: + decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS) + data = decompressor.decompress(body, MAX_DECOMPRESSED_BYTES + 1) + except (zlib.error, gzip.BadGzipFile): + raise _error(400, "body is not valid gzip") from None + if len(data) > MAX_DECOMPRESSED_BYTES or decompressor.unconsumed_tail: + raise _error(413, f"decompressed body over {MAX_DECOMPRESSED_BYTES} bytes") + return data + return body + + +def _with_app_env(lines: list[str], app: str | None, env: str | None) -> list[str]: + """Live sensors send app/env as query params (SECFOO_APP / SECFOO_ENV); + fill them into session lines that don't name their own.""" + if not (app or env): + return lines + out = [] + for line in lines: + try: + event = json.loads(line) + except json.JSONDecodeError: + out.append(line) # ingest_events reports it with its line number + continue + if isinstance(event, dict) and event.get("type") == "session": + if app and not event.get("app"): + event["app"] = app + if env and not event.get("env"): + event["env"] = env + line = json.dumps(event) + out.append(line) + return out + + +@router.post("/events") +async def post_events( + request: Request, + who: Annotated[_Caller, Depends(caller)], + app: str | None = Query(None, max_length=200), + env: str | None = Query(None, max_length=200), +): + lines = (await _read_body(request)).decode("utf-8", errors="replace").splitlines() + if len(lines) > MAX_BATCH_LINES: + raise _error(413, f"batch over {MAX_BATCH_LINES} lines; flush more often") + report = ingest_events(who.repo.connection, who.project_id, _with_app_env(lines, app, env)) + return { + "sessions": report.sessions, + "loads": report.loads, + "hits": report.hits, + "skipped": [{"line": n, "reason": r} for n, r in report.skipped[:100]], + "skipped_count": len(report.skipped), + "unknown_fingerprints": sorted(report.unknown_fingerprints)[:100], + } + + +def _bounded_text(payload: dict, key: str, *, required: bool = False) -> str | None: + value = payload.get(key) + if value is None and not required: + return None + if not isinstance(value, str) or not value.strip() or len(value) > 200: + raise _error(422, f"{key!r} must be a non-empty string of at most 200 characters") + return value + + +def _count(payload: dict, key: str) -> int: + value = payload.get(key, 0) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise _error(422, f"{key!r} must be a non-negative integer") + return value + + +@router.post("/heartbeat") +async def post_heartbeat(request: Request, who: Annotated[_Caller, Depends(caller)]): + try: + payload = json.loads(await _read_body(request) or b"{}") + except json.JSONDecodeError: + raise _error(400, "heartbeat body must be JSON") from None + if not isinstance(payload, dict): + raise _error(400, "heartbeat body must be a JSON object") + sensor_id = _bounded_text(payload, "sensor_id", required=True) + now = datetime.now(timezone.utc).isoformat() + conn = who.repo.connection + conn.execute( + """ + INSERT INTO runtime_sensors (project_id, sensor_id, version, app, env, instances, dropped_total, + first_seen_at, last_heartbeat_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (project_id, sensor_id) DO UPDATE SET + version = excluded.version, app = excluded.app, env = excluded.env, instances = excluded.instances, + dropped_total = runtime_sensors.dropped_total + excluded.dropped_total, + last_heartbeat_at = excluded.last_heartbeat_at + """, + (who.project_id, sensor_id, _bounded_text(payload, "version"), _bounded_text(payload, "app"), + _bounded_text(payload, "env"), _count(payload, "instances"), _count(payload, "dropped"), now, now), + ) + conn.commit() + return {"ok": True, "received_at": now} + + +@router.get("/evidence") +async def get_evidence(who: Annotated[_Caller, Depends(caller)]): + from secfoo.web import sca_runtime + + findings = [] + for finding in who.repo.list_sca_findings(project_id=who.project_id, status="open"): + status = sca_runtime.status_for( + who.repo.connection, project_id=who.project_id, fingerprint=finding.fingerprint, package=finding.package, + ecosystem=finding.ecosystem, scanned_version=finding.version, + ) + findings.append({ + "finding_fingerprint": finding.fingerprint, + "package": finding.package, + "version": finding.version, + "severity": finding.severity, + "static_reachability": finding.reachability, + "runtime_state": status.state, + "runtime_summary": status.summary, + "hits": status.hits, + "routes": status.routes, + "loaded_versions": status.loaded_versions, + "version_drift": status.version_drift, + "coverage_percent": status.coverage.percent, + "observed_hours": round(status.coverage.observed_hours, 2), + "conflict": sca_runtime.is_conflict(finding.reachability or "", status), + }) + return {"project": who.repo.get_project(who.project_id).identifier, "findings": findings} diff --git a/src/secfoo/runtime/client.py b/src/secfoo/runtime/client.py new file mode 100644 index 0000000..8083046 --- /dev/null +++ b/src/secfoo/runtime/client.py @@ -0,0 +1,85 @@ +"""Sensor-side configuration for the runtime ingest API (#66). + +Set explicitly at deploy time, never guessed: + + SECFOO_API_KEY_FILE=/var/run/secrets/secfoo/api-key # or SECFOO_API_KEY + SECFOO_SERVER=https://app.secfoo.com # default: `secfoo cloud login`'s portal_url + SECFOO_PROJECT=https://github.com/acme/payments-api # exactly as `secfoo run` recorded it + SECFOO_APP=payments-api + SECFOO_ENV=prod + +There is no fallback account: a missing key, server or project is an error. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + + +class SensorConfigError(RuntimeError): + pass + + +@dataclass(frozen=True) +class SensorConfig: + api_key: str + server: str + project: str + app: str | None = None + env: str | None = None + + def __repr__(self) -> str: # never print the key + return f"SensorConfig(server={self.server!r}, project={self.project!r}, app={self.app!r}, env={self.env!r})" + + def url(self, endpoint: str) -> str: + return f"{self.server.rstrip('/')}/api/runtime/v1/{endpoint.lstrip('/')}" + + +def _api_key(environ: dict[str, str]) -> str: + key_file = environ.get("SECFOO_API_KEY_FILE", "").strip() + if key_file: + try: + key = Path(key_file).read_text(encoding="utf-8").strip() + except OSError as exc: + raise SensorConfigError(f"can't read SECFOO_API_KEY_FILE ({key_file}): {exc.strerror}") from None + if not key: + raise SensorConfigError(f"SECFOO_API_KEY_FILE ({key_file}) is empty") + return key + key = environ.get("SECFOO_API_KEY", "").strip() + if key: + return key + raise SensorConfigError("no API key: set SECFOO_API_KEY_FILE (recommended, from a mounted secret) or SECFOO_API_KEY") + + +def _server(environ: dict[str, str]) -> str: + server = environ.get("SECFOO_SERVER", "").strip() + if server: + return server + from secfoo.cloud import CloudError, load_cloud_config + + try: + config = load_cloud_config() + except CloudError: + config = None + if config is None: + raise SensorConfigError("no server: set SECFOO_SERVER, or run `secfoo cloud login` so its portal URL is used") + return config.portal_url + + +def load_sensor_config(environ: dict[str, str] | None = None) -> SensorConfig: + environ = dict(os.environ if environ is None else environ) + project = environ.get("SECFOO_PROJECT", "").strip() + if not project: + raise SensorConfigError("no project: set SECFOO_PROJECT to the identifier `secfoo run` recorded") + server = _server(environ) + if not server.startswith(("https://", "http://127.0.0.1", "http://localhost", "http://[::1]")): + raise SensorConfigError(f"SECFOO_SERVER must be https:// (or a loopback http:// address), got {server!r}") + return SensorConfig( + api_key=_api_key(environ), + server=server, + project=project, + app=environ.get("SECFOO_APP", "").strip() or None, + env=environ.get("SECFOO_ENV", "").strip() or None, + ) diff --git a/src/secfoo/runtime/keys.py b/src/secfoo/runtime/keys.py new file mode 100644 index 0000000..8bce767 --- /dev/null +++ b/src/secfoo/runtime/keys.py @@ -0,0 +1,97 @@ +"""API keys for a self-hosted `secfoo serve` (#66). + +`secfoo serve key create` issues a key that both runtime sensors and the +CLI (`secfoo cloud login --api-key --portal-url `) use, so a +project has one key and one view. Only a SHA-256 of the key is stored; the +plaintext is shown once and never logged. +""" + +from __future__ import annotations + +import hashlib +import secrets +import sqlite3 +from dataclasses import dataclass +from datetime import datetime, timezone + +KEY_PREFIX = "sfk_" +SCOPES = ("all", "runtime") + + +@dataclass(frozen=True) +class ApiKey: + id: int + name: str + key_prefix: str + scope: str + project_id: int | None + created_at: str + last_used_at: str | None + revoked_at: str | None + + @property + def active(self) -> bool: + return self.revoked_at is None + + def can_access(self, project_id: int) -> bool: + return self.active and (self.project_id is None or self.project_id == project_id) + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def hash_key(key: str) -> str: + return hashlib.sha256(key.encode("utf-8")).hexdigest() + + +def _row(row) -> ApiKey: + return ApiKey(*row) + + +_COLUMNS = "id, name, key_prefix, scope, project_id, created_at, last_used_at, revoked_at" + + +def create_key(conn: sqlite3.Connection, *, name: str, scope: str = "all", project_id: int | None = None) -> tuple[str, ApiKey]: + """(plaintext key, record). The plaintext is not recoverable later.""" + if scope not in SCOPES: + raise ValueError(f"scope must be one of {', '.join(SCOPES)}") + if not name.strip(): + raise ValueError("a key needs a name") + key = KEY_PREFIX + secrets.token_urlsafe(32) + cur = conn.execute( + "INSERT INTO runtime_api_keys (name, key_prefix, key_hash, scope, project_id, created_at) VALUES (?, ?, ?, ?, ?, ?)", + (name.strip(), key[:10], hash_key(key), scope, project_id, _now()), + ) + conn.commit() + record = conn.execute(f"SELECT {_COLUMNS} FROM runtime_api_keys WHERE id = ?", (cur.lastrowid,)).fetchone() # nosec B608 + return key, _row(record) + + +def list_keys(conn: sqlite3.Connection) -> list[ApiKey]: + return [_row(r) for r in conn.execute(f"SELECT {_COLUMNS} FROM runtime_api_keys ORDER BY id").fetchall()] # nosec B608 + + +def revoke_key(conn: sqlite3.Connection, key_id: int) -> bool: + cur = conn.execute( + "UPDATE runtime_api_keys SET revoked_at = ? WHERE id = ? AND revoked_at IS NULL", (_now(), key_id) + ) + conn.commit() + return cur.rowcount == 1 + + +def verify_key(conn: sqlite3.Connection, key: str) -> ApiKey | None: + """The active key record for `key`, or None (unknown or revoked).""" + if not key or not key.startswith(KEY_PREFIX): + return None + row = conn.execute( + f"SELECT {_COLUMNS} FROM runtime_api_keys WHERE key_hash = ?", (hash_key(key),) # nosec B608 + ).fetchone() + if row is None: + return None + record = _row(row) + if not record.active: + return None + conn.execute("UPDATE runtime_api_keys SET last_used_at = ? WHERE id = ?", (_now(), record.id)) + conn.commit() + return record diff --git a/src/secfoo/storage/schema.sql b/src/secfoo/storage/schema.sql index 4c5638a..c1b160d 100644 --- a/src/secfoo/storage/schema.sql +++ b/src/secfoo/storage/schema.sql @@ -376,3 +376,35 @@ CREATE TABLE IF NOT EXISTS runtime_packages ( ); CREATE INDEX IF NOT EXISTS idx_runtime_packages_project ON runtime_packages(project_id, ecosystem, package); + +-- #66: API keys a self-hosted `secfoo serve` issues for runtime sensors and +-- the CLI (`secfoo serve key create`). Only a SHA-256 of the key is stored; +-- the key itself is shown once, at creation. `project_id` NULL = every +-- project on this server; `scope` 'runtime' = runtime endpoints only. +CREATE TABLE IF NOT EXISTS runtime_api_keys ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + key_prefix TEXT NOT NULL, -- first characters, for `key list` only + key_hash TEXT NOT NULL UNIQUE, + scope TEXT NOT NULL CHECK (scope IN ('all','runtime')), + project_id INTEGER REFERENCES projects(id), + created_at TEXT NOT NULL, + last_used_at TEXT, + revoked_at TEXT +); + +-- #66: sensors register on their first heartbeat; the runtime page (#72) +-- lists them. `dropped_total` sums the events sensors reported dropping. +CREATE TABLE IF NOT EXISTS runtime_sensors ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id INTEGER NOT NULL REFERENCES projects(id), + sensor_id TEXT NOT NULL, + version TEXT, + app TEXT, + env TEXT, + instances INTEGER, + dropped_total INTEGER NOT NULL DEFAULT 0, + first_seen_at TEXT NOT NULL, + last_heartbeat_at TEXT NOT NULL, + UNIQUE (project_id, sensor_id) +); diff --git a/src/secfoo/web/app.py b/src/secfoo/web/app.py index b727e0e..56130e5 100644 --- a/src/secfoo/web/app.py +++ b/src/secfoo/web/app.py @@ -6,6 +6,7 @@ from fastapi.staticfiles import StaticFiles from secfoo.config import STORE_DIR +from secfoo.runtime import api as runtime_api from secfoo.web.routes import ( activities, agents, @@ -33,8 +34,12 @@ def mermaid_bundle_available() -> bool: return MERMAID_BUNDLE.is_file() -def create_app() -> FastAPI: +def create_app(*, bind_host: str = "127.0.0.1", allow_insecure_runtime: bool = False) -> FastAPI: app = FastAPI(title="secfoo dashboard") + # Read by the runtime ingest API (#66): on a non-loopback bind it refuses + # API keys over plain HTTP unless the operator opted in. + app.state.bind_host = bind_host + app.state.allow_insecure_runtime = allow_insecure_runtime @app.middleware("http") async def revalidate_static(request, call_next): @@ -59,4 +64,5 @@ async def revalidate_static(request, call_next): app.include_router(responsible_ai.router) app.include_router(projects.router) app.include_router(runs.router) + app.include_router(runtime_api.router) return app diff --git a/tests/test_runtime/test_runtime_api.py b/tests/test_runtime/test_runtime_api.py new file mode 100644 index 0000000..3b2c7eb --- /dev/null +++ b/tests/test_runtime/test_runtime_api.py @@ -0,0 +1,258 @@ +"""#66: runtime ingest API on `secfoo serve`, authenticated with the user's API key.""" +from __future__ import annotations + +import gzip +import json + +import pytest +from fastapi.testclient import TestClient +from typer.testing import CliRunner + +from secfoo.cli import app as cli_app +from secfoo.runtime import api as runtime_api +from secfoo.runtime import keys +from secfoo.runtime.client import SensorConfigError, load_sensor_config +from secfoo.storage.repository import RunRepository +from secfoo.web.app import create_app + +PROJECT = "https://github.com/acme/payments-api" +FP = "fp-yaml" +SESSION = {"type": "session", "id": "s-1", "language": "python", "source": "sensor", + "started_at": "2026-11-12T10:00:00Z", "ended_at": "2026-11-12T11:00:00Z", + "app_functions_total": 100, "app_functions_executed": 40} +HIT = {"type": "hit", "session": "s-1", "finding_fingerprint": FP, "symbol": "yaml.load", + "route": "POST /upload", "auth": "anonymous", "count": 7} + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr("secfoo.storage.repository.DB_PATH", tmp_path / "db.sqlite") + monkeypatch.setattr("secfoo.storage.repository.ensure_store_dirs", lambda: None) + runtime_api.rate_limiter.reset() + with RunRepository() as repo: + pid = repo.upsert_project(PROJECT, "payments-api", "github") + other = repo.upsert_project("https://github.com/acme/other", "other", "github") + 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, run_id=run_id, current_ref="D1", package="pyyaml", + severity="High", version="5.3", ecosystem="PyPI", reachability="Not reachable") + repo.upsert_osv_lookup(ecosystem="PyPI", package="pyyaml", version="5.3", vulns=[]) + key, _ = keys.create_key(repo.connection, name="all-projects") + scoped, _ = keys.create_key(repo.connection, name="other-only", project_id=other, scope="runtime") + return {"key": key, "scoped": scoped, "project_id": pid} + + +@pytest.fixture +def client(store): + return TestClient(create_app()) + + +def _auth(key): + return {"Authorization": f"Bearer {key}"} + + +def _ndjson(*events) -> bytes: + return ("\n".join(json.dumps(e) for e in events) + "\n").encode() + + +# ---------- events ---------- + + +def test_valid_batch_is_stored(client, store): + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT, "app": "payments-api", "env": "prod"}, + content=_ndjson(SESSION, HIT), headers=_auth(store["key"])) + assert resp.status_code == 200, resp.text + assert resp.json()["sessions"] == 1 and resp.json()["hits"] == 1 + with RunRepository() as repo: + app, env = repo.connection.execute("SELECT app, env FROM runtime_sessions").fetchone() + assert (app, env) == ("payments-api", "prod") # filled from the sensor's SECFOO_APP / SECFOO_ENV + + +def test_replayed_batch_does_not_double_count(client, store): + body = gzip.compress(_ndjson(SESSION, HIT)) + for _ in range(2): + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, content=body, + headers={**_auth(store["key"]), "Content-Encoding": "gzip"}) + assert resp.status_code == 200 + with RunRepository() as repo: + assert [tuple(r) for r in repo.connection.execute("SELECT count FROM runtime_evidence")] == [(7,)] + + +def test_malformed_lines_are_reported_not_fatal(client, store): + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, + content=_ndjson(SESSION) + b"{broken\n" + _ndjson(HIT), headers=_auth(store["key"])) + assert resp.status_code == 200 + assert resp.json()["hits"] == 1 and resp.json()["skipped"][0]["line"] == 2 + + +# ---------- auth and access ---------- + + +def test_missing_or_invalid_key_is_rejected(client): + assert client.get("/api/runtime/v1/watchlist", params={"project": PROJECT}).status_code == 401 + bad = client.get("/api/runtime/v1/watchlist", params={"project": PROJECT}, headers=_auth("sfk_nope")) + assert bad.status_code == 401 + + +def test_key_without_access_to_the_project_is_rejected(client, store): + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, content=_ndjson(SESSION), + headers=_auth(store["scoped"])) + assert resp.status_code == 403 and "no access" in resp.json()["detail"] + + +def test_unknown_project_is_rejected_not_created(client, store): + resp = client.post("/api/runtime/v1/events", params={"project": "https://github.com/acme/never-scanned"}, + content=_ndjson(SESSION), headers=_auth(store["key"])) + assert resp.status_code == 404 and "run `secfoo run`" in resp.json()["detail"] + with RunRepository() as repo: + assert repo.find_project_id("https://github.com/acme/never-scanned") is None + + +def test_revoked_key_is_rejected(client, store): + with RunRepository() as repo: + record = next(k for k in keys.list_keys(repo.connection) if k.name == "all-projects") + assert keys.revoke_key(repo.connection, record.id) + resp = client.get("/api/runtime/v1/evidence", params={"project": PROJECT}, headers=_auth(store["key"])) + assert resp.status_code == 401 + + +def test_errors_never_echo_the_key(client, store): + resp = client.get("/api/runtime/v1/evidence", params={"project": "nope"}, headers=_auth(store["key"])) + assert store["key"] not in resp.text + + +def test_keys_are_stored_hashed(store): + with RunRepository() as repo: + dump = json.dumps([list(r) for r in repo.connection.execute("SELECT * FROM runtime_api_keys")]) + assert store["key"] not in dump and keys.hash_key(store["key"]) in dump + + +# ---------- hardening ---------- + + +def test_oversized_body_is_rejected(client, store, monkeypatch): + monkeypatch.setattr(runtime_api, "MAX_BODY_BYTES", 1000) + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, content=b"x" * 2000, + headers=_auth(store["key"])) + assert resp.status_code == 413 + + +def test_gzip_bomb_is_rejected(client, store, monkeypatch): + monkeypatch.setattr(runtime_api, "MAX_DECOMPRESSED_BYTES", 10_000) + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, + content=gzip.compress(b"\n" * 100_000), headers={**_auth(store["key"]), "Content-Encoding": "gzip"}) + assert resp.status_code == 413 + + +def test_too_many_lines_is_rejected(client, store, monkeypatch): + monkeypatch.setattr(runtime_api, "MAX_BATCH_LINES", 3) + resp = client.post("/api/runtime/v1/events", params={"project": PROJECT}, content=_ndjson(*[SESSION] * 5), + headers=_auth(store["key"])) + assert resp.status_code == 413 + + +def test_rate_limit_per_key(client, store, monkeypatch): + monkeypatch.setattr(runtime_api, "RATE_LIMIT_REQUESTS", 2) + codes = [client.get("/api/runtime/v1/evidence", params={"project": PROJECT}, headers=_auth(store["key"])).status_code + for _ in range(3)] + assert codes == [200, 200, 429] + + +def test_plain_http_on_a_non_loopback_bind_is_refused_unless_opted_in(store): + exposed = TestClient(create_app(bind_host="0.0.0.0")) # nosec B104 -- test of the guard itself + resp = exposed.get("/api/runtime/v1/evidence", params={"project": PROJECT}, headers=_auth(store["key"])) + assert resp.status_code == 403 and "plain HTTP" in resp.json()["detail"] + over_tls = TestClient(create_app(bind_host="0.0.0.0"), base_url="https://testserver") # nosec B104 + assert over_tls.get("/api/runtime/v1/evidence", params={"project": PROJECT}, + headers=_auth(store["key"])).status_code == 200 + opted_in = TestClient(create_app(bind_host="0.0.0.0", allow_insecure_runtime=True)) # nosec B104 + assert opted_in.get("/api/runtime/v1/evidence", params={"project": PROJECT}, + headers=_auth(store["key"])).status_code == 200 + + +# ---------- watchlist, heartbeat, evidence ---------- + + +def test_watchlist_with_etag(client, store): + first = client.get("/api/runtime/v1/watchlist", params={"project": PROJECT}, headers=_auth(store["key"])) + assert first.status_code == 200 and first.json()["entries"][0]["finding_fingerprint"] == FP + etag = first.headers["ETag"] + again = client.get("/api/runtime/v1/watchlist", params={"project": PROJECT}, + headers={**_auth(store["key"]), "If-None-Match": etag}) + assert again.status_code == 304 and again.content == b"" + + +def test_heartbeat_registers_the_sensor_and_sums_drops(client, store): + for dropped in (3, 4): + resp = client.post("/api/runtime/v1/heartbeat", params={"project": PROJECT}, headers=_auth(store["key"]), + json={"sensor_id": "sensor-a", "version": "0.4.0", "app": "payments-api", "env": "prod", + "instances": 2, "dropped": dropped}) + assert resp.status_code == 200 + with RunRepository() as repo: + row = repo.connection.execute("SELECT app, env, instances, dropped_total FROM runtime_sensors").fetchone() + assert tuple(row) == ("payments-api", "prod", 2, 7) + bad = client.post("/api/runtime/v1/heartbeat", params={"project": PROJECT}, headers=_auth(store["key"]), + json={"dropped": 1}) + assert bad.status_code == 422 + + +def test_evidence_endpoint_reads_runtime_state_per_finding(client, store): + client.post("/api/runtime/v1/events", params={"project": PROJECT}, content=_ndjson(SESSION, HIT), + headers=_auth(store["key"])) + resp = client.get("/api/runtime/v1/evidence", params={"project": PROJECT}, headers=_auth(store["key"])) + assert resp.status_code == 200 + [finding] = resp.json()["findings"] + assert finding["runtime_state"] == "reached" and finding["conflict"] is True + assert finding["routes"] == [{"route": "POST /upload", "auth": "anonymous", "count": 7}] + + +# ---------- sensor-side config ---------- + + +def test_sensor_reads_key_file_and_server(tmp_path): + key_file = tmp_path / "api-key" + key_file.write_text("sfk_secret\n", encoding="utf-8") + config = load_sensor_config({"SECFOO_API_KEY_FILE": str(key_file), "SECFOO_SERVER": "https://secfoo.internal", + "SECFOO_PROJECT": PROJECT, "SECFOO_APP": "payments-api", "SECFOO_ENV": "prod"}) + assert config.api_key == "sfk_secret" and config.server == "https://secfoo.internal" + assert config.url("events") == "https://secfoo.internal/api/runtime/v1/events" + assert "sfk_secret" not in repr(config) + + +def test_sensor_server_defaults_to_cloud_login(monkeypatch): + from secfoo.cloud import CloudConfig + + monkeypatch.setattr("secfoo.cloud.load_cloud_config", lambda: CloudConfig(api_key="x", portal_url="https://app.secfoo.com")) + config = load_sensor_config({"SECFOO_API_KEY": "sfk_k", "SECFOO_PROJECT": PROJECT}) + assert config.server == "https://app.secfoo.com" + + +@pytest.mark.parametrize("environ, message", [ + ({"SECFOO_SERVER": "https://s", "SECFOO_PROJECT": PROJECT}, "no API key"), + ({"SECFOO_API_KEY": "k", "SECFOO_SERVER": "https://s"}, "no project"), + ({"SECFOO_API_KEY": "k", "SECFOO_SERVER": "http://secfoo.example.com", "SECFOO_PROJECT": PROJECT}, "https://"), + ({"SECFOO_API_KEY_FILE": "/does/not/exist", "SECFOO_SERVER": "https://s", "SECFOO_PROJECT": PROJECT}, "can't read"), +]) +def test_sensor_refuses_to_start_without_explicit_config(environ, message, monkeypatch): + monkeypatch.setattr("secfoo.cloud.load_cloud_config", lambda: None) + with pytest.raises(SensorConfigError, match=message): + load_sensor_config(environ) + + +# ---------- CLI ---------- + + +def test_serve_key_create_list_revoke(store): + runner = CliRunner() + created = runner.invoke(cli_app, ["serve", "key", "create", "--name", "ci", "--project", "payments-api", + "--runtime-only"]) + assert created.exit_code == 0, created.output + key = created.stdout.strip().splitlines()[-1] + assert key.startswith("sfk_") + listed = runner.invoke(cli_app, ["serve", "key", "list"], env={"COLUMNS": "200"}) + assert "ci" in listed.output and key not in listed.output + with RunRepository() as repo: + record = next(k for k in keys.list_keys(repo.connection) if k.name == "ci") + assert (record.scope, record.project_id) == ("runtime", store["project_id"]) + assert runner.invoke(cli_app, ["serve", "key", "revoke", str(record.id)]).exit_code == 0 + assert runner.invoke(cli_app, ["serve", "key", "revoke", str(record.id)]).exit_code == 1