diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 387ae8e..6bc6717 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -350,6 +350,83 @@ 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. + +**In the dashboard.** Once a project has runtime data, the SCA +vulnerabilities list gets a **Runtime** column. Each finding shows the +strongest state seen across the projects it affects: + +| State | Meaning | +|---|---| +| Reached from traffic | A watched symbol ran while handling a request route (shows route, auth state and hits). | +| Executed | It ran, with no request context (startup, a job, or no route). | +| Loaded | The package was in memory, but no watched symbol was called. Flags version drift when the loaded version differs from the scanned one. | +| Not loaded / Not observed | Sessions exist, nothing was seen for this finding. Always shown with the coverage % and observation time. | +| No runtime data | No sensor or coverage import for the project. | + +A finding the static scan rated **Not reachable** whose code was Executed or +Reached is highlighted as a conflict. The vulnerability page shows sessions, +routes, entry frames and stack samples. The SCA overview counts +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) + ## Install Pick whichever fits how you work — they're all the same tool underneath. diff --git a/docs/assets/sca-runtime.jpg b/docs/assets/sca-runtime.jpg new file mode 100644 index 0000000..689f0bd Binary files /dev/null and b/docs/assets/sca-runtime.jpg differ 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/src/secfoo/web/sca_dashboard.py b/src/secfoo/web/sca_dashboard.py index 1cb2018..542ca0d 100644 --- a/src/secfoo/web/sca_dashboard.py +++ b/src/secfoo/web/sca_dashboard.py @@ -21,10 +21,12 @@ extract_cve_by_dependency_id, extract_narrative_by_dependency_id, extract_risk_register_rows, + findings_with_fingerprints, inventory_lookup, resolve_ecosystem, ) from secfoo.storage.repository import RunRepository +from secfoo.web import sca_runtime from secfoo.web.dashboard_common import read_report _read = read_report @@ -66,6 +68,10 @@ class ScaVulnerability: reachability_evidence: str | None = None recommendation: str | None = None affected_projects: list[dict] = field(default_factory=list) + # #57: strongest runtime state across affected projects (each project's + # own state is on its affected_projects entry as "runtime"). + runtime: sca_runtime.RuntimeStatus = field(default_factory=sca_runtime.RuntimeStatus) + runtime_conflict: bool = False @dataclass @@ -80,6 +86,12 @@ class ScaDashboard: vulnerabilities_by_key: dict[str, ScaVulnerability] = field(default_factory=dict) unmapped_ecosystem_count: int = 0 osv_unreachable: bool = False + # #57: vulnerabilities runtime evidence confirmed, and any where it + # contradicts a static "Not reachable". + runtime_reached_count: int = 0 + runtime_executed_count: int = 0 + runtime_conflict_count: int = 0 + has_runtime_data: bool = False def _is_worse_reachability(candidate: str, current: str) -> bool: @@ -113,6 +125,9 @@ def build(repo: RunRepository) -> ScaDashboard: continue inventory_by_package_version = inventory_lookup(text) + # The persisted sca_findings fingerprint per Risk Register row, which + # runtime evidence (#53) attaches to. + fingerprint_by_id = {r["id"]: r["fingerprint"] for r in findings_with_fingerprints(text)} cve_by_id = extract_cve_by_dependency_id(text) narrative_by_id = extract_narrative_by_dependency_id(text) @@ -130,6 +145,7 @@ def build(repo: RunRepository) -> ScaDashboard: "project_id": project["project_id"], "project_display_name": project["project_display_name"], "dependency_id": row["id"], + "fingerprint": fingerprint_by_id.get(row["id"]), "package": package, "version": version, "issue": row["issue"].strip(), @@ -210,6 +226,14 @@ def build(repo: RunRepository) -> ScaDashboard: if _is_worse_reachability(finding["reachability"], vuln.reachability): vuln.reachability = finding["reachability"] + runtime = sca_runtime.status_for( + repo.connection, + project_id=finding["project_id"], + fingerprint=finding["fingerprint"], + package=finding["package"], + ecosystem=finding["ecosystem_osv"], + scanned_version=finding["version"], + ) vuln.affected_projects.append( { "project_id": finding["project_id"], @@ -218,9 +242,15 @@ def build(repo: RunRepository) -> ScaDashboard: "severity": finding["severity"], "reachability": finding["reachability"], "fixed_in": finding["fixed_in"], + "runtime": runtime, + "runtime_conflict": sca_runtime.is_conflict(finding["reachability"], runtime), } ) + for vuln in vulnerabilities.values(): + vuln.runtime = sca_runtime.strongest([p["runtime"] for p in vuln.affected_projects]) + vuln.runtime_conflict = any(p["runtime_conflict"] for p in vuln.affected_projects) + dash.vulnerabilities = sorted( vulnerabilities.values(), key=lambda v: (-_SEVERITY_RANK.get(v.severity.strip().lower(), -1), v.package.lower()), @@ -250,4 +280,9 @@ def build(repo: RunRepository) -> ScaDashboard: } dash.ecosystem_counts = ecosystem_counter.most_common() + dash.runtime_reached_count = sum(1 for v in dash.vulnerabilities if v.runtime.state == "reached") + dash.runtime_executed_count = sum(1 for v in dash.vulnerabilities if v.runtime.state == "executed") + dash.runtime_conflict_count = sum(1 for v in dash.vulnerabilities if v.runtime_conflict) + dash.has_runtime_data = any(v.runtime.state != "none" for v in dash.vulnerabilities) + return dash diff --git a/src/secfoo/web/sca_runtime.py b/src/secfoo/web/sca_runtime.py new file mode 100644 index 0000000..c6c5e2e --- /dev/null +++ b/src/secfoo/web/sca_runtime.py @@ -0,0 +1,205 @@ +"""Runtime evidence for the SCA dashboard (#57), read from #53's tables. + +Each (project, SCA finding) gets one state from the evidence ladder in #51, +strongest first: + + reached -- a watched symbol was called while handling a request route + executed -- a watched symbol was called, but with no request context + loaded -- the package was in memory, but no watched symbol was called + not_loaded / not_observed -- sessions exist, nothing seen for this finding + none -- no runtime data for the project at all + +"Not observed" never means "not reachable" (#51 rule 4): those states always +carry the coverage and observation window they were measured over. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime + +# Strongest first; the list row shows the strongest state across projects. +STATES = ("reached", "executed", "loaded", "not_loaded", "not_observed", "none") +_RANK = {state: i for i, state in enumerate(STATES)} +LABELS = { + "reached": "Reached from traffic", + "executed": "Executed", + "loaded": "Loaded", + "not_loaded": "Not loaded", + "not_observed": "Not observed", + "none": "No runtime data", +} +# Existing badge styles only (design-system/components/badges.css). +BADGES = { + "reached": "badge--error", + "executed": "badge--warning", + "loaded": "badge--primary", + "not_loaded": "badge--teal", + "not_observed": "badge--neutral", + "none": "badge--neutral", +} +# A hit on one of these "routes" had no inbound request behind it. +_NO_REQUEST_ROUTES = ("", "unknown", "startup") + + +def _is_request_route(route: str) -> bool: + return route not in _NO_REQUEST_ROUTES and not route.startswith("job:") + + +def _parse(ts: str | None) -> datetime | None: + if not ts: + return None + try: + return datetime.fromisoformat(ts[:-1] + "+00:00" if ts.endswith("Z") else ts) + except ValueError: + return None + + +@dataclass +class Coverage: + sessions: int = 0 + functions_total: int = 0 + functions_executed: int = 0 + observed_hours: float = 0.0 + + @property + def percent(self) -> int | None: + if not self.functions_total: + return None + return round(self.functions_executed / self.functions_total * 100) + + @property + def hours_label(self) -> str: + if self.observed_hours >= 48: + return f"{self.observed_hours / 24:.0f}d" + if self.observed_hours >= 1: + return f"{self.observed_hours:.0f}h" + return f"{self.observed_hours * 60:.0f}m" + + +@dataclass +class RuntimeStatus: + state: str = "none" + hits: int = 0 + last_seen: str | None = None + routes: list[dict] = field(default_factory=list) # route, auth, count -- request routes only + evidence: list[dict] = field(default_factory=list) # every hit row, with its session + loaded_versions: list[str] = field(default_factory=list) + version_drift: bool = False + coverage: Coverage = field(default_factory=Coverage) + sources: list[str] = field(default_factory=list) # sensor / coverage + + @property + def label(self) -> str: + return LABELS[self.state] + + @property + def badge(self) -> str: + return BADGES[self.state] + + @property + def summary(self) -> str: + """One-line cell text, e.g. 'Reached · POST /upload (anonymous) · 312'.""" + cov = self.coverage + if self.state == "reached": + top = self.routes[0] + more = f" +{len(self.routes) - 1} more" if len(self.routes) > 1 else "" + auth = f" ({top['auth']})" if top["auth"] else "" + return f"Reached · {top['route']}{auth} · {top['count']:,}{more}" + if self.state == "executed": + return f"Executed · {self.hits:,} hits" + (f" · last {self.last_seen[:10]}" if self.last_seen else "") + if self.state == "loaded": + return "Loaded · no watched call" + (" · version drift" if self.version_drift else "") + if self.state in ("not_loaded", "not_observed"): + pct = f"{cov.percent}% coverage" if cov.percent is not None else "coverage unknown" + return f"{LABELS[self.state]} · {pct} · {cov.hours_label}" + return "No runtime data" + + +def _coverage(conn, project_id: int) -> tuple[Coverage, list[str], bool]: + rows = conn.execute( + "SELECT source, started_at, ended_at, app_functions_total, app_functions_executed, " + "EXISTS (SELECT 1 FROM runtime_packages p WHERE p.session_id = s.id) AS has_inventory " + "FROM runtime_sessions s WHERE project_id = ?", + (project_id,), + ).fetchall() + cov = Coverage(sessions=len(rows)) + sources: set[str] = set() + has_inventory = False + for source, started, ended, total, executed, inventory in rows: + sources.add(source) + has_inventory = has_inventory or bool(inventory) + if total and executed is not None: + cov.functions_total += total + cov.functions_executed += executed + start, end = _parse(started), _parse(ended) + if start and end and end > start: + cov.observed_hours += (end - start).total_seconds() / 3600 + return cov, sorted(sources), has_inventory + + +def status_for(conn, *, project_id: int, fingerprint: str | None, package: str, + ecosystem: str | None, scanned_version: str | None) -> RuntimeStatus: + """The runtime state of one SCA finding in one project.""" + status = RuntimeStatus() + status.coverage, status.sources, has_inventory = _coverage(conn, project_id) + if not status.coverage.sessions: + return status # "none": no sensor, no import -- SCA works exactly as before + + if fingerprint: + rows = conn.execute( + "SELECT e.symbol, e.route, e.auth, e.count, e.first_seen, e.last_seen, e.entry_frame, e.stack_sample, " + "s.app, s.env, s.instance, s.source, s.language, s.started_at, s.ended_at " + "FROM runtime_evidence e JOIN runtime_sessions s ON s.id = e.session_id " + "WHERE e.project_id = ? AND e.finding_fingerprint = ? ORDER BY e.count DESC", + (project_id, fingerprint), + ).fetchall() + routes: dict[tuple[str, str | None], int] = {} + for (symbol, route, auth, count, first_seen, last_seen, entry_frame, stack, app, env, instance, source, + language, started_at, ended_at) in rows: + status.hits += count + if last_seen and (status.last_seen is None or last_seen > status.last_seen): + status.last_seen = last_seen + if _is_request_route(route): + routes[(route, auth)] = routes.get((route, auth), 0) + count + status.evidence.append({ + "symbol": symbol, "route": route or "—", "auth": auth, "count": count, + "first_seen": first_seen, "last_seen": last_seen, "entry_frame": entry_frame, + "stack_sample": json.loads(stack) if stack else [], + "app": app, "env": env, "instance": instance, "source": source, "language": language, + "started_at": started_at, "ended_at": ended_at, + }) + status.routes = [ + {"route": r, "auth": a, "count": c} + for (r, a), c in sorted(routes.items(), key=lambda item: -item[1]) + ] + if status.routes: + status.state = "reached" + return status + if status.hits: + status.state = "executed" + return status + + loaded = conn.execute( + "SELECT DISTINCT version FROM runtime_packages WHERE project_id = ? AND lower(package) = lower(?) " + "AND (? IS NULL OR ecosystem = ?)", + (project_id, package, ecosystem, ecosystem), + ).fetchall() + status.loaded_versions = sorted(v for (v,) in loaded) + if status.loaded_versions: + status.state = "loaded" + status.version_drift = bool(scanned_version) and scanned_version not in status.loaded_versions + return status + status.state = "not_loaded" if has_inventory else "not_observed" + return status + + +def strongest(statuses: list[RuntimeStatus]) -> RuntimeStatus: + return min(statuses, key=lambda s: _RANK[s.state]) if statuses else RuntimeStatus() + + +def is_conflict(static_reachability: str, status: RuntimeStatus) -> bool: + """The case runtime evidence exists to catch: the static scan said + "Not reachable", but the code actually ran.""" + return static_reachability.strip().lower() == "not reachable" and status.state in ("reached", "executed") diff --git a/src/secfoo/web/static/css/dashboard.css b/src/secfoo/web/static/css/dashboard.css index 51ad39d..b4e9832 100644 --- a/src/secfoo/web/static/css/dashboard.css +++ b/src/secfoo/web/static/css/dashboard.css @@ -412,3 +412,9 @@ tr.run-row:hover { background-color: var(--color-primary-50); } .stride-key { display: inline-flex; align-items: center; gap: var(--space-1); } .stride-key__swatch { width: 0.75rem; height: 0.75rem; border-radius: 3px; display: inline-block; } + +/* #57: an SCA row whose static verdict is "Not reachable" but whose code ran + at runtime -- the case runtime evidence exists to catch. Existing error + tokens only, no new colors. */ +.table tr.runtime-conflict > td { background-color: var(--color-error-bg); } +.table tr.runtime-conflict > td:first-child { box-shadow: inset 3px 0 0 var(--color-error); } diff --git a/src/secfoo/web/templates/activities/sca_dashboard.html b/src/secfoo/web/templates/activities/sca_dashboard.html index 0b96b55..b3356e6 100644 --- a/src/secfoo/web/templates/activities/sca_dashboard.html +++ b/src/secfoo/web/templates/activities/sca_dashboard.html @@ -42,6 +42,19 @@ {{ sca_dash.reachability_counts.reachable }} Reachable + {% if sca_dash.has_runtime_data %} + {% set runtime_confirmed = sca_dash.runtime_reached_count + sca_dash.runtime_executed_count %} +
+ {% call info_tip("Runtime-confirmed reachable") %} +

