diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 387ae8e..920a2ed 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -350,6 +350,64 @@ score. CWE tags use `external/cwe/cwe-N`. Findings without a CWE still get a rul `--no-locations` removes the same structured location fields as JSONL and omits SARIF locations; free-text finding titles and descriptions are not redacted. +### Runtime evidence + +Runtime evidence records what a running app actually did with a vulnerable +dependency: whether it was loaded, whether a watched vulnerable function was +called, and on which route. It is **opt-in**: nothing produces it unless you +deploy a probe or sensor, or import a coverage file. Without it, SCA findings +work exactly as before. Design rules and roadmap: issue #51. + +```bash +secfoo runtime ingest events.ndjson --project payments # project id, name, or scanned path/URL +``` + +`ingest` stores the events against the project's SCA findings. It skips +malformed lines and hits for unknown findings, lists each one with its line +number, and still stores the rest; it exits `1` if anything was skipped. +Ingesting the same file twice updates rows instead of double-counting. + +**Event file:** NDJSON, one JSON object per line. A `session` line, then any +number of `load` and `hit` lines that reference it by `session` id. A later +file may add hits to a session stored earlier. The JSON Schema is +[`src/secfoo/runtime/evidence.schema.json`](../src/secfoo/runtime/evidence.schema.json). + +```json +{"type":"session","id":"","language":"python","runtime":"3.12.4","source":"sensor","started_at":"2026-11-08T10:00:00Z","ended_at":"2026-11-08T16:00:00Z","app":"payments-api","env":"prod","instance":"pod-1","sensor_version":"0.4.0","app_functions_total":1840,"app_functions_executed":1362} +{"type":"load","session":"","ecosystem":"PyPI","package":"pyyaml","version":"5.3","path":"/app/.venv/lib/python3.12/site-packages/yaml","first_seen":"2026-11-08T10:00:01Z"} +{"type":"hit","session":"","finding_fingerprint":"","symbol":"yaml.load","route":"POST /upload","auth":"anonymous","count":312,"first_seen":"2026-11-08T10:05:00Z","last_seen":"2026-11-08T15:59:00Z","entry_frame":"app/routes/upload.py:42","stack_sample":["app/routes/upload.py:42","app/services/parse.py:17"]} +``` + +| Line | Required | Optional | +|---|---|---| +| `session` | `id` (producer's own UUID), `language` (`python` `node` `jvm` `dotnet` `go` `rust` `ruby` `php` `native`), `source` (`sensor` or `coverage`), `started_at` | `runtime`, `ended_at`, `app`, `env`, `instance`, `sensor_version`, `app_functions_total`, `app_functions_executed` | +| `load` | `session`, `ecosystem` (OSV name), `package`, `version` | `path`, `first_seen` | +| `hit` | `session`, `finding_fingerprint`, `symbol`, `count` (≥ 1) | `route`, `auth` (`anonymous` or `authenticated`), `first_seen`, `last_seen`, `entry_frame`, `stack_sample` | + +- Timestamps are ISO 8601. +- A hit's `count` is the session's **running total** for its (fingerprint, + symbol, route), so a producer can flush the same session repeatedly and + the stored value is replaced, never summed. +- `route` is a template (`POST /upload`), `job:`, `startup` or + `unknown`, never a concrete URL. Values with a scheme (`://`), query string + or fragment are rejected. +- `entry_frame` and `stack_sample` (at most 50 frames) are `file:line` frames + only. Producers drop frames outside the app (site-packages, node_modules) + except the vulnerable function itself. +- Only the fields above are stored. Anything else on a line is ignored and + never written, so argument values, request bodies and headers can't reach + the database even if a producer sends them. +- `app_functions_total` / `app_functions_executed` record how much of the + app the observed traffic covered. "Not observed" is always shown with this + coverage, never as "not reachable". + +**Storage** (in `schema.sql`, added automatically to existing stores): +`runtime_sessions` (one row per session), `runtime_evidence` (per session, +finding, symbol and route, with a foreign key to `sca_findings`) and +`runtime_packages` (loaded inventory per session). Validation and storage +live in `secfoo/runtime/evidence.py` (`ingest_events`), which the live +ingest API will reuse. + ## 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 7262435..1c33579 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -45,6 +45,8 @@ app.add_typer(mcp_app, name="mcp") app.add_typer(memory_app, name="memory") app.add_typer(findings_app, name="findings") +runtime_app = typer.Typer(name="runtime", help="Runtime evidence for SCA findings (opt-in; see docs/REFERENCE.md).") +app.add_typer(runtime_app, name="runtime") app.add_typer(config_app, name="config") app.add_typer(assessment_app, name="assessment") app.add_typer(exception_app, name="exception") @@ -1513,6 +1515,47 @@ def findings_export( err_console.print(f"Wrote {len(records)} finding(s) to {output}.") +@runtime_app.command(name="ingest") +def runtime_ingest( + events: Annotated[ + Path, typer.Argument(exists=True, dir_okay=False, readable=True, help="NDJSON event file.") + ], + project: str = typer.Option(..., "--project", help="Project id, name, or path/URL it was scanned as."), +) -> None: + """Store runtime evidence (session / load / hit lines) against a project's SCA findings. + + Malformed lines and hits for unknown findings are skipped and listed; + the rest of the file is still stored. Ingesting the same file again + updates rows rather than double-counting. Exits 1 if any line was skipped. + """ + from secfoo.runtime.evidence import ingest_file + + with RunRepository() as repo: + project_id = repo.resolve_project_ref(project) + if project_id is None: + err_console.print( + f"[red]Error:[/] no single project matches {project!r}. " + "Use the project id from `secfoo list`, its name, or the path/URL it was scanned as." + ) + raise typer.Exit(code=1) + report = ingest_file(repo.connection, project_id, events) + console.print( + f"Stored {report.sessions} session(s), {report.loads} package load(s), " + f"{report.hits} hit(s) from {events}." + ) + if report.unknown_fingerprints: + err_console.print( + f"[yellow]Warning:[/] {len(report.unknown_fingerprints)} finding fingerprint(s) aren't SCA findings " + "of this project, so their hits were skipped. Was the watchlist built for another project?" + ) + for number, reason in report.skipped[:20]: + err_console.print(f" line {number}: {reason}") + if len(report.skipped) > 20: + err_console.print(f" ... and {len(report.skipped) - 20} more skipped line(s)") + if report.skipped: + raise typer.Exit(code=1) + + MERMAID_CDN_URL = "https://cdn.jsdelivr.net/npm/mermaid@11/dist/mermaid.min.js" diff --git a/src/secfoo/runtime/__init__.py b/src/secfoo/runtime/__init__.py new file mode 100644 index 0000000..a43d5ed --- /dev/null +++ b/src/secfoo/runtime/__init__.py @@ -0,0 +1,5 @@ +"""Runtime evidence for SCA findings (#51). + +`evidence` holds the shared event format (#53) that probes, the coverage +importer and the live sensor all write, plus its validation and storage. +""" diff --git a/src/secfoo/runtime/evidence.py b/src/secfoo/runtime/evidence.py new file mode 100644 index 0000000..49cee00 --- /dev/null +++ b/src/secfoo/runtime/evidence.py @@ -0,0 +1,338 @@ +"""Runtime evidence events (#53): validate NDJSON lines and store them. + +This is the one contract that every producer writes -- the Python and Node +probes (#55, #56), the coverage importer (#54) and the live sensor via the +ingest API (#66/#67). The file format and the live API share it, so both +call `ingest_events()`. + +An event file is NDJSON. Each line is one object with a `type`: + + {"type":"session","id":"","language":"python","source":"sensor","started_at":"..."} + {"type":"load","session":"","ecosystem":"PyPI","package":"pyyaml","version":"5.3"} + {"type":"hit","session":"","finding_fingerprint":"...","symbol":"yaml.load","count":312} + +The full field list is in docs/REFERENCE.md ("Runtime evidence") and in +evidence.schema.json next to this module. + +Ingest rules: + +- A malformed line is skipped and reported with its line number. It never + aborts the rest of the file. +- A `hit` whose `finding_fingerprint` isn't an SCA finding of the project is + skipped with a warning (it can't attach to anything). +- Idempotent: a session is keyed by its own `id`, and a hit by (session, + fingerprint, symbol, route). A hit's `count` is that session's running + total for the key, so a re-ingest -- or a later flush of the same session + -- replaces the row instead of adding to it. +- Privacy (#51 rule 3): only the fields below are stored. Unknown fields + are ignored, never persisted. Routes that look like concrete URLs (a + scheme or query string) and stack frames that aren't `file:line` are + rejected. +""" + +from __future__ import annotations + +import json +import re +import sqlite3 +from collections.abc import Iterable, Iterator +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path + +LANGUAGES = frozenset({"python", "node", "jvm", "dotnet", "go", "rust", "ruby", "php", "native"}) +SOURCES = frozenset({"sensor", "coverage"}) +AUTH_STATES = frozenset({"anonymous", "authenticated"}) + +MAX_LINE_BYTES = 64 * 1024 +MAX_TEXT = 1024 +MAX_STACK_FRAMES = 50 + +_FRAME_RE = re.compile(rf"^\S.{{0,{MAX_TEXT - 8}}}:\d+$") +_CONCRETE_URL_RE = re.compile(r"://|[?#]") + + +class EvidenceError(ValueError): + """One line failed validation. The message says which field and why.""" + + +@dataclass +class IngestReport: + sessions: int = 0 + loads: int = 0 + hits: int = 0 + skipped: list[tuple[int, str]] = field(default_factory=list) # (line number, reason) + unknown_fingerprints: set[str] = field(default_factory=set) + + @property + def stored(self) -> int: + return self.sessions + self.loads + self.hits + + +# ---------- validation ---------- + + +def _text(event: dict, key: str, *, required: bool = False) -> str | None: + value = event.get(key) + if value is None: + if required: + raise EvidenceError(f"missing required field {key!r}") + return None + if not isinstance(value, str) or not value.strip(): + raise EvidenceError(f"{key!r} must be a non-empty string") + if len(value) > MAX_TEXT: + raise EvidenceError(f"{key!r} is longer than {MAX_TEXT} characters") + return value + + +def _timestamp(event: dict, key: str, *, required: bool = False) -> str | None: + value = _text(event, key, required=required) + if value is None: + return None + try: + # Python 3.10's fromisoformat doesn't accept a trailing "Z". + datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value) + except ValueError: + raise EvidenceError(f"{key!r} must be an ISO 8601 timestamp, got {value!r}") from None + return value + + +def _count(event: dict, key: str, *, required: bool = False, minimum: int = 0) -> int | None: + value = event.get(key) + if value is None: + if required: + raise EvidenceError(f"missing required field {key!r}") + return None + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + raise EvidenceError(f"{key!r} must be an integer >= {minimum}") + return value + + +def _frame(value: object, key: str) -> str: + if not isinstance(value, str) or not _FRAME_RE.match(value): + raise EvidenceError(f"{key!r} frames must be 'file:line', never variables or arguments") + return value + + +def validate_event(event: object) -> dict: + """The stored fields of one event, or EvidenceError. Unknown fields are + dropped here, so nothing outside the contract can reach the database.""" + if not isinstance(event, dict): + raise EvidenceError("each line must be a JSON object") + kind = event.get("type") + if kind == "session": + language = _text(event, "language", required=True) + if language not in LANGUAGES: + raise EvidenceError(f"'language' must be one of {', '.join(sorted(LANGUAGES))}") + source = _text(event, "source", required=True) + if source not in SOURCES: + raise EvidenceError("'source' must be 'sensor' or 'coverage'") + total = _count(event, "app_functions_total") + executed = _count(event, "app_functions_executed") + if total is not None and executed is not None and executed > total: + raise EvidenceError("'app_functions_executed' can't exceed 'app_functions_total'") + return { + "type": "session", + "id": _text(event, "id", required=True), + "language": language, + "runtime": _text(event, "runtime"), + "source": source, + "app": _text(event, "app"), + "env": _text(event, "env"), + "instance": _text(event, "instance"), + "sensor_version": _text(event, "sensor_version"), + "started_at": _timestamp(event, "started_at", required=True), + "ended_at": _timestamp(event, "ended_at"), + "app_functions_total": total, + "app_functions_executed": executed, + } + if kind == "load": + return { + "type": "load", + "session": _text(event, "session", required=True), + "ecosystem": _text(event, "ecosystem", required=True), + "package": _text(event, "package", required=True), + "version": _text(event, "version", required=True), + "path": _text(event, "path"), + "first_seen": _timestamp(event, "first_seen"), + } + if kind == "hit": + route = _text(event, "route") + if route is not None and _CONCRETE_URL_RE.search(route): + raise EvidenceError("'route' must be a template like 'POST /upload', not a concrete URL") + auth = _text(event, "auth") + if auth is not None and auth not in AUTH_STATES: + raise EvidenceError("'auth' must be 'anonymous' or 'authenticated'") + stack = event.get("stack_sample") + if stack is not None: + if not isinstance(stack, list) or len(stack) > MAX_STACK_FRAMES: + raise EvidenceError(f"'stack_sample' must be a list of at most {MAX_STACK_FRAMES} frames") + stack = [_frame(frame, "stack_sample") for frame in stack] + entry = event.get("entry_frame") + return { + "type": "hit", + "session": _text(event, "session", required=True), + "finding_fingerprint": _text(event, "finding_fingerprint", required=True), + "symbol": _text(event, "symbol", required=True), + "route": route or "", + "auth": auth, + "count": _count(event, "count", required=True, minimum=1), + "first_seen": _timestamp(event, "first_seen"), + "last_seen": _timestamp(event, "last_seen"), + "entry_frame": None if entry is None else _frame(entry, "entry_frame"), + "stack_sample": stack, + } + raise EvidenceError("'type' must be 'session', 'load' or 'hit'") + + +def parse_lines(lines: Iterable[str]) -> Iterator[tuple[int, dict | None, str | None]]: + """(line number, validated event or None, error or None) per non-blank line.""" + for number, raw in enumerate(lines, start=1): + if not raw.strip(): + continue + if len(raw.encode("utf-8", errors="replace")) > MAX_LINE_BYTES: + yield number, None, f"line is longer than {MAX_LINE_BYTES} bytes" + continue + try: + yield number, validate_event(json.loads(raw)), None + except json.JSONDecodeError as exc: + yield number, None, f"not valid JSON ({exc.msg})" + except EvidenceError as exc: + yield number, None, str(exc) + + +# ---------- storage ---------- + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _session_id(conn: sqlite3.Connection, project_id: int, session_uuid: str) -> int | None: + row = conn.execute( + "SELECT id FROM runtime_sessions WHERE project_id = ? AND session_uuid = ?", (project_id, session_uuid) + ).fetchone() + return row[0] if row else None + + +def _store_session(conn: sqlite3.Connection, project_id: int, event: dict) -> int: + conn.execute( + """ + INSERT INTO runtime_sessions ( + project_id, session_uuid, language, runtime, source, app, env, instance, sensor_version, + started_at, ended_at, app_functions_total, app_functions_executed, ingested_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (project_id, session_uuid) DO UPDATE SET + language = excluded.language, runtime = excluded.runtime, source = excluded.source, + app = excluded.app, env = excluded.env, instance = excluded.instance, + sensor_version = excluded.sensor_version, started_at = excluded.started_at, + ended_at = excluded.ended_at, app_functions_total = excluded.app_functions_total, + app_functions_executed = excluded.app_functions_executed, ingested_at = excluded.ingested_at + """, + ( + project_id, event["id"], event["language"], event["runtime"], event["source"], event["app"], + event["env"], event["instance"], event["sensor_version"], event["started_at"], event["ended_at"], + event["app_functions_total"], event["app_functions_executed"], _now(), + ), + ) + session_id = _session_id(conn, project_id, event["id"]) + assert session_id is not None # nosec B101 -- just upserted + return session_id + + +def _store_load(conn: sqlite3.Connection, project_id: int, session_id: int, event: dict) -> None: + conn.execute( + """ + INSERT INTO runtime_packages (project_id, session_id, ecosystem, package, version, path, first_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (session_id, ecosystem, package, version) DO UPDATE SET + path = excluded.path, + first_seen = COALESCE(MIN(runtime_packages.first_seen, excluded.first_seen), + runtime_packages.first_seen, excluded.first_seen) + """, + (project_id, session_id, event["ecosystem"], event["package"], event["version"], event["path"], + event["first_seen"]), + ) + + +def _store_hit(conn: sqlite3.Connection, project_id: int, session_id: int, event: dict) -> None: + conn.execute( + """ + INSERT INTO runtime_evidence ( + project_id, session_id, finding_fingerprint, symbol, route, auth, count, + first_seen, last_seen, entry_frame, stack_sample + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (session_id, finding_fingerprint, symbol, route) DO UPDATE SET + count = excluded.count, + auth = COALESCE(excluded.auth, runtime_evidence.auth), + first_seen = COALESCE(MIN(runtime_evidence.first_seen, excluded.first_seen), + runtime_evidence.first_seen, excluded.first_seen), + last_seen = COALESCE(MAX(runtime_evidence.last_seen, excluded.last_seen), + runtime_evidence.last_seen, excluded.last_seen), + entry_frame = COALESCE(excluded.entry_frame, runtime_evidence.entry_frame), + stack_sample = COALESCE(excluded.stack_sample, runtime_evidence.stack_sample) + """, + ( + project_id, session_id, event["finding_fingerprint"], event["symbol"], event["route"], event["auth"], + event["count"], event["first_seen"], event["last_seen"], event["entry_frame"], + None if event["stack_sample"] is None else json.dumps(event["stack_sample"]), + ), + ) + + +def ingest_events(conn: sqlite3.Connection, project_id: int, lines: Iterable[str]) -> IngestReport: + """Validate and store every line, in one transaction. Bad lines are + skipped and listed in the report; they never stop the others.""" + report = IngestReport() + sessions: dict[str, int] = {} + known_fingerprints: dict[str, bool] = {} + + def session_for(uuid: str) -> int | None: + if uuid not in sessions: + found = _session_id(conn, project_id, uuid) # from an earlier ingest + if found is None: + return None + sessions[uuid] = found + return sessions[uuid] + + def is_finding(fingerprint: str) -> bool: + if fingerprint not in known_fingerprints: + known_fingerprints[fingerprint] = conn.execute( + "SELECT 1 FROM sca_findings WHERE project_id = ? AND fingerprint = ?", (project_id, fingerprint) + ).fetchone() is not None + return known_fingerprints[fingerprint] + + try: + for number, event, error in parse_lines(lines): + if event is None: + report.skipped.append((number, error or "invalid line")) + continue + if event["type"] == "session": + sessions[event["id"]] = _store_session(conn, project_id, event) + report.sessions += 1 + continue + session_id = session_for(event["session"]) + if session_id is None: + report.skipped.append((number, f"unknown session {event['session']!r} (no session line for it)")) + continue + if event["type"] == "load": + _store_load(conn, project_id, session_id, event) + report.loads += 1 + elif is_finding(event["finding_fingerprint"]): + _store_hit(conn, project_id, session_id, event) + report.hits += 1 + else: + report.unknown_fingerprints.add(event["finding_fingerprint"]) + report.skipped.append( + (number, f"finding_fingerprint {event['finding_fingerprint']!r} is not an SCA finding of this project") + ) + conn.commit() + except BaseException: + conn.rollback() + raise + return report + + +def ingest_file(conn: sqlite3.Connection, project_id: int, path: Path) -> IngestReport: + with path.open(encoding="utf-8", errors="replace") as handle: + return ingest_events(conn, project_id, handle) diff --git a/src/secfoo/runtime/evidence.schema.json b/src/secfoo/runtime/evidence.schema.json new file mode 100644 index 0000000..6e9aa5b --- /dev/null +++ b/src/secfoo/runtime/evidence.schema.json @@ -0,0 +1,74 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/secfoo-com/secfoo/blob/main/src/secfoo/runtime/evidence.schema.json", + "title": "Secfoo runtime evidence event (one NDJSON line)", + "description": "One line of a runtime evidence file (#53). secfoo/runtime/evidence.py:validate_event enforces the same rules; tests/test_runtime/test_evidence.py checks the two agree. Unknown fields are allowed for forward compatibility but are never stored.", + "oneOf": [ + {"$ref": "#/$defs/session"}, + {"$ref": "#/$defs/load"}, + {"$ref": "#/$defs/hit"} + ], + "$defs": { + "text": {"type": "string", "minLength": 1, "maxLength": 1024, "pattern": "\\S"}, + "timestamp": {"type": "string", "minLength": 1, "maxLength": 1024, "format": "date-time"}, + "frame": { + "description": "file:line only, never local variables or arguments.", + "type": "string", + "pattern": "^\\S.*:\\d+$", + "maxLength": 1024 + }, + "session": { + "type": "object", + "required": ["type", "id", "language", "source", "started_at"], + "properties": { + "type": {"const": "session"}, + "id": {"$ref": "#/$defs/text", "description": "The producer's own session id (a UUID). Re-ingesting the same id updates, never duplicates."}, + "language": {"enum": ["python", "node", "jvm", "dotnet", "go", "rust", "ruby", "php", "native"]}, + "runtime": {"$ref": "#/$defs/text", "description": "Runtime version, e.g. 3.12.4."}, + "source": {"enum": ["sensor", "coverage"]}, + "app": {"$ref": "#/$defs/text"}, + "env": {"$ref": "#/$defs/text"}, + "instance": {"$ref": "#/$defs/text", "description": "Host or pod name."}, + "sensor_version": {"$ref": "#/$defs/text"}, + "started_at": {"$ref": "#/$defs/timestamp"}, + "ended_at": {"$ref": "#/$defs/timestamp"}, + "app_functions_total": {"type": "integer", "minimum": 0}, + "app_functions_executed": {"type": "integer", "minimum": 0} + } + }, + "load": { + "type": "object", + "required": ["type", "session", "ecosystem", "package", "version"], + "properties": { + "type": {"const": "load"}, + "session": {"$ref": "#/$defs/text"}, + "ecosystem": {"$ref": "#/$defs/text", "description": "OSV ecosystem name: PyPI, npm, Maven, NuGet, Go, crates.io, RubyGems, Packagist."}, + "package": {"$ref": "#/$defs/text"}, + "version": {"$ref": "#/$defs/text"}, + "path": {"$ref": "#/$defs/text"}, + "first_seen": {"$ref": "#/$defs/timestamp"} + } + }, + "hit": { + "type": "object", + "required": ["type", "session", "finding_fingerprint", "symbol", "count"], + "properties": { + "type": {"const": "hit"}, + "session": {"$ref": "#/$defs/text"}, + "finding_fingerprint": {"$ref": "#/$defs/text", "description": "sca_findings.fingerprint of the project."}, + "symbol": {"$ref": "#/$defs/text", "description": "Watched symbol in the #52 notation, or the package name for package-granularity entries."}, + "route": { + "description": "Route template ('POST /upload'), job:, startup or unknown. Never a concrete URL.", + "type": "string", "minLength": 1, "maxLength": 1024, "pattern": "^[^?#]*$", + "not": {"pattern": "://"} + }, + "auth": {"enum": ["anonymous", "authenticated"]}, + "count": {"type": "integer", "minimum": 1, "description": "The session's running total for this (fingerprint, symbol, route)."}, + "first_seen": {"$ref": "#/$defs/timestamp"}, + "last_seen": {"$ref": "#/$defs/timestamp"}, + "entry_frame": {"$ref": "#/$defs/frame"}, + "stack_sample": {"type": "array", "maxItems": 50, "items": {"$ref": "#/$defs/frame"}} + } + } + } +} diff --git a/src/secfoo/storage/repository.py b/src/secfoo/storage/repository.py index aa08b5a..d428f80 100644 --- a/src/secfoo/storage/repository.py +++ b/src/secfoo/storage/repository.py @@ -363,6 +363,28 @@ def complete_run( ) self._conn.commit() + def resolve_project_ref(self, ref: str) -> int | None: + """A project id from what a user would type for `--project`: its + numeric id, its identifier (path / repository URL), or its display + name. None when nothing matches, or when a display name is shared + by several projects (the caller should ask for the id instead).""" + ref = ref.strip() + if ref.isdigit(): + row = self._conn.execute("SELECT id FROM projects WHERE id = ?", (int(ref),)).fetchone() + if row: + return row["id"] + found = self.find_project_id(ref) + if found is not None: + return found + rows = self._conn.execute("SELECT id FROM projects WHERE display_name = ?", (ref,)).fetchall() + return rows[0]["id"] if len(rows) == 1 else None + + @property + def connection(self) -> sqlite3.Connection: + """The underlying connection, for modules that own a table group's + writes end to end (runtime/evidence.py) and need one transaction.""" + return self._conn + def find_project_id(self, identifier: str) -> int | None: """The id of an existing project, without creating one (unlike upsert_project). Used to check for a repeat scan before running.""" diff --git a/src/secfoo/storage/schema.sql b/src/secfoo/storage/schema.sql index cb9587b..9156e81 100644 --- a/src/secfoo/storage/schema.sql +++ b/src/secfoo/storage/schema.sql @@ -302,3 +302,71 @@ CREATE TABLE IF NOT EXISTS secret_findings ( ); CREATE INDEX IF NOT EXISTS idx_secret_findings_project_status ON secret_findings(project_id, status); + +-- Runtime evidence for SCA findings (#51 / #53): what a probe, sensor or +-- imported coverage file observed in a running app. Written only by +-- runtime/evidence.py (`secfoo runtime ingest`, later the live API in #66), +-- which validates every line first. Never holds argument values, request +-- bodies, concrete URLs or user identity -- see the design rules in #51. +-- +-- One row per observation session (a process run, a coverage file). +-- `session_uuid` is the producer's own id, so re-ingesting the same file +-- updates rows instead of adding duplicates. +CREATE TABLE IF NOT EXISTS runtime_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id INTEGER NOT NULL REFERENCES projects(id), + session_uuid TEXT NOT NULL, + language TEXT NOT NULL, -- python / node / jvm / dotnet / go / rust / ruby / php / native + runtime TEXT, -- runtime version, e.g. "3.12.4" + source TEXT NOT NULL CHECK (source IN ('sensor','coverage')), + app TEXT, + env TEXT, + instance TEXT, + sensor_version TEXT, + started_at TEXT NOT NULL, + ended_at TEXT, + app_functions_total INTEGER, -- with _executed, the coverage behind any "not observed" + app_functions_executed INTEGER, + ingested_at TEXT NOT NULL, + UNIQUE (project_id, session_uuid) +); + +-- A watched vulnerable symbol (or package, for package-granularity +-- watchlist entries) seen during a session, aggregated per route. `route` +-- is a template ("POST /upload") or job: / startup / unknown, never +-- a concrete URL; '' when the producer has no request context. +CREATE TABLE IF NOT EXISTS runtime_evidence ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id INTEGER NOT NULL REFERENCES projects(id), + session_id INTEGER NOT NULL REFERENCES runtime_sessions(id), + finding_fingerprint TEXT NOT NULL, + symbol TEXT NOT NULL, + route TEXT NOT NULL DEFAULT '', + auth TEXT CHECK (auth IS NULL OR auth IN ('anonymous','authenticated')), + count INTEGER NOT NULL CHECK (count > 0), + first_seen TEXT, + last_seen TEXT, + entry_frame TEXT, -- file:line, app frames only + stack_sample TEXT, -- JSON array of file:line frames, never locals or arguments + FOREIGN KEY (project_id, finding_fingerprint) REFERENCES sca_findings(project_id, fingerprint), + UNIQUE (session_id, finding_fingerprint, symbol, route) +); + +CREATE INDEX IF NOT EXISTS idx_runtime_evidence_finding ON runtime_evidence(project_id, finding_fingerprint); + +-- Packages loaded in a running process (#68's inventory), per session, so +-- "Loaded" can be shown and version drift against the scanned lockfile +-- detected. +CREATE TABLE IF NOT EXISTS runtime_packages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id INTEGER NOT NULL REFERENCES projects(id), + session_id INTEGER NOT NULL REFERENCES runtime_sessions(id), + ecosystem TEXT NOT NULL, + package TEXT NOT NULL, + version TEXT NOT NULL, + path TEXT, + first_seen TEXT, + UNIQUE (session_id, ecosystem, package, version) +); + +CREATE INDEX IF NOT EXISTS idx_runtime_packages_project ON runtime_packages(project_id, ecosystem, package); diff --git a/tests/test_runtime/test_evidence.py b/tests/test_runtime/test_evidence.py new file mode 100644 index 0000000..4cb7601 --- /dev/null +++ b/tests/test_runtime/test_evidence.py @@ -0,0 +1,291 @@ +"""#53: runtime evidence format, storage and `secfoo runtime ingest`.""" +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from secfoo.cli import app +from secfoo.runtime import evidence +from secfoo.runtime.evidence import ( + EvidenceError, + ingest_events, + ingest_file, + validate_event, +) +from secfoo.storage.repository import RunRepository + +FP = "sca:pypi:pyyaml:GHSA-xxxx" +OTHER_FP = "sca:npm:lodash:GHSA-yyyy" +SESSION = "8f8a1c2e-0000-4000-8000-000000000001" + +SESSION_LINE = { + "type": "session", "id": SESSION, "language": "python", "runtime": "3.12.4", "source": "sensor", + "started_at": "2026-11-08T10:00:00Z", "ended_at": "2026-11-08T16:00:00Z", "app": "payments-api", + "env": "prod", "instance": "pod-1", "sensor_version": "0.4.0", + "app_functions_total": 1840, "app_functions_executed": 1362, +} +LOAD_LINE = { + "type": "load", "session": SESSION, "ecosystem": "PyPI", "package": "pyyaml", "version": "5.3", + "path": "/app/.venv/lib/python3.12/site-packages/yaml", "first_seen": "2026-11-08T10:00:01Z", +} +HIT_LINE = { + "type": "hit", "session": SESSION, "finding_fingerprint": FP, "symbol": "yaml.load", "route": "POST /upload", + "auth": "anonymous", "count": 312, "first_seen": "2026-11-08T10:05:00Z", "last_seen": "2026-11-08T15:59:00Z", + "entry_frame": "app/routes/upload.py:42", "stack_sample": ["app/routes/upload.py:42", "app/services/parse.py:17"], +} + + +def _lines(*events) -> list[str]: + return [json.dumps(e) for e in events] + + +@pytest.fixture +def repo(tmp_path): + with RunRepository(db_path=tmp_path / "db.sqlite") as repo: + yield repo + + +@pytest.fixture +def project_id(repo): + pid = repo.upsert_project("https://github.com/acme/payments", "payments", "github") + run_uuid = repo.create_run( + project_id=pid, skill_id="sca-reachability", skill_name="SCA", agent_id="claude", confluence_urls=[] + ) + run_id = repo.get_run(run_uuid).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") + return pid + + +def _rows(repo, table): + return [dict(r) for r in repo.connection.execute(f"SELECT * FROM {table}").fetchall()] # nosec B608 -- test + + +# ---------- a valid file ---------- + + +def test_valid_file_stores_session_load_and_hit(repo, project_id): + report = ingest_events(repo.connection, project_id, _lines(SESSION_LINE, LOAD_LINE, HIT_LINE)) + + assert (report.sessions, report.loads, report.hits, report.skipped) == (1, 1, 1, []) + [session] = _rows(repo, "runtime_sessions") + assert session["session_uuid"] == SESSION and session["source"] == "sensor" + assert (session["app_functions_total"], session["app_functions_executed"]) == (1840, 1362) + [package] = _rows(repo, "runtime_packages") + assert (package["ecosystem"], package["package"], package["version"]) == ("PyPI", "pyyaml", "5.3") + [hit] = _rows(repo, "runtime_evidence") + assert hit["finding_fingerprint"] == FP and hit["route"] == "POST /upload" and hit["count"] == 312 + assert json.loads(hit["stack_sample"]) == HIT_LINE["stack_sample"] + + +def test_optional_fields_can_be_left_out(repo, project_id): + minimal = _lines( + {"type": "session", "id": "s1", "language": "node", "source": "coverage", "started_at": "2026-11-08T10:00:00"}, + {"type": "hit", "session": "s1", "finding_fingerprint": FP, "symbol": "pyyaml", "count": 1}, + ) + report = ingest_events(repo.connection, project_id, minimal) + assert report.skipped == [] and report.hits == 1 + assert _rows(repo, "runtime_evidence")[0]["route"] == "" # no request context + + +# ---------- malformed lines: skip and report, never abort ---------- + + +def test_malformed_lines_are_skipped_and_reported_but_the_rest_is_stored(repo, project_id): + lines = [ + json.dumps(SESSION_LINE), + "{not json", + json.dumps({**HIT_LINE, "count": 0}), + json.dumps({"type": "bogus"}), + "", # blank lines are ignored, not errors + json.dumps(LOAD_LINE), + json.dumps(HIT_LINE), + ] + report = ingest_events(repo.connection, project_id, lines) + + assert (report.sessions, report.loads, report.hits) == (1, 1, 1) + assert [n for n, _ in report.skipped] == [2, 3, 4] + reasons = dict(report.skipped) + assert "not valid JSON" in reasons[2] + assert "'count'" in reasons[3] + assert "'type'" in reasons[4] + + +def test_hit_for_an_unknown_fingerprint_is_skipped_with_a_warning(repo, project_id): + report = ingest_events( + repo.connection, project_id, _lines(SESSION_LINE, {**HIT_LINE, "finding_fingerprint": OTHER_FP}) + ) + assert report.hits == 0 and report.unknown_fingerprints == {OTHER_FP} + assert "not an SCA finding of this project" in report.skipped[0][1] + + +def test_hit_before_any_session_line_is_skipped(repo, project_id): + report = ingest_events(repo.connection, project_id, _lines(HIT_LINE)) + assert report.hits == 0 and "unknown session" in report.skipped[0][1] + + +def test_findings_of_another_project_do_not_count(repo, project_id): + other = repo.upsert_project("/elsewhere", "elsewhere", "local") + report = ingest_events(repo.connection, other, _lines({**SESSION_LINE}, HIT_LINE)) + assert report.hits == 0 and report.unknown_fingerprints == {FP} + + +# ---------- idempotent re-ingest ---------- + + +def test_reingesting_the_same_file_does_not_double_count(repo, project_id): + lines = _lines(SESSION_LINE, LOAD_LINE, HIT_LINE) + ingest_events(repo.connection, project_id, lines) + ingest_events(repo.connection, project_id, lines) + + assert len(_rows(repo, "runtime_sessions")) == 1 + assert len(_rows(repo, "runtime_packages")) == 1 + [hit] = _rows(repo, "runtime_evidence") + assert hit["count"] == 312 + + +def test_a_later_flush_of_the_same_session_replaces_the_running_total(repo, project_id): + ingest_events(repo.connection, project_id, _lines(SESSION_LINE, HIT_LINE)) + later = {**HIT_LINE, "count": 400, "last_seen": "2026-11-08T17:00:00Z", "first_seen": "2026-11-08T11:00:00Z"} + # The session line isn't repeated: hits attach to a session stored earlier. + ingest_events(repo.connection, project_id, _lines(later)) + + [hit] = _rows(repo, "runtime_evidence") + assert hit["count"] == 400 + assert hit["first_seen"] == "2026-11-08T10:05:00Z" # earliest kept + assert hit["last_seen"] == "2026-11-08T17:00:00Z" # latest kept + + +def test_each_route_is_its_own_row(repo, project_id): + ingest_events(repo.connection, project_id, _lines(SESSION_LINE, HIT_LINE, {**HIT_LINE, "route": "GET /export"})) + assert sorted(r["route"] for r in _rows(repo, "runtime_evidence")) == ["GET /export", "POST /upload"] + + +# ---------- privacy (#51 rule 3) ---------- + + +@pytest.mark.parametrize("route", ["https://shop.example/upload", "POST /upload?token=abc", "GET /a#frag"]) +def test_concrete_urls_are_rejected_as_routes(route): + with pytest.raises(EvidenceError, match="template"): + validate_event({**HIT_LINE, "route": route}) + + +@pytest.mark.parametrize("stack", [["password=hunter2"], [{"frame": "a.py:1"}], ["a.py:1"] * 51, "a.py:1"]) +def test_stack_samples_must_be_file_line_frames(stack): + with pytest.raises(EvidenceError, match="stack_sample"): + validate_event({**HIT_LINE, "stack_sample": stack}) + + +def test_unknown_fields_are_never_stored(repo, project_id): + sneaky = {**HIT_LINE, "args": ["s3cr3t"], "request_body": "card=4111"} + ingest_events(repo.connection, project_id, _lines(SESSION_LINE, sneaky)) + dumped = json.dumps(_rows(repo, "runtime_evidence")) + assert "s3cr3t" not in dumped and "4111" not in dumped + + +@pytest.mark.parametrize( + "event, field", + [ + ({**SESSION_LINE, "language": "cobol"}, "language"), + ({**SESSION_LINE, "source": "guess"}, "source"), + ({**SESSION_LINE, "started_at": "yesterday"}, "started_at"), + ({**SESSION_LINE, "app_functions_executed": 2000}, "app_functions_executed"), + ({**HIT_LINE, "auth": "admin"}, "auth"), + ({**HIT_LINE, "count": True}, "count"), + ({k: v for k, v in LOAD_LINE.items() if k != "version"}, "version"), + ], +) +def test_invalid_fields_name_the_field(event, field): + with pytest.raises(EvidenceError, match=field): + validate_event(event) + + +# ---------- the JSON Schema says the same thing as the validator ---------- + + +def test_json_schema_agrees_with_the_validator(): + jsonschema = pytest.importorskip("jsonschema") + schema = json.loads((Path(evidence.__file__).parent / "evidence.schema.json").read_text(encoding="utf-8")) + jsonschema.Draft202012Validator.check_schema(schema) + validator = jsonschema.Draft202012Validator(schema) + for good in (SESSION_LINE, LOAD_LINE, HIT_LINE): + validate_event(good) + assert validator.is_valid(good), list(validator.iter_errors(good)) + for bad in ( + {**HIT_LINE, "route": "POST /upload?x=1"}, + {**HIT_LINE, "count": 0}, + {**HIT_LINE, "stack_sample": ["no-line-number"]}, + {**SESSION_LINE, "language": "cobol"}, + {"type": "load", "session": SESSION}, + ): + assert not validator.is_valid(bad) + with pytest.raises(EvidenceError): + validate_event(bad) + + +# ---------- migration: works on an existing database ---------- + + +def test_tables_are_added_to_a_database_created_before_they_existed(tmp_path): + db = tmp_path / "old.sqlite" + with RunRepository(db_path=db): + pass + conn = sqlite3.connect(db) + for table in ("runtime_evidence", "runtime_packages", "runtime_sessions"): + conn.execute(f"DROP TABLE {table}") # simulate a store from before #53 + conn.commit() + conn.close() + + with RunRepository(db_path=db) as repo: + tables = {r[0] for r in repo.connection.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert {"runtime_sessions", "runtime_evidence", "runtime_packages"} <= tables + + +# ---------- CLI ---------- + + +def test_cli_ingest_reports_counts_and_skipped_lines(tmp_path, monkeypatch): + db = tmp_path / "db.sqlite" + monkeypatch.setattr("secfoo.cli.RunRepository", lambda: RunRepository(db_path=db)) + with RunRepository(db_path=db) as repo: + pid = repo.upsert_project("https://github.com/acme/payments", "payments", "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") + events = tmp_path / "events.ndjson" + events.write_text("\n".join(_lines(SESSION_LINE, LOAD_LINE, HIT_LINE)) + "\n{broken\n", encoding="utf-8") + + result = CliRunner().invoke(app, ["runtime", "ingest", str(events), "--project", "payments"]) + + assert result.exit_code == 1 # one line was skipped + assert "Stored 1 session(s), 1 package load(s), 1 hit(s)" in result.output + assert "line 4: not valid JSON" in result.output + + clean = tmp_path / "clean.ndjson" + clean.write_text("\n".join(_lines(SESSION_LINE, HIT_LINE)), encoding="utf-8") + again = CliRunner().invoke(app, ["runtime", "ingest", str(clean), "--project", str(pid)]) + assert again.exit_code == 0, again.output + with RunRepository(db_path=db) as repo: + assert repo.connection.execute("SELECT count FROM runtime_evidence").fetchone()[0] == 312 + + +def test_cli_ingest_unknown_project(tmp_path, monkeypatch): + db = tmp_path / "db.sqlite" + monkeypatch.setattr("secfoo.cli.RunRepository", lambda: RunRepository(db_path=db)) + events = tmp_path / "events.ndjson" + events.write_text(json.dumps(SESSION_LINE), encoding="utf-8") + result = CliRunner().invoke(app, ["runtime", "ingest", str(events), "--project", "nope"]) + assert result.exit_code == 1 + assert "no single project matches 'nope'" in result.output + + +def test_ingest_file_reads_a_path(repo, project_id, tmp_path): + path = tmp_path / "e.ndjson" + path.write_text("\n".join(_lines(SESSION_LINE, HIT_LINE)), encoding="utf-8") + assert ingest_file(repo.connection, project_id, path).hits == 1