From 8d0053b3db50ebc3a16fc4811a0fcd81982b9adb Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 04:16:10 +0000 Subject: [PATCH] feat(runtime): runtime evidence in the CLI, SARIF and a CI gate (#80) RuntimeLookup merges evidence from the local store and, only when a cloud key is configured, the #66 evidence endpoint (cached for offline use). secfoo findings list/show gain a Runtime column sorted by what production saw and a remediation hint; findings export adds properties.secfoo.runtime to SARIF and --runtime-priority; findings check --fail-on reached[:severity] exits 4; secfoo run prints a one-line runtime summary. --no-runtime skips it all. Closes #80 Co-Authored-By: Claude Opus 5.5 --- docs/REFERENCE.md | 38 ++++ src/secfoo/cli.py | 185 ++++++++++++++++++-- src/secfoo/report/sarif.py | 15 +- src/secfoo/runtime/developer.py | 192 +++++++++++++++++++++ tests/test_runtime/test_developer.py | 249 +++++++++++++++++++++++++++ 5 files changed, 667 insertions(+), 12 deletions(-) create mode 100644 src/secfoo/runtime/developer.py create mode 100644 tests/test_runtime/test_developer.py diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index c028715..10114a4 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -540,6 +540,44 @@ request-like handler): The fallback fires on every Python call, so prefer 3.12+ in production or keep the fallback to staging. +### Runtime evidence in the CLI and CI + +Runtime evidence is most useful where fixes are chosen: in the branch, the PR +and CI. The CLI attaches it to findings automatically. + +```bash +secfoo findings list # Runtime column; sorted by severity, then by what production saw +secfoo findings show 3f2a9c # one finding, its routes and a "Fix first" hint +secfoo findings export --format sarif --runtime-priority -o findings.sarif +secfoo findings check --fail-on reached:high # CI gate: exit 4 if an open High+ finding is reached in production +``` + +Evidence comes from two places, and the stronger state wins: + +- **The local store**, for anything ingested here or by a self-hosted + `secfoo serve` that shares the store. +- **The server your `secfoo cloud login` key belongs to**, via + `GET /api/runtime/v1/evidence`. The answer is cached under + `$SECFOO_HOME/runtime-cache/`, so offline runs reuse the last one. + +With no key, no network call is made. **No runtime data** is the normal state +when no sensor is deployed, not an error. `--no-runtime` skips all of it. + +- **Ordering** within each severity: Reached, Executed, Loaded, No runtime + data, Not observed / Not loaded. Not-observed findings are never hidden, and + always show their coverage. +- **SARIF.** Each SCA result carries `properties.secfoo.runtime` (state, + summary, routes, hits, last seen, version drift). A reached finding's + message says "Reached in production via …". `--runtime-priority` raises + reached findings to `error`. +- **Gate.** `--fail-on reached[:]` counts findings reached on a + request route. `executed[:]` also counts code that ran with no + request context. The gate exits **`4`** (distinct from `1`, `2` and `3` of + `secfoo run`) and is off unless you run it. +- **After `secfoo run`**, when runtime data exists for the project, the + summary adds one line, e.g. *"2 of 12 findings are reached in production: + D2 (POST /upload, anonymous), …"*. + ### Runtime ingest API Live sensors send evidence to the same server your CLI results go to. They diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index ec48f84..e6cfff4 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -540,12 +540,45 @@ def handle_complete(skill_name: str, result_status: str) -> None: f"-- reported spend {format_cost(total_cost)}" + (f", {unpriced_runs} run(s) unpriced" if unpriced_runs else "") ) + line = _runtime_summary_line(outcomes) + if line: + console.print(line) console.print("Run [bold]secfoo serve[/] to view full reports in your browser.") if exit_code: raise typer.Exit(code=exit_code) +def _runtime_summary_line(outcomes) -> str | None: + """#80: '3 of 12 findings are reached in production: D2 (POST /upload, anonymous), ...' + -- only when runtime data exists for the scanned project. Never fails a run.""" + try: + from secfoo.runtime.developer import RuntimeLookup + + with RunRepository() as repo: + project_ids = {run.project_id for o in outcomes if (run := repo.get_run(o.run_uuid)) is not None} + lookup = RuntimeLookup(repo) + rows = [row for pid in project_ids + for _, row in repo.list_structured_findings(skill_ids=["sca-reachability"], project_id=pid, + status="open")] + states = [(row, lookup.for_finding("sca-reachability", row)) for row in rows] + except Exception: # noqa: BLE001 -- a summary line must never break `secfoo run` + return None + if not any(runtime.state != "none" for _, runtime in states): + return None + reached = [(row, runtime) for row, runtime in states if runtime.is_reached] + if not reached: + return f"Runtime: none of {len(states)} open SCA finding(s) are reached in production yet." + examples = [] + for row, runtime in reached[:3]: + top = runtime.routes[0] if runtime.routes else {} + where = ", ".join(x for x in (top.get("route"), top.get("auth")) if x) + examples.append(f"{row.get('current_ref') or row['fingerprint'][:8]} ({where})" if where else row.get("current_ref", "")) + more = ", ..." if len(reached) > 3 else "" + return (f"[bold red]{len(reached)} of {len(states)} findings are reached in production:[/] " + f"{', '.join(examples)}{more}") + + @app.command(name="list") def list_runs( project: Optional[str] = typer.Option(None, "--project", "-p", help="Filter by project name/URL substring."), @@ -1582,29 +1615,44 @@ def findings_export( False, "--no-locations", help="Drop file paths, line numbers, code-region hashes and reachability evidence.", ), + no_runtime: Annotated[bool, typer.Option( + "--no-runtime", help="Don't attach runtime evidence (and don't fetch it from the server).", + )] = False, + runtime_priority: Annotated[bool, typer.Option( + "--runtime-priority", help="SARIF: raise findings reached in production to level 'error'.", + )] = False, ) -> None: """Export structured findings (SAST, SCA, secret scanning) as JSONL or SARIF. One record per finding, with its skill, severity, stack (language/ framework) and code-free construct description -- the same shape for - every skill, for feeding other tools. + every skill, for feeding other tools. SCA findings carry their runtime + evidence ("runtime"; in SARIF, properties.secfoo.runtime). """ + from secfoo.runtime.developer import RuntimeLookup + + records = [] with RunRepository() as repo: rows = repo.list_structured_findings( skill_ids=[s.value for s in skill] or None, project_id=project, status=None if status is FindingsStatus.ALL else status.value, ) - records = [] - for skill_id, row in rows: - record = {"skill_id": skill_id, "project": row.pop("display_name", None)} - for column, value in row.items(): - if column in _EXPORT_DROP_COLUMNS or (no_locations and column in _EXPORT_LOCATION_COLUMNS): - continue - record[column] = value - records.append(record) + lookup = RuntimeLookup(repo, enabled=not no_runtime) + for skill_id, row in rows: + runtime = lookup.for_finding(skill_id, row) + record = {"skill_id": skill_id, "project": row.pop("display_name", None)} + for column, value in row.items(): + if column in _EXPORT_DROP_COLUMNS or (no_locations and column in _EXPORT_LOCATION_COLUMNS): + continue + record[column] = value + if skill_id == "sca-reachability" and not no_runtime: + record["runtime"] = runtime.as_sarif() + records.append(record) + for note in lookup.notes: + err_console.print(f"[yellow]Note:[/] {note}") if format is FindingsFormat.SARIF: - text = json.dumps(findings_to_sarif(records), indent=2, sort_keys=True) + "\n" + text = json.dumps(findings_to_sarif(records, runtime_priority=runtime_priority), indent=2, sort_keys=True) + "\n" else: lines = [json.dumps(record, sort_keys=True) for record in records] text = "\n".join(lines) + ("\n" if lines else "") @@ -1615,6 +1663,123 @@ def findings_export( err_console.print(f"Wrote {len(records)} finding(s) to {output}.") +EXIT_RUNTIME_REACHED = 4 + + +def _finding_title(skill_id: str, row: dict) -> str: + if skill_id == "sca-reachability": + return f"{row.get('package')}@{row.get('version') or '?'}: {row.get('issue') or row.get('cve') or ''}".strip() + return row.get("title") or row.get("secret_type") or skill_id + + +def _finding_rows(repo, *, skill, project, status, no_runtime): + from secfoo.runtime.developer import RuntimeLookup, sort_key + + rows = repo.list_structured_findings( + skill_ids=[s.value for s in skill] or None, project_id=project, + status=None if status is FindingsStatus.ALL else status.value, + ) + lookup = RuntimeLookup(repo, enabled=not no_runtime) + items = [(skill_id, row, lookup.for_finding(skill_id, row)) for skill_id, row in rows] + items.sort(key=lambda item: sort_key(item[1], item[2])) + return items, lookup.notes + + +@findings_app.command(name="list") +def findings_list( + skill: Annotated[list[FindingsSkill] | None, typer.Option("--skill", "-s", help="Limit to these skills.")] = None, + project: Annotated[int | None, typer.Option("--project", help="Limit to one project id.")] = None, + status: Annotated[FindingsStatus, typer.Option("--status", help="open, closed, or all.")] = FindingsStatus.OPEN, + no_runtime: Annotated[bool, typer.Option("--no-runtime", help="Skip runtime evidence (no fetch).")] = False, +) -> None: + """Findings with their runtime evidence, sorted by severity and then by what production saw. + + Runtime: Reached (route, hits) > Executed > Loaded > No runtime data > + Not observed (always shown with its coverage). Fetched from the server + your `secfoo cloud login` key belongs to, and cached for offline use. + """ + with RunRepository() as repo: + items, notes = _finding_rows(repo, skill=skill or [], project=project, status=status, no_runtime=no_runtime) + table = Table(show_header=True, header_style="bold") + for column in ("fingerprint", "skill", "project", "severity", "finding", "runtime"): + table.add_column(column, overflow="fold") + for skill_id, row, runtime in items: + table.add_row(row["fingerprint"][:10], skill_id, row.get("display_name") or "", row.get("severity") or "", + _finding_title(skill_id, row), "-" if no_runtime else runtime.summary) + console.print(table) + for note in notes: + err_console.print(f"[yellow]Note:[/] {note}") + + +@findings_app.command(name="show") +def findings_show( + fingerprint: Annotated[str, typer.Argument(help="Fingerprint (or its first characters) from `findings list`.")], + no_runtime: Annotated[bool, typer.Option("--no-runtime", help="Skip runtime evidence (no fetch).")] = False, +) -> None: + """One finding in full, with its runtime evidence and a remediation hint.""" + from secfoo.runtime.developer import remediation_hint + + with RunRepository() as repo: + items, notes = _finding_rows(repo, skill=[], project=None, status=FindingsStatus.ALL, no_runtime=no_runtime) + matches = [item for item in items if item[1]["fingerprint"].startswith(fingerprint)] + if len(matches) != 1: + err_console.print(f"[red]Error:[/] {len(matches)} findings match {fingerprint!r}; give more characters.") + raise typer.Exit(code=1) + skill_id, row, runtime = matches[0] + console.print(f"[bold]{_finding_title(skill_id, row)}[/] ({skill_id}, {row.get('severity')}, {row.get('status')})") + console.print(f"Project: {row.get('display_name')} Fingerprint: {row['fingerprint']}") + for label, key in (("Location", "location_file"), ("Reachability", "reachability"), ("Fixed in", "fixed_in"), + ("Recommendation", "recommendation")): + if row.get(key): + console.print(f"{label}: {row[key]}") + if not no_runtime: + console.print(f"Runtime: {runtime.summary}" + (f" [dim](from {runtime.source})[/]" if runtime.source != "none" else "")) + for route in runtime.routes: + console.print(f" {route['route']} {route.get('auth') or ''} {route['count']:,} hit(s)") + if runtime.loaded_versions: + console.print(f" loaded: {', '.join(runtime.loaded_versions)}" + (" (version drift)" if runtime.version_drift else "")) + hint = remediation_hint(row, runtime) + if hint: + console.print(f"[bold]Fix first:[/] {hint}") + for note in notes: + err_console.print(f"[yellow]Note:[/] {note}") + + +@findings_app.command(name="check") +def findings_check( + fail_on: Annotated[str, typer.Option( + "--fail-on", help="reached[:] or executed[:], e.g. reached:high.", + )], + project: Annotated[int | None, typer.Option("--project", help="Limit to one project id.")] = None, +) -> None: + """CI gate: exit 4 when an open finding at that severity or worse is reached in production. + + Off unless you run it. With no runtime data (no sensor, no key, offline + without a cache) nothing is reached and the gate passes. + """ + from secfoo.runtime.developer import parse_gate, trips_gate + + try: + gate = parse_gate(fail_on) + except ValueError as exc: + err_console.print(f"[red]Error:[/] {exc}") + raise typer.Exit(code=2) from None + with RunRepository() as repo: + items, notes = _finding_rows(repo, skill=[FindingsSkill.SCA], project=project, status=FindingsStatus.OPEN, + no_runtime=False) + tripped = [(row, runtime) for skill_id, row, runtime in items if trips_gate(gate, row, runtime)] + for note in notes: + err_console.print(f"[yellow]Note:[/] {note}") + if not tripped: + console.print(f"Runtime gate ({fail_on}): pass, no open finding matches.") + return + console.print(f"[red]Runtime gate ({fail_on}): {len(tripped)} open finding(s) reached in production:[/]") + for row, runtime in tripped: + console.print(f" {row['fingerprint'][:10]} {row.get('severity')} {_finding_title('sca-reachability', row)}" + f" -- {runtime.summary}") + raise typer.Exit(code=EXIT_RUNTIME_REACHED) + + @runtime_app.command(name="ingest") def runtime_ingest( events: Annotated[ diff --git a/src/secfoo/report/sarif.py b/src/secfoo/report/sarif.py index f5c74c8..2bea2e8 100644 --- a/src/secfoo/report/sarif.py +++ b/src/secfoo/report/sarif.py @@ -38,13 +38,18 @@ def _location(record: dict) -> dict | None: return {"physicalLocation": physical} -def findings_to_sarif(records: list[dict]) -> dict: +def findings_to_sarif(records: list[dict], *, runtime_priority: bool = False) -> dict: """Build one run; group rules by skill, CWE and severity. Severity is part of the rule identity because GitHub reads security-severity from the rule, while findings with the same CWE can have different levels. Only explicit result fields are copied: internal IDs and secret evidence are not needed by code scanning. + + A record carrying "runtime" (#80) gets it as properties.secfoo.runtime, + so PR annotations can say "reached in production via POST /upload"; + with runtime_priority, a finding reached in production is raised to + level "error". """ rules = [] rule_indices = {} @@ -71,12 +76,18 @@ def findings_to_sarif(records: list[dict]) -> dict: title = record.get("title") or record.get("issue") or record.get("secret_type") or f"{skill} finding" description = record.get("description") or record.get("construct") message = str(title) + (f"\n\n{description}" if description else "") + runtime = record.get("runtime") + if runtime and runtime.get("state") == "reached": + routes = ", ".join(r["route"] for r in runtime.get("routes", [])[:3]) + message += f"\n\nReached in production via {routes}." if routes else "\n\nReached in production." result = { "ruleId": rule_id, "ruleIndex": rule_indices[rule_id], - "level": level, + "level": "error" if runtime_priority and runtime and runtime.get("state") == "reached" else level, "message": {"text": message}, } + if runtime: + result["properties"] = {"secfoo": {"runtime": runtime}} location = _location(record) if location is not None: result["locations"] = [location] diff --git a/src/secfoo/runtime/developer.py b/src/secfoo/runtime/developer.py new file mode 100644 index 0000000..18dc5e5 --- /dev/null +++ b/src/secfoo/runtime/developer.py @@ -0,0 +1,192 @@ +"""Runtime evidence at the developer stage (#80): the CLI, SARIF and CI gate. + +`RuntimeLookup` answers "what did production see for this finding?" for +`secfoo findings list/show/check/export` and the `secfoo run` summary. + +Sources, merged (the stronger state wins): + +1. the local store -- evidence ingested here (`secfoo runtime ingest`, or a + self-hosted `secfoo serve` sharing this store). No network. +2. the server the user's API key belongs to (`secfoo cloud login`), via + `GET /api/runtime/v1/evidence` (#66). Cached under + $SECFOO_HOME/runtime-cache/, so offline runs reuse the last answer. + +Without a key there is no network call at all, and findings show "No +runtime data" -- the normal state when no sensor is deployed, never an error. +""" + +from __future__ import annotations + +import hashlib +import json +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass, field + +from secfoo.web import sca_runtime + +# Display / sort order within a severity (#80): strongest evidence first, +# and "not observed" last but never hidden (#51 rule 4). +ORDER = ("reached", "executed", "loaded", "none", "not_loaded", "not_observed") +_RANK = {state: i for i, state in enumerate(ORDER)} +FETCH_TIMEOUT_SECONDS = 5 + + +@dataclass +class Runtime: + state: str = "none" + summary: str = "No runtime data" + routes: list[dict] = field(default_factory=list) + hits: int = 0 + last_seen: str | None = None + loaded_versions: list[str] = field(default_factory=list) + version_drift: bool = False + source: str = "none" # local / server / cache / none + + @property + def rank(self) -> int: + return _RANK.get(self.state, len(ORDER)) + + @property + def is_reached(self) -> bool: + return self.state == "reached" + + def label(self) -> str: + return self.summary + + def as_sarif(self) -> dict: + return {"state": self.state, "summary": self.summary, "routes": self.routes, "hits": self.hits, + "last_seen": self.last_seen, "version_drift": self.version_drift} + + +def _stronger(a: Runtime, b: Runtime) -> Runtime: + # sca_runtime's ladder: reached > executed > loaded > not_loaded > not_observed > none + order = sca_runtime.STATES + return a if order.index(a.state) <= order.index(b.state) else b + + +def _from_status(status: sca_runtime.RuntimeStatus, source: str) -> Runtime: + return Runtime(state=status.state, summary=status.summary, routes=status.routes, hits=status.hits, + last_seen=status.last_seen, loaded_versions=status.loaded_versions, + version_drift=status.version_drift, source=source if status.state != "none" else "none") + + +def _from_server(row: dict) -> Runtime: + return Runtime(state=row.get("runtime_state") or "none", summary=row.get("runtime_summary") or "No runtime data", + routes=row.get("routes") or [], hits=row.get("hits") or 0, + loaded_versions=row.get("loaded_versions") or [], version_drift=bool(row.get("version_drift")), + source="server") + + +class RuntimeLookup: + def __init__(self, repo, *, enabled: bool = True, fetch=None, cloud_config=None) -> None: + """`fetch(url, api_key) -> dict` and `cloud_config` are injectable for tests; + by default the key and URL come from `secfoo cloud login`.""" + self.repo = repo + self.enabled = enabled + self._fetch = fetch or _http_get_json + self._cloud = cloud_config + self._cloud_loaded = cloud_config is not None + self._server: dict[int, dict[str, Runtime]] = {} + self.notes: list[str] = [] + + def _cloud_config(self): + if not self._cloud_loaded: + self._cloud_loaded = True + try: + from secfoo.cloud import load_cloud_config + + self._cloud = load_cloud_config() + except Exception: # noqa: BLE001 -- a broken cloud.toml just means "no server" + self._cloud = None + return self._cloud + + def _cache_path(self, project_identifier: str): + from secfoo.config import STORE_DIR + + digest = hashlib.sha256(project_identifier.encode("utf-8")).hexdigest()[:24] + return STORE_DIR / "runtime-cache" / f"{digest}.json" + + def _server_states(self, project_id: int) -> dict[str, Runtime]: + if project_id in self._server: + return self._server[project_id] + states: dict[str, Runtime] = {} + config = self._cloud_config() + project = self.repo.get_project(project_id) + if config is not None and project is not None: # no key: no network call, ever + cache = self._cache_path(project.identifier) + url = (f"{config.portal_url.rstrip('/')}/api/runtime/v1/evidence?" + + urllib.parse.urlencode({"project": project.identifier})) + payload, source = None, "server" + try: + payload = self._fetch(url, config.api_key) + cache.parent.mkdir(parents=True, exist_ok=True) + cache.write_text(json.dumps(payload), encoding="utf-8") + except (OSError, ValueError, urllib.error.URLError) as exc: + if cache.is_file(): + payload, source = json.loads(cache.read_text(encoding="utf-8")), "cache" + self.notes.append(f"runtime evidence from cache (server unreachable: {exc.__class__.__name__})") + for row in (payload or {}).get("findings", []): + runtime = _from_server(row) + runtime.source = source + states[row.get("finding_fingerprint", "")] = runtime + self._server[project_id] = states + return states + + def for_finding(self, skill_id: str, row: dict) -> Runtime: + """Runtime state of one findings row (only SCA findings carry evidence).""" + if not self.enabled or skill_id != "sca-reachability": + return Runtime() + local = _from_status(sca_runtime.status_for( + self.repo.connection, project_id=row["project_id"], fingerprint=row["fingerprint"], + package=row.get("package") or "", ecosystem=row.get("ecosystem"), scanned_version=row.get("version"), + ), "local") + remote = self._server_states(row["project_id"]).get(row["fingerprint"]) + return _stronger(local, remote) if remote is not None else local + + +_SEVERITY_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4, "informational": 4} + + +def severity_rank(severity: str | None) -> int: + return _SEVERITY_RANK.get((severity or "").strip().lower(), 5) + + +def sort_key(row: dict, runtime: Runtime) -> tuple: + return (severity_rank(row.get("severity")), runtime.rank, row.get("display_name") or "", row.get("fingerprint") or "") + + +def remediation_hint(row: dict, runtime: Runtime) -> str | None: + """For a reached SCA finding: the fixed version and the live routes it protects.""" + if not runtime.is_reached: + return None + routes = ", ".join(f"{r['route']}" + (f" ({r['auth']})" if r.get("auth") else "") for r in runtime.routes[:5]) + fixed = (row.get("fixed_in") or "").strip() + fix = f"upgrade {row.get('package')} to {fixed}" if fixed and fixed.lower() not in ("unknown", "n/a", "-") \ + else f"replace or patch {row.get('package')}" + return f"{fix}; this protects live route(s): {routes}" + + +def parse_gate(spec: str) -> tuple[str, int]: + """`reached[:]` or `executed[:]` -> (state, severity rank).""" + state, _, severity = spec.partition(":") + if state not in ("reached", "executed"): + raise ValueError("--fail-on must be reached[:] or executed[:]") + if severity and severity.lower() not in _SEVERITY_RANK: + raise ValueError(f"unknown severity {severity!r}; use critical, high, medium, low or info") + return state, (_SEVERITY_RANK[severity.lower()] if severity else 4) + + +def trips_gate(gate: tuple[str, int], row: dict, runtime: Runtime) -> bool: + state, max_rank = gate + hit = runtime.state == "reached" or (state == "executed" and runtime.state == "executed") + return hit and severity_rank(row.get("severity")) <= max_rank + + +def _http_get_json(url: str, api_key: str) -> dict: + request = urllib.request.Request( # nosec B310 -- the user's own portal URL from `secfoo cloud login` + url, headers={"Authorization": f"Bearer {api_key}", "Accept": "application/json", "User-Agent": "secfoo-cli"} + ) + with urllib.request.urlopen(request, timeout=FETCH_TIMEOUT_SECONDS) as response: # nosec B310 + return json.loads(response.read().decode("utf-8")) diff --git a/tests/test_runtime/test_developer.py b/tests/test_runtime/test_developer.py new file mode 100644 index 0000000..1876476 --- /dev/null +++ b/tests/test_runtime/test_developer.py @@ -0,0 +1,249 @@ +"""#80: runtime evidence in the CLI, SARIF and the CI gate.""" +from __future__ import annotations + +import json +import urllib.error + +import pytest +from typer.testing import CliRunner + +from secfoo.cli import EXIT_RUNTIME_REACHED, _runtime_summary_line, app +from secfoo.cloud import CloudConfig +from secfoo.report.sarif import findings_to_sarif +from secfoo.runtime.developer import RuntimeLookup, parse_gate +from secfoo.runtime.evidence import ingest_events +from secfoo.storage.repository import RunRepository + +runner = CliRunner(env={"COLUMNS": "250"}) +PROJECT = "https://github.com/acme/payments" +# fingerprint -> (package, severity, static reachability, fixed_in) +FINDINGS = { + "fp-reached": ("pyyaml", "High", "Not reachable", "5.4"), + "fp-executed": ("jinja2", "High", "Reachable", "2.10.1"), + "fp-loaded": ("requests", "High", "Reachable", "2.31.0"), + "fp-quiet": ("lxml", "High", "Not reachable", "4.6.3"), + "fp-low": ("pillow", "Low", "Reachable", "8.1.1"), +} + + +def _events(): + s = {"type": "session", "id": "s-1", "language": "python", "source": "sensor", + "started_at": "2026-11-12T00:00:00Z", "ended_at": "2026-11-12T06:00:00Z", + "app_functions_total": 100, "app_functions_executed": 74} + return [ + s, + {"type": "load", "session": "s-1", "ecosystem": "PyPI", "package": "requests", "version": "2.31.0"}, + {"type": "hit", "session": "s-1", "finding_fingerprint": "fp-reached", "symbol": "yaml.load", + "route": "POST /upload", "auth": "anonymous", "count": 312, "last_seen": "2026-11-12T05:00:00Z"}, + {"type": "hit", "session": "s-1", "finding_fingerprint": "fp-executed", "symbol": "jinja2.Template.render", + "route": "startup", "count": 2}, + {"type": "hit", "session": "s-1", "finding_fingerprint": "fp-low", "symbol": "PIL.Image.open", + "route": "GET /avatar", "count": 5}, + ] + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr("secfoo.storage.repository.DB_PATH", tmp_path / "db.sqlite") + monkeypatch.setattr("secfoo.storage.repository.ensure_store_dirs", lambda: None) + monkeypatch.setattr("secfoo.cloud.load_cloud_config", lambda *a, **k: None) # no key unless a test adds one + with RunRepository() as repo: + pid = repo.upsert_project(PROJECT, "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 + for i, (fp, (package, severity, reach, fixed)) in enumerate(FINDINGS.items(), start=1): + repo.upsert_sca_finding(project_id=pid, fingerprint=fp, run_id=run_id, current_ref=f"D{i}", + package=package, severity=severity, version="1.0", ecosystem="PyPI", + reachability=reach, fixed_in=fixed, issue=f"{package} issue") + return pid + + +def _ingest(pid): + with RunRepository() as repo: + assert ingest_events(repo.connection, pid, [json.dumps(e) for e in _events()]).skipped == [] + + +def _states(**kwargs): + with RunRepository() as repo: + lookup = RuntimeLookup(repo, **kwargs) + return {row["fingerprint"]: lookup.for_finding(skill, row) + for skill, row in repo.list_structured_findings(skill_ids=["sca-reachability"])} + + +# ---------- every runtime state ---------- + + +def test_every_state_from_local_evidence(store): + _ingest(store) + states = _states() + assert states["fp-reached"].state == "reached" + assert states["fp-reached"].summary == "Reached · POST /upload (anonymous) · 312" + assert states["fp-executed"].state == "executed" + assert states["fp-loaded"].state == "loaded" and states["fp-loaded"].version_drift # scanned 1.0, running 2.31.0 + assert states["fp-quiet"].state == "not_loaded" + assert "74% coverage" in states["fp-quiet"].summary # never "not reachable" without its coverage + + +def test_no_sensor_means_no_runtime_data_and_no_errors(store): + assert {s.state for s in _states().values()} == {"none"} + assert all(s.summary == "No runtime data" for s in _states().values()) + + +def test_no_key_means_no_network_call(store): + calls = [] + states = _states(fetch=lambda url, key: calls.append(url) or {}) + assert calls == [] and {s.state for s in states.values()} == {"none"} + + +def test_no_runtime_skips_everything(store): + _ingest(store) + calls = [] + states = _states(enabled=False, fetch=lambda u, k: calls.append(u) or {}, + cloud_config=CloudConfig(api_key="sfk_x", portal_url="https://s")) + assert calls == [] and {s.state for s in states.values()} == {"none"} + + +# ---------- server + offline cache ---------- + + +SERVER = {"findings": [{"finding_fingerprint": "fp-quiet", "runtime_state": "reached", + "runtime_summary": "Reached · GET /export (authenticated) · 9", "hits": 9, + "routes": [{"route": "GET /export", "auth": "authenticated", "count": 9}]}]} + + +def test_server_evidence_is_fetched_with_the_key_and_cached_for_offline(store): + config = CloudConfig(api_key="sfk_secret", portal_url="https://secfoo.internal") + seen = [] + + def online(url, key): + seen.append((url, key)) + return SERVER + + states = _states(fetch=online, cloud_config=config) + assert states["fp-quiet"].state == "reached" and states["fp-quiet"].source == "server" + assert seen == [("https://secfoo.internal/api/runtime/v1/evidence?project=https%3A%2F%2Fgithub.com%2Facme%2Fpayments", + "sfk_secret")] + + def offline(url, key): + raise urllib.error.URLError("no route to host") + + with RunRepository() as repo: + lookup = RuntimeLookup(repo, fetch=offline, cloud_config=config) + rows = {row["fingerprint"]: row for _, row in repo.list_structured_findings(skill_ids=["sca-reachability"])} + cached = lookup.for_finding("sca-reachability", rows["fp-quiet"]) + assert cached.state == "reached" and cached.source == "cache" + assert any("cache" in note for note in lookup.notes) + + +def test_stronger_of_local_and_server_wins(store): + _ingest(store) + weaker = {"findings": [{"finding_fingerprint": "fp-reached", "runtime_state": "loaded", "runtime_summary": "Loaded"}]} + states = _states(fetch=lambda u, k: weaker, cloud_config=CloudConfig(api_key="k", portal_url="https://s")) + assert states["fp-reached"].state == "reached" + + +# ---------- CLI: list / show ---------- + + +def test_findings_list_shows_runtime_and_sorts_within_severity(store): + _ingest(store) + result = runner.invoke(app, ["findings", "list"]) + assert result.exit_code == 0, result.output + # Order by position in the output: rich may wrap cells on a narrow CI terminal. + order = sorted(FINDINGS, key=lambda fp: result.output.index(fp[:10])) + assert order == ["fp-reached", "fp-executed", "fp-loaded", "fp-quiet", "fp-low"] # High by runtime, then Low + flat = " ".join(result.output.split()) + assert "pyyaml" in flat and "Not loaded" in flat + + +def test_findings_list_no_runtime(store): + _ingest(store) + result = runner.invoke(app, ["findings", "list", "--no-runtime"]) + assert result.exit_code == 0 and "Reached" not in result.output + + +def test_findings_show_gives_the_remediation_hint(store): + _ingest(store) + result = runner.invoke(app, ["findings", "show", "fp-reached"]) + assert result.exit_code == 0, result.output + flat = " ".join(result.output.split()) # rich wraps long lines on narrow CI terminals + assert "Fix first: upgrade pyyaml to 5.4; this protects live route(s): POST /upload (anonymous)" in flat + + +# ---------- SARIF ---------- + + +def test_sarif_carries_runtime_properties_and_priority(store): + _ingest(store) + result = runner.invoke(app, ["findings", "export", "--format", "sarif", "--skill", "sca-reachability", + "--runtime-priority"]) + assert result.exit_code == 0, result.output + results = json.loads(result.stdout)["runs"][0]["results"] + by_state = {r["properties"]["secfoo"]["runtime"]["state"]: r for r in results} + reached = by_state["reached"] + assert reached["properties"]["secfoo"]["runtime"]["routes"][0]["route"] == "POST /upload" + assert "Reached in production via POST /upload" in reached["message"]["text"] + assert reached["level"] == "error" + assert {"reached", "executed", "loaded", "not_loaded"} <= set(by_state) + + +def test_runtime_priority_only_raises_reached_findings(): + records = [{"skill_id": "sca-reachability", "severity": "Low", "issue": "x", "runtime": {"state": "reached", "routes": []}}, + {"skill_id": "sca-reachability", "severity": "Low", "issue": "y", "runtime": {"state": "loaded"}}] + levels = [r["level"] for r in findings_to_sarif(records, runtime_priority=True)["runs"][0]["results"]] + assert levels == ["error", "note"] + assert [r["level"] for r in findings_to_sarif(records)["runs"][0]["results"]] == ["note", "note"] + + +# ---------- CI gate ---------- + + +def test_gate_trips_with_its_own_exit_code(store): + _ingest(store) + result = runner.invoke(app, ["findings", "check", "--fail-on", "reached:high"]) + assert result.exit_code == EXIT_RUNTIME_REACHED == 4 + assert "pyyaml" in result.output and "pillow" not in result.output # Low is below the threshold + + +def test_gate_executed_level_and_severity(store): + _ingest(store) + assert runner.invoke(app, ["findings", "check", "--fail-on", "reached:critical"]).exit_code == 0 + tripped = runner.invoke(app, ["findings", "check", "--fail-on", "executed:high"]) + assert tripped.exit_code == 4 and "jinja2" in tripped.output + assert runner.invoke(app, ["findings", "check", "--fail-on", "reached"]).exit_code == 4 # any severity + + +def test_gate_passes_without_runtime_data(store): + assert runner.invoke(app, ["findings", "check", "--fail-on", "reached"]).exit_code == 0 + + +@pytest.mark.parametrize("spec", ["loaded", "reached:urgent", ""]) +def test_gate_spec_is_validated(spec): + with pytest.raises(ValueError): + parse_gate(spec) + + +# ---------- after secfoo run ---------- + + +def test_run_summary_line(store): + _ingest(store) + + class Outcome: + def __init__(self, uuid): + self.run_uuid = uuid + + with RunRepository() as repo: + uuid = repo.connection.execute("SELECT run_uuid FROM runs").fetchone()[0] + line = _runtime_summary_line([Outcome(uuid)]) + assert "2 of 5 findings are reached in production" in line and "D1 (POST /upload, anonymous)" in line + + +def test_run_summary_is_silent_without_runtime_data(store): + with RunRepository() as repo: + uuid = repo.connection.execute("SELECT run_uuid FROM runs").fetchone()[0] + + class Outcome: + run_uuid = uuid + + assert _runtime_summary_line([Outcome()]) is None