Vulnerabilities whose vulnerable code actually ran in a deployed app, from a runtime sensor or an imported coverage file.

+

Reached from traffic means it ran while handling a request. Executed means it ran, with no request context.

+ {% endcall %} + {{ runtime_confirmed }} + + Runtime-confirmed reachable: {{ sca_dash.runtime_reached_count }} reached from traffic, {{ sca_dash.runtime_executed_count }} executed{% if sca_dash.runtime_conflict_count %}. {{ sca_dash.runtime_conflict_count }} contradict{{ "s" if sca_dash.runtime_conflict_count == 1 }} a static "Not reachable"{% endif %} + +
+ {% endif %}
{% call info_tip("Unrecognized ecosystem") %}

Findings whose package ecosystem OSV.dev does not recognize, so they could not be checked against a known advisory.

diff --git a/src/secfoo/web/templates/activities/sca_vulnerabilities_list.html b/src/secfoo/web/templates/activities/sca_vulnerabilities_list.html index a488282..162a670 100644 --- a/src/secfoo/web/templates/activities/sca_vulnerabilities_list.html +++ b/src/secfoo/web/templates/activities/sca_vulnerabilities_list.html @@ -25,12 +25,12 @@

All vulnerabilities

SeverityCVE / IssuePackageVersion - ReachabilityFixed inApplications + ReachabilityRuntimeFixed inApplications {% for v in vulnerabilities %} - + {{ v.severity }} {% if v.cve_id %}{{ v.cve_id }}{% else %}unverified {{ v.issue }}{% endif %} @@ -38,6 +38,11 @@

All vulnerabilities

{{ v.package }} {{ v.version }} {{ v.reachability }} + + {{ v.runtime.label }} + {% if v.runtime.state != "none" %}{{ v.runtime.summary }}{% endif %} + {% if v.runtime_conflict %}Conflicts with static verdict{% endif %} + {{ v.fixed_in }} {{ v.affected_projects | length }} diff --git a/src/secfoo/web/templates/activities/sca_vulnerability_detail.html b/src/secfoo/web/templates/activities/sca_vulnerability_detail.html index f3c2ae1..1447020 100644 --- a/src/secfoo/web/templates/activities/sca_vulnerability_detail.html +++ b/src/secfoo/web/templates/activities/sca_vulnerability_detail.html @@ -56,6 +56,65 @@

Risk & recommendation

{% endif %} + {% if vulnerability.runtime_conflict %} + + {% endif %} + +
+

Runtime evidence

+ {% for p in vulnerability.affected_projects %} + {% set rt = p.runtime %} +
+
+ {{ p.project_display_name }} + {{ rt.label }} + {% if p.runtime_conflict %}Conflicts with static verdict{% endif %} + {% if rt.sources %}Source: {{ rt.sources | join(", ") }}{% endif %} +
+ {% if rt.state == "none" %} +

No runtime sensor or coverage import for this project yet. See secfoo runtime ingest.

+ {% elif rt.state in ("not_loaded", "not_observed") %} +

+ {{ rt.summary }} across {{ rt.coverage.sessions }} session(s). + This is not proof the code is unreachable: traffic only exercised + {% if rt.coverage.percent is not none %}{{ rt.coverage.percent }}%{% else %}an unknown share{% endif %} of the app. +

+ {% elif rt.state == "loaded" %} +

+ Loaded version{{ "s" if rt.loaded_versions | length > 1 }}: {{ rt.loaded_versions | join(", ") }}. + {% if rt.version_drift %}Version drift: production runs a different version than the {{ vulnerability.version }} that was scanned.{% endif %} + No watched vulnerable function was called. +

+ {% else %} +
+ + + + {% for e in rt.evidence %} + + + + + + + + + + {% if e.stack_sample %} + + {% endif %} + {% endfor %} + +
SymbolRouteAuthHitsLast seenApp / envEntry frame
{{ e.symbol }}{{ e.route }}{{ e.auth or "—" }}{{ "{:,}".format(e.count) }}{{ (e.last_seen or "—")[:19] }}{{ e.app or "—" }}{% if e.env %} / {{ e.env }}{% endif %}{% if e.entry_frame %}{{ e.entry_frame }}{% else %}—{% endif %}
Stack sample: {% for f in e.stack_sample %}{{ f }}{% if not loop.last %} ← {% endif %}{% endfor %}
+
+ {% endif %} +
+ {% endfor %} +
+

Affected applications

{% if vulnerability.affected_projects %}
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 diff --git a/tests/test_web/test_sca_runtime.py b/tests/test_web/test_sca_runtime.py new file mode 100644 index 0000000..e6846b8 --- /dev/null +++ b/tests/test_web/test_sca_runtime.py @@ -0,0 +1,216 @@ +"""#57: runtime evidence in the SCA dashboard -- every runtime state and the conflict highlight.""" +from __future__ import annotations + +import json + +import pytest +from fastapi.testclient import TestClient + +from secfoo.report.sca import findings_with_fingerprints +from secfoo.runtime.evidence import ingest_events +from secfoo.storage.repository import RunRepository +from secfoo.web import sca_dashboard, sca_runtime +from secfoo.web.app import create_app + +REPORT = """\ +# SCA Report — Reachability & Upgrade Triage + +## 1. Executive Summary +Test summary. **Overall risk rating:** High + +## 2. Dependency Inventory + +| Package | Version | Direct/Transitive | Ecosystem | Source manifest | +|---------|---------|--------------------|-----------|------------------| +| pyyaml | 5.3 | Direct | Python | requirements.txt | + +## 4. Risk Register + +| ID | Package | Version | Issue | Severity | Reachability | Fixed in | +|----|---------|---------|-------|----------|---------------|----------| +| D1 | pyyaml | 5.3 | Arbitrary code execution in full_load | High | {reachability} | 5.4 | + +## 5. Detailed Findings + +### [HIGH] D1: pyyaml@5.3 — code execution +- **Package / version:** pyyaml 5.3, **CVE:** unverified, **Severity:** High +- **Reachability:** {reachability} +- **Recommendation:** upgrade to 5.4 +""" + +SESSION = {"type": "session", "id": "s-1", "language": "python", "source": "sensor", + "started_at": "2026-11-08T10:00:00Z", "ended_at": "2026-11-08T16:00:00Z", + "app": "payments-api", "env": "prod", "app_functions_total": 1000, "app_functions_executed": 740} + + +@pytest.fixture +def store(tmp_path, monkeypatch): + db_path = tmp_path / "db.sqlite" + monkeypatch.setattr("secfoo.storage.repository.DB_PATH", db_path) + monkeypatch.setattr("secfoo.storage.repository.ensure_store_dirs", lambda: None) + monkeypatch.setattr("secfoo.web.sca_dashboard.osv.enrich_lookups", lambda repo, keys, **kw: False) + return tmp_path + + +def _seed(tmp_path, *, reachability="Not reachable", events=()): + """One project, one SCA finding (persisted the way the runner does it), + and optional runtime events ingested through #53's real code path.""" + text = REPORT.format(reachability=reachability) + with RunRepository() as repo: + pid = repo.upsert_project("https://github.com/acme/payments", "payments", "github") + aid = repo.create_assessment(project_id=pid, assessment_type="internal", status="ready") + run_uuid = repo.create_run(project_id=pid, skill_id="sca-reachability", skill_name="SCA", + agent_id="claude", confluence_urls=[], assessment_id=aid) + path = tmp_path / f"{run_uuid}.md" + path.write_text(text, encoding="utf-8") + repo.complete_run(run_uuid, status="success", exit_code=0, duration_seconds=1.0, report_path=str(path), + prompt_path=None, stderr_excerpt=None, high_count=1) + [row] = findings_with_fingerprints(text) + repo.upsert_sca_finding(project_id=pid, fingerprint=row["fingerprint"], run_id=repo.get_run(run_uuid).id, + current_ref="D1", package="pyyaml", severity="High", version="5.3", + ecosystem="PyPI", reachability=reachability) + if events: + lines = [json.dumps({**e, "finding_fingerprint": row["fingerprint"]} if e["type"] == "hit" else e) + for e in events] + report = ingest_events(repo.connection, pid, lines) + assert report.skipped == [], report.skipped + return row["fingerprint"] + + +def _hit(**overrides): + return {"type": "hit", "session": "s-1", "symbol": "yaml.full_load", "count": 312, + "last_seen": "2026-11-08T15:59:00Z", **overrides} + + +def _vuln(): + with RunRepository() as repo: + [vuln] = sca_dashboard.build(repo).vulnerabilities + return vuln + + +# ---------- each state of the evidence ladder ---------- + + +def test_no_runtime_data(store): + _seed(store) + vuln = _vuln() + assert vuln.runtime.state == "none" and vuln.runtime.label == "No runtime data" + assert not vuln.runtime_conflict + + +def test_reached_from_traffic_shows_route_auth_and_count(store): + _seed(store, events=[SESSION, _hit(route="POST /upload", auth="anonymous")]) + rt = _vuln().runtime + assert rt.state == "reached" + assert rt.summary == "Reached · POST /upload (anonymous) · 312" + + +def test_hits_without_request_context_are_executed(store): + _seed(store, events=[SESSION, _hit(route="startup", count=4), _hit(route="job:nightly-import", count=9)]) + rt = _vuln().runtime + assert rt.state == "executed" and rt.hits == 13 + assert rt.summary.startswith("Executed · 13 hits · last 2026-11-08") + + +def test_loaded_package_with_no_watched_call_and_version_drift(store): + _seed(store, events=[SESSION, {"type": "load", "session": "s-1", "ecosystem": "PyPI", + "package": "PyYAML", "version": "5.4.1"}]) + rt = _vuln().runtime + assert rt.state == "loaded" and rt.loaded_versions == ["5.4.1"] + assert rt.version_drift # production runs 5.4.1, the scan saw 5.3 + + +def test_not_loaded_when_the_inventory_lacks_the_package_shows_coverage(store): + _seed(store, events=[SESSION, {"type": "load", "session": "s-1", "ecosystem": "PyPI", + "package": "requests", "version": "2.31.0"}]) + rt = _vuln().runtime + assert rt.state == "not_loaded" + assert rt.summary == "Not loaded · 74% coverage · 6h" # never shown without its coverage + + +def test_not_observed_when_sessions_have_no_inventory(store): + _seed(store, events=[SESSION]) + rt = _vuln().runtime + assert rt.state == "not_observed" and rt.summary == "Not observed · 74% coverage · 6h" + + +def test_strongest_state_wins_across_projects(): + reached, loaded = sca_runtime.RuntimeStatus(state="reached"), sca_runtime.RuntimeStatus(state="loaded") + assert sca_runtime.strongest([loaded, reached]) is reached + assert sca_runtime.strongest([]).state == "none" + + +# ---------- the conflict highlight ---------- + + +@pytest.mark.parametrize("route", ["POST /upload", "startup"]) +def test_not_reachable_but_ran_is_a_conflict(store, route): + _seed(store, reachability="Not reachable", events=[SESSION, _hit(route=route)]) + assert _vuln().runtime_conflict + + +def test_reachable_and_ran_is_not_a_conflict(store): + _seed(store, reachability="Reachable", events=[SESSION, _hit(route="POST /upload")]) + assert not _vuln().runtime_conflict + + +def test_not_reachable_and_only_loaded_is_not_a_conflict(store): + _seed(store, reachability="Not reachable", events=[SESSION, {"type": "load", "session": "s-1", + "ecosystem": "PyPI", "package": "pyyaml", "version": "5.3"}]) + assert not _vuln().runtime_conflict + + +# ---------- rendered pages ---------- + + +def test_list_page_shows_runtime_column_and_highlights_the_conflict(store): + _seed(store, events=[SESSION, _hit(route="POST /upload", auth="anonymous")]) + html = TestClient(create_app()).get("/activities/sca-reachability/vulnerabilities").text + assert "Runtime" in html + assert 'data-runtime-state="reached"' in html + assert "Reached · POST /upload (anonymous) · 312" in html + assert 'class="run-row runtime-conflict"' in html + assert "Conflicts with static verdict" in html + + +def test_list_page_without_runtime_data_says_so_and_has_no_highlight(store): + _seed(store) + html = TestClient(create_app()).get("/activities/sca-reachability/vulnerabilities").text + assert 'data-runtime-state="none"' in html and "No runtime data" in html + assert "runtime-conflict" not in html + + +def test_detail_page_shows_sessions_routes_frames_and_the_conflict(store): + _seed(store, events=[SESSION, _hit(route="POST /upload", auth="anonymous", entry_frame="app/routes/upload.py:42", + stack_sample=["app/routes/upload.py:42", "app/services/parse.py:17"])]) + client = TestClient(create_app()) + key = _vuln().key + html = client.get(f"/activities/sca-reachability/vulnerabilities/{key}").text + assert "Runtime evidence contradicts the static verdict" in html + assert "yaml.full_load" in html and "POST /upload" in html and "anonymous" in html + assert "payments-api / prod" in html + assert "app/routes/upload.py:42" in html and "app/services/parse.py:17" in html + assert "Source: sensor" in html + + +def test_detail_page_not_observed_never_claims_unreachable(store): + _seed(store, events=[SESSION]) + key = _vuln().key + html = TestClient(create_app()).get(f"/activities/sca-reachability/vulnerabilities/{key}").text + assert "This is not proof the code is unreachable" in html and "74%" in html + + +def test_overview_counts_runtime_confirmed_only_when_there_is_runtime_data(store): + _seed(store) + client = TestClient(create_app()) + assert 'id="runtime-confirmed"' not in client.get("/activities/sca-reachability").text + + with RunRepository() as repo: + [finding] = repo.list_sca_findings() + ingest_events(repo.connection, finding.project_id, [ + json.dumps(SESSION), json.dumps({**_hit(route="POST /upload"), "finding_fingerprint": finding.fingerprint}), + ]) + html = client.get("/activities/sca-reachability").text + assert 'id="runtime-confirmed"' in html + assert "1 reached from traffic, 0 executed" in html + assert '1 contradicts a static "Not reachable"' in html