From afc09fbb8fb1d0e5a2f6e2c86a2710a7685fd468 Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 04:33:40 +0000 Subject: [PATCH 1/2] feat(runtime): runtime package inventory, drift, unscanned deps and live advisories (#68) secfoo runtime inventory compares what production loaded (load lines) with the latest SCA scan: Loaded at the scanned version, drift (running version checked in OSV), unscanned dependencies (OSV-checked, raised as runtime findings) and never-loaded vulnerable packages with the observation window; flags packages new after a baseline. secfoo runtime recheck (and a 6-hourly job in secfoo serve) queries OSV querybatch for every running version so a newly published advisory raises a finding with no new scan. --format cyclonedx writes a runtime SBOM. Closes #68 Co-Authored-By: Claude Opus 5.5 --- docs/REFERENCE.md | 33 +++ src/secfoo/cli.py | 114 +++++++++++ src/secfoo/runtime/inventory.py | 290 +++++++++++++++++++++++++++ src/secfoo/storage/schema.sql | 20 ++ tests/test_runtime/test_inventory.py | 180 +++++++++++++++++ 5 files changed, 637 insertions(+) create mode 100644 src/secfoo/runtime/inventory.py create mode 100644 tests/test_runtime/test_inventory.py diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index f126f50..bb44ea8 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -473,6 +473,39 @@ runtime-confirmed findings, but only once there is runtime data. ![SCA vulnerabilities list with the Runtime column and a highlighted conflict](assets/sca-runtime.jpg) +### Runtime package inventory + +Probes report which packages a process actually **loaded**, and at which +version (`load` lines). `secfoo runtime inventory` compares that with the +latest SCA scan: + +```bash +secfoo runtime inventory --project payments # table +secfoo runtime inventory --project payments --app payments-api --env prod --format cyclonedx -o runtime-sbom.json +secfoo runtime recheck # OSV querybatch for every running version +``` + +| Case | Result | +|---|---| +| A vulnerable package is loaded at the scanned version | The SCA finding shows **Loaded**. | +| A package is loaded at a *different* version than scanned | **Drift.** That version is checked against OSV, and an advisory raises a runtime finding. | +| A package is loaded but no scanned manifest lists it | **Unscanned dependency.** OSV is checked, and an advisory raises a runtime finding. | +| A vulnerable package is never loaded | **Not loaded**, shown with the observation window (it is not proof of unreachability). | + +- **New in production.** A package first seen more than `--baseline-hours` + (default 24) after its app's first session is marked **new**, even with no + known advisory. +- **Real-time advisories.** `secfoo runtime recheck` asks OSV + (`/v1/querybatch`) about every package version that is running now. An + advisory the scans don't already account for raises a `new_advisory` + runtime finding, with no new scan. `secfoo serve` runs the same recheck in + the background every `--runtime-recheck-hours` (default 6; `0` disables + it). It only queries when runtime data exists. +- **Storage.** Runtime findings live in `runtime_findings` (kind, package, + version, advisory, and the app/env running it). +- **SBOM.** `--format cyclonedx` writes a CycloneDX 1.5 SBOM of what is + actually loaded, with purls and each package's runtime status. + ### Coverage files as runtime evidence The easiest evidence needs no sensor: coverage from your test suite or a diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index c75187a..056eeb0 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -669,6 +669,10 @@ def serve( "--allow-insecure-runtime", help="Accept runtime API keys over plain HTTP on a non-loopback --host. Prefer TLS in front instead.", )] = False, + runtime_recheck_hours: Annotated[float, typer.Option( + "--runtime-recheck-hours", + help="Re-check running package versions against OSV this often (#68). 0 disables.", + )] = 6.0, ) -> None: """Launch the local web dashboard.""" if ctx.invoked_subcommand is not None: @@ -677,10 +681,31 @@ def serve( from secfoo.web.app import create_app + if runtime_recheck_hours > 0: + _start_runtime_recheck(runtime_recheck_hours) console.print(f"Serving dashboard at [bold]http://{host}:{port}[/]") uvicorn.run(create_app(bind_host=host, allow_insecure_runtime=allow_insecure_runtime), host=host, port=port) +def _start_runtime_recheck(hours: float) -> None: + """Background OSV re-check of running versions (#68). Only does anything + once runtime data exists; never stops the server.""" + import threading + + def loop() -> None: + from secfoo.runtime.inventory import recheck + + stop = threading.Event() + while not stop.wait(hours * 3600): + try: + with RunRepository() as repo: + recheck(repo) + except Exception: # noqa: BLE001, S112 -- OSV down or a bad row: try again next interval + continue + + threading.Thread(target=loop, name="secfoo-runtime-recheck", daemon=True).start() + + @serve_key_app.command(name="create") def serve_key_create( name: Annotated[str, typer.Option("--name", help="A label, e.g. payments-prod-sensor.")], @@ -1858,6 +1883,95 @@ def runtime_import_coverage( err_console.print(f"{cov.language} coverage: {hits} watched symbol(s) executed" + (f". Wrote {output}." if output else ".")) +class InventoryFormat(str, Enum): + TABLE = "table" + JSON = "json" + CYCLONEDX = "cyclonedx" + + +@runtime_app.command(name="inventory") +def runtime_inventory( + project: Annotated[str, typer.Option("--project", help="Project id, name, or path/URL it was scanned as.")], + app_name: Annotated[str | None, typer.Option("--app", help="Only this app (SECFOO_APP).")] = None, + env: Annotated[str | None, typer.Option("--env", help="Only this environment (SECFOO_ENV).")] = None, + format: Annotated[InventoryFormat, typer.Option("--format")] = InventoryFormat.TABLE, + output: Annotated[Path | None, typer.Option("--output", "-o")] = None, + baseline_hours: Annotated[float, typer.Option( + "--baseline-hours", help="Packages first seen after this long into an app's observation are flagged new.", + )] = 24.0, +) -> None: + """What production actually loads, compared with the scan: version drift, + unscanned dependencies, packages new in production, and vulnerable packages + never loaded. --format cyclonedx writes a runtime SBOM.""" + from secfoo.runtime.inventory import analyze, cyclonedx + + 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}.") + raise typer.Exit(code=1) + identifier = repo.get_project(project_id).identifier + report = analyze(repo, project_id, app=app_name, env=env, baseline_hours=baseline_hours) + if format is InventoryFormat.CYCLONEDX: + text = json.dumps(cyclonedx(report, project=identifier, app=app_name, env=env), indent=2) + "\n" + elif format is InventoryFormat.JSON: + text = json.dumps({ + "packages": [{"ecosystem": p.ecosystem, "package": p.package, "version": p.version, "status": p.status, + "scanned_versions": p.scanned_versions, "advisories": p.advisories, "new": p.new, + "apps": sorted(p.apps), "first_seen": p.first_seen} for p in report.packages], + "not_loaded": report.not_loaded, "observed_hours": round(report.observed_hours, 2), + "raised": report.raised, + }, indent=2) + "\n" + else: + table = Table(show_header=True, header_style="bold") + for column in ("package", "version", "status", "advisories", "apps"): + table.add_column(column, overflow="fold") + for p in report.packages: + status = p.status + (" (scanned " + ", ".join(p.scanned_versions) + ")" if p.status == "drift" else "") + table.add_row(p.package, p.version, status + (" · new" if p.new else ""), ", ".join(p.advisories) or "-", + ", ".join(sorted(p.apps))) + console.print(table) + if report.not_loaded: + console.print(f"Not loaded in {report.observed_hours:.1f}h of observation (not proof of unreachability): " + + ", ".join(f"{f['package']}@{f['version']}" for f in report.not_loaded)) + if report.raised: + console.print(f"[bold]{len(report.raised)} runtime finding(s):[/] " + + ", ".join(f"{r['kind']} {r['package']}@{r['version']} {r['advisory']}" for r in report.raised)) + return + if output is None: + sys.stdout.write(text) + else: + output.write_text(text, encoding="utf-8") + err_console.print(f"Wrote {output}.") + + +@runtime_app.command(name="recheck") +def runtime_recheck( + project: Annotated[str | None, typer.Option("--project", help="One project (default: all with runtime data).")] = None, +) -> None: + """Ask OSV about every package version running right now; a new advisory raises a finding with no new scan.""" + from secfoo.osv import OsvError + from secfoo.runtime.inventory import recheck + + with RunRepository() as repo: + project_id = None + if project is not None: + project_id = repo.resolve_project_ref(project) + if project_id is None: + err_console.print(f"[red]Error:[/] no single project matches {project!r}.") + raise typer.Exit(code=1) + try: + raised = recheck(repo, project_id=project_id) + except OsvError as exc: + err_console.print(f"[red]Error:[/] {exc}") + raise typer.Exit(code=1) from None + if not raised: + console.print("No advisories against running versions beyond what the scans already report.") + return + for r in raised: + console.print(f"[red]{r['kind']}[/] {r['package']}@{r['version']} {r['advisory']} running in {', '.join(r['apps'])}") + + @runtime_app.command(name="watchlist") def runtime_watchlist( project: Annotated[str, typer.Option("--project", help="Project id, name, or path/URL it was scanned as.")], diff --git a/src/secfoo/runtime/inventory.py b/src/secfoo/runtime/inventory.py new file mode 100644 index 0000000..912689e --- /dev/null +++ b/src/secfoo/runtime/inventory.py @@ -0,0 +1,290 @@ +"""Runtime package inventory (#68): what is loaded in production, at which +version, compared with what was scanned. + + | Case | Result | + |----------------------------------------------|-------------------------------------------| + | vulnerable package loaded at scanned version | finding shows Loaded (web/sca_runtime.py) | + | loaded at a different version than scanned | drift: that version is checked in OSV | + | loaded but absent from scanned manifests | unscanned: OSV checked, finding raised | + | vulnerable package never loaded | Not loaded, with the observation window | + +Plus "new in production" (first seen after a sensor's baseline window), +`recheck()` -- OSV querybatch for every running version, so a newly +published advisory raises a finding with no new scan -- and a CycloneDX +runtime SBOM. Findings raised here live in `runtime_findings`. +""" + +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from collections import defaultdict +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from secfoo import osv +from secfoo.report.sca import SKILL_ID, inventory_lookup + +DEFAULT_BASELINE_HOURS = 24 +OSV_BATCH_URL = "https://api.osv.dev/v1/querybatch" +OSV_BATCH_SIZE = 1000 + +QueryBatch = Callable[[list[tuple[str, str, str]]], list[list[dict]]] + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _parse(ts: str | None) -> datetime | None: + if not ts: + return None + try: + value = datetime.fromisoformat(ts[:-1] + "+00:00" if ts.endswith("Z") else ts) + except ValueError: + return None + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + + +@dataclass +class LoadedPackage: + ecosystem: str + package: str + version: str + apps: set[str] = field(default_factory=set) # "app/env" + first_seen: str | None = None + status: str = "scanned" # scanned / drift / unscanned + scanned_versions: list[str] = field(default_factory=list) + advisories: list[str] = field(default_factory=list) + new: bool = False + + +@dataclass +class InventoryReport: + packages: list[LoadedPackage] = field(default_factory=list) + not_loaded: list[dict] = field(default_factory=list) # open SCA findings never loaded + observed_hours: float = 0.0 + raised: list[dict] = field(default_factory=list) # runtime findings raised/refreshed now + + +def _scanned(repo, project_id: int) -> dict[tuple[str, str], set[str]]: + """(OSV ecosystem, package lower) -> versions the latest SCA scan saw.""" + scanned: dict[tuple[str, str], set[str]] = defaultdict(set) + for project in repo.activity_summary_by_project(SKILL_ID): + if project["project_id"] != project_id or not project.get("latest_success_report_path"): + continue + path = Path(project["latest_success_report_path"]) + if not path.is_file(): + continue + for (package, version), ecosystem_raw in inventory_lookup(path.read_text(encoding="utf-8")).items(): + ecosystem = osv.normalize_ecosystem(ecosystem_raw) + if ecosystem: + scanned[(ecosystem, package.lower())].add(version) + for finding in repo.list_sca_findings(project_id=project_id): + ecosystem = osv.normalize_ecosystem(finding.ecosystem) or finding.ecosystem + if ecosystem and finding.version: + scanned[(ecosystem, finding.package.lower())].add(finding.version) + return scanned + + +def _loaded(conn, project_id: int, *, app: str | None, env: str | None) -> tuple[dict, dict]: + rows = conn.execute( + "SELECT p.ecosystem, p.package, p.version, s.app, s.env, COALESCE(p.first_seen, s.started_at), s.started_at " + "FROM runtime_packages p JOIN runtime_sessions s ON s.id = p.session_id WHERE p.project_id = ?" + " AND (? IS NULL OR s.app = ?) AND (? IS NULL OR s.env = ?)", + (project_id, app, app, env, env), + ).fetchall() + loaded: dict[tuple[str, str, str], LoadedPackage] = {} + app_start: dict[str, datetime] = {} + for ecosystem, package, version, s_app, s_env, first_seen, started in rows: + label = f"{s_app or '-'}/{s_env or '-'}" + start = _parse(started) + if start and (label not in app_start or start < app_start[label]): + app_start[label] = start + key = (ecosystem, package.lower(), version) + item = loaded.setdefault(key, LoadedPackage(ecosystem, package, version)) + item.apps.add(label) + if first_seen and (item.first_seen is None or first_seen < item.first_seen): + item.first_seen = first_seen + return loaded, app_start + + +def _advisory_ids(vulns: list[dict]) -> list[str]: + ids = [] + for vuln in vulns: + ids.append(next((a for a in sorted(vuln.get("aliases") or []) if a.startswith("GHSA-")), vuln.get("id", ""))) + return sorted({i for i in ids if i}) + + +def _known_advisories(repo, project_id: int) -> dict[tuple[str, str], set[str]]: + known: dict[tuple[str, str], set[str]] = defaultdict(set) + lookups = repo.osv_lookup_map() + for finding in repo.list_sca_findings(project_id=project_id): + ecosystem = osv.normalize_ecosystem(finding.ecosystem) or finding.ecosystem or "" + key = (ecosystem, finding.package.lower()) + if finding.cve: + known[key].add(finding.cve.strip()) + record = lookups.get((ecosystem, finding.package, finding.version or "")) + for vuln in record.vulns if record else []: + known[key] |= {vuln.get("id", ""), *(vuln.get("aliases") or [])} + return known + + +def _raise(conn, project_id: int, kind: str, item: LoadedPackage, advisory: str) -> dict: + now = _now() + conn.execute( + """ + INSERT INTO runtime_findings (project_id, kind, ecosystem, package, version, advisory, scanned_versions, apps, + first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (project_id, kind, ecosystem, package, version, advisory) DO UPDATE SET + scanned_versions = excluded.scanned_versions, apps = excluded.apps, last_seen_at = excluded.last_seen_at + """, + (project_id, kind, item.ecosystem, item.package, item.version, advisory, json.dumps(item.scanned_versions), + json.dumps(sorted(item.apps)), now, now), + ) + return {"kind": kind, "ecosystem": item.ecosystem, "package": item.package, "version": item.version, + "advisory": advisory, "apps": sorted(item.apps), "scanned_versions": item.scanned_versions} + + +def analyze(repo, project_id: int, *, app: str | None = None, env: str | None = None, + baseline_hours: float = DEFAULT_BASELINE_HOURS, + known: dict[tuple[str, str], set[str]] | None = None) -> InventoryReport: + """Compare the loaded inventory with the scan; raise/refresh runtime findings. + + `known` is the advisories the scan already accounts for, per (ecosystem, + package). recheck() snapshots it BEFORE refreshing the OSV cache, so an + advisory OSV publishes later counts as new rather than as known. + """ + conn = repo.connection + report = InventoryReport() + loaded, app_start = _loaded(conn, project_id, app=app, env=env) + scanned = _scanned(repo, project_id) + known = _known_advisories(repo, project_id) if known is None else known + lookups = repo.osv_lookup_map() + + for item in loaded.values(): + versions = scanned.get((item.ecosystem, item.package.lower())) + record = lookups.get((item.ecosystem, item.package, item.version)) or lookups.get( + (item.ecosystem, item.package.lower(), item.version)) + item.advisories = _advisory_ids(record.vulns) if record else [] + if versions is None: + item.status = "unscanned" + for advisory in item.advisories: + report.raised.append(_raise(conn, project_id, "unscanned", item, advisory)) + elif item.version not in versions: + item.status, item.scanned_versions = "drift", sorted(versions) + for advisory in item.advisories: + report.raised.append(_raise(conn, project_id, "drift", item, advisory)) + else: + item.status, item.scanned_versions = "scanned", sorted(versions) + for advisory in item.advisories: + if advisory not in known.get((item.ecosystem, item.package.lower()), set()): + report.raised.append(_raise(conn, project_id, "new_advisory", item, advisory)) + # New in production: first seen after its app's baseline window. + seen = _parse(item.first_seen) + for label in item.apps: + start = app_start.get(label) + if seen and start and seen > start + timedelta(hours=baseline_hours): + item.new = True + conn.commit() + + loaded_names = {(e, p) for (e, p, _v) in loaded} + for finding in repo.list_sca_findings(project_id=project_id, status="open"): + ecosystem = osv.normalize_ecosystem(finding.ecosystem) or finding.ecosystem + if (ecosystem, finding.package.lower()) not in loaded_names: + report.not_loaded.append({"package": finding.package, "version": finding.version, + "fingerprint": finding.fingerprint, "severity": finding.severity}) + if app_start: + rows = conn.execute("SELECT started_at, ended_at FROM runtime_sessions WHERE project_id = ?", (project_id,)) + for started, ended in rows.fetchall(): + start, end = _parse(started), _parse(ended) + if start and end and end > start: + report.observed_hours += (end - start).total_seconds() / 3600 + report.packages = sorted(loaded.values(), key=lambda p: ({"unscanned": 0, "drift": 1, "scanned": 2}[p.status], + p.package.lower(), p.version)) + return report + + +# ---------- real-time advisories ---------- + + +def osv_querybatch(keys: list[tuple[str, str, str]]) -> list[list[dict]]: + """OSV /v1/querybatch: one list of {id, modified} per (ecosystem, package, version).""" + results: list[list[dict]] = [] + for start in range(0, len(keys), OSV_BATCH_SIZE): + chunk = keys[start:start + OSV_BATCH_SIZE] + body = json.dumps({"queries": [{"package": {"ecosystem": e, "name": p}, "version": v} for e, p, v in chunk]}) + request = urllib.request.Request( # nosec B310 -- fixed public OSV.dev URL + OSV_BATCH_URL, data=body.encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(request, timeout=osv.REQUEST_TIMEOUT_SECONDS * 4) as response: # nosec B310 + data = json.loads(response.read().decode("utf-8")) + except (urllib.error.URLError, OSError) as exc: + raise osv.OsvError(f"Could not reach OSV.dev: {exc}") from exc + results.extend(r.get("vulns", []) for r in data.get("results", [])) + return results + + +def recheck(repo, *, project_id: int | None = None, query_batch: QueryBatch = osv_querybatch) -> list[dict]: + """Query OSV for every package version currently loaded (in one project or + all), refresh the cache, and return the runtime findings raised.""" + conn = repo.connection + where, params = ("WHERE project_id = ?", (project_id,)) if project_id is not None else ("", ()) + keys = sorted({tuple(r) for r in conn.execute( + f"SELECT DISTINCT ecosystem, package, version FROM runtime_packages {where}", params)}) # nosec B608 + if not keys: + return [] + projects = [project_id] if project_id is not None else [ + r[0] for r in conn.execute("SELECT DISTINCT project_id FROM runtime_packages").fetchall()] + known_before = {pid: _known_advisories(repo, pid) for pid in projects} + for key, vulns in zip(keys, query_batch(keys)): + repo.upsert_osv_lookup(ecosystem=key[0], package=key[1], version=key[2], vulns=vulns) + raised: list[dict] = [] + for pid in projects: + raised.extend(analyze(repo, pid, known=known_before[pid]).raised) + return raised + + +# ---------- CycloneDX runtime SBOM ---------- + +_PURL_TYPES = {"PyPI": "pypi", "npm": "npm", "Maven": "maven", "NuGet": "nuget", "Go": "golang", + "crates.io": "cargo", "RubyGems": "gem", "Packagist": "composer"} + + +def _purl(item: LoadedPackage) -> str | None: + kind = _PURL_TYPES.get(item.ecosystem) + if kind is None: + return None + name = item.package + if kind == "pypi": + name = name.lower().replace("_", "-") + elif kind == "npm" and name.startswith("@"): + name = "%40" + name[1:] + elif kind == "maven" and ":" in name: + name = name.replace(":", "/", 1) + return f"pkg:{kind}/{name}@{item.version}" + + +def cyclonedx(report: InventoryReport, *, project: str, app: str | None, env: str | None) -> dict: + components = [] + for item in report.packages: + component = {"type": "library", "name": item.package, "version": item.version, + "properties": [{"name": "secfoo:runtime:status", "value": item.status}, + {"name": "secfoo:runtime:apps", "value": ",".join(sorted(item.apps))}]} + purl = _purl(item) + if purl: + component["purl"] = purl + component["bom-ref"] = purl + components.append(component) + return { + "bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1, + "metadata": {"timestamp": _now(), "tools": {"components": [{"type": "application", "name": "secfoo"}]}, + "component": {"type": "application", "name": app or project, + "properties": [{"name": "secfoo:project", "value": project}, + {"name": "secfoo:env", "value": env or "all"}]}}, + "components": components, + } diff --git a/src/secfoo/storage/schema.sql b/src/secfoo/storage/schema.sql index c1b160d..8c00400 100644 --- a/src/secfoo/storage/schema.sql +++ b/src/secfoo/storage/schema.sql @@ -408,3 +408,23 @@ CREATE TABLE IF NOT EXISTS runtime_sensors ( last_heartbeat_at TEXT NOT NULL, UNIQUE (project_id, sensor_id) ); + +-- #68: findings raised from what production actually loaded, not from a +-- scan: an advisory against a version that differs from the scanned one +-- (drift), against a package no scanned manifest listed (unscanned), or a +-- new advisory published against a version that is running now. Not in +-- sca_findings because no scan run produced them. +CREATE TABLE IF NOT EXISTS runtime_findings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id INTEGER NOT NULL REFERENCES projects(id), + kind TEXT NOT NULL CHECK (kind IN ('drift','unscanned','new_advisory')), + ecosystem TEXT NOT NULL, + package TEXT NOT NULL, + version TEXT NOT NULL, + advisory TEXT NOT NULL, + scanned_versions TEXT, -- JSON list; empty for unscanned + apps TEXT, -- JSON list of "app/env" running it + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE (project_id, kind, ecosystem, package, version, advisory) +); diff --git a/tests/test_runtime/test_inventory.py b/tests/test_runtime/test_inventory.py new file mode 100644 index 0000000..48fe9c8 --- /dev/null +++ b/tests/test_runtime/test_inventory.py @@ -0,0 +1,180 @@ +"""#68: runtime package inventory -- drift, unscanned dependencies, never loaded, real-time advisories.""" +from __future__ import annotations + +import json + +import pytest +from typer.testing import CliRunner + +from secfoo.cli import app +from secfoo.report.sca import findings_with_fingerprints +from secfoo.runtime.evidence import ingest_events +from secfoo.runtime.inventory import analyze, cyclonedx, recheck +from secfoo.storage.repository import RunRepository +from secfoo.web import sca_runtime + +REPORT = """\ +# SCA Report — Reachability & Upgrade Triage + +## 1. Executive Summary +Test. **Overall risk rating:** High + +## 2. Dependency Inventory + +| Package | Version | Direct/Transitive | Ecosystem | Source manifest | +|---------|---------|--------------------|-----------|------------------| +| pyyaml | 5.3 | Direct | Python | requirements.txt | +| requests | 2.19.0 | Direct | Python | requirements.txt | +| lxml | 4.6.2 | Direct | Python | requirements.txt | + +## 4. Risk Register + +| ID | Package | Version | Issue | Severity | Reachability | Fixed in | +|----|---------|---------|-------|----------|---------------|----------| +| D1 | pyyaml | 5.3 | Code execution in full_load | High | Not reachable | 5.4 | +| D2 | lxml | 4.6.2 | XSS in clean | Medium | Reachable | 4.6.3 | + +## 5. Detailed Findings + +### [HIGH] D1: pyyaml +- **Package / version:** pyyaml 5.3, **CVE:** GHSA-8q59-q68h-6hv4, **Severity:** High +### [MEDIUM] D2: lxml +- **Package / version:** lxml 4.6.2, **CVE:** unverified, **Severity:** Medium +""" + +EVENTS = [ + {"type": "session", "id": "s-1", "language": "python", "source": "sensor", "app": "payments-api", "env": "prod", + "started_at": "2026-11-10T00:00:00Z", "ended_at": "2026-11-12T00:00:00Z"}, + {"type": "load", "session": "s-1", "ecosystem": "PyPI", "package": "pyyaml", "version": "5.3", + "first_seen": "2026-11-10T00:00:01Z"}, # case 1: scanned version + {"type": "load", "session": "s-1", "ecosystem": "PyPI", "package": "requests", "version": "2.31.0", + "first_seen": "2026-11-10T00:00:01Z"}, # case 2: drift + {"type": "load", "session": "s-1", "ecosystem": "PyPI", "package": "left-pad-py", "version": "1.0", + "first_seen": "2026-11-11T06:00:00Z"}, # case 3: unscanned, and new after the 24h baseline +] +# Fixture OSV responses (already in the cache, as the dashboard/runner would leave them). +OSV = { + ("PyPI", "pyyaml", "5.3"): [{"id": "GHSA-8q59-q68h-6hv4"}], + ("PyPI", "requests", "2.31.0"): [{"id": "PYSEC-2024-1", "aliases": ["GHSA-drft-0000-0000"]}], + ("PyPI", "left-pad-py", "1.0"): [{"id": "GHSA-unsc-0000-0000"}], +} + + +@pytest.fixture +def project(tmp_path, monkeypatch): + monkeypatch.setattr("secfoo.storage.repository.DB_PATH", tmp_path / "db.sqlite") + monkeypatch.setattr("secfoo.storage.repository.ensure_store_dirs", lambda: None) + 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") + 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 / "sca.md" + path.write_text(REPORT, encoding="utf-8") + repo.complete_run(uuid, status="success", exit_code=0, duration_seconds=1.0, report_path=str(path), + prompt_path=None, stderr_excerpt=None, high_count=1) + run_id = repo.get_run(uuid).id + for row in findings_with_fingerprints(REPORT): + repo.upsert_sca_finding(project_id=pid, fingerprint=row["fingerprint"], run_id=run_id, + current_ref=row["id"], package=row["package"], severity=row["severity"], + version=row["version"], ecosystem="PyPI", reachability=row["reachability"], + cve=row["cve"]) + for (eco, pkg, ver), vulns in OSV.items(): + repo.upsert_osv_lookup(ecosystem=eco, package=pkg, version=ver, vulns=vulns) + assert ingest_events(repo.connection, pid, [json.dumps(e) for e in EVENTS]).skipped == [] + return pid + + +def _by_name(report): + return {p.package: p for p in report.packages} + + +def test_case_1_vulnerable_package_loaded_at_the_scanned_version(project): + with RunRepository() as repo: + report = analyze(repo, project) + pyyaml = _by_name(report)["pyyaml"] + assert pyyaml.status == "scanned" and not pyyaml.new + # The SCA finding shows Loaded, and its known advisory raises nothing new. + finding = next(f for f in repo.list_sca_findings(project_id=project) if f.package == "pyyaml") + status = sca_runtime.status_for(repo.connection, project_id=project, fingerprint=finding.fingerprint, + package="pyyaml", ecosystem="PyPI", scanned_version="5.3") + assert status.state == "loaded" and not status.version_drift + assert not [r for r in report.raised if r["package"] == "pyyaml"] + + +def test_case_2_version_drift_is_checked_against_osv(project): + with RunRepository() as repo: + report = analyze(repo, project) + requests = _by_name(report)["requests"] + assert requests.status == "drift" and requests.scanned_versions == ["2.19.0"] + assert requests.advisories == ["GHSA-drft-0000-0000"] # GHSA alias preferred + [raised] = [r for r in report.raised if r["package"] == "requests"] + assert raised["kind"] == "drift" and raised["apps"] == ["payments-api/prod"] + + +def test_case_3_unscanned_dependency_raises_a_runtime_finding(project): + with RunRepository() as repo: + report = analyze(repo, project) + rows = [tuple(r) for r in repo.connection.execute( + "SELECT kind, package, version, advisory FROM runtime_findings ORDER BY kind")] + leftpad = _by_name(report)["left-pad-py"] + assert leftpad.status == "unscanned" and leftpad.new # first seen 30h in, after the 24h baseline + assert ("unscanned", "left-pad-py", "1.0", "GHSA-unsc-0000-0000") in rows + + +def test_case_4_vulnerable_package_never_loaded_shows_the_window(project): + with RunRepository() as repo: + report = analyze(repo, project) + assert [f["package"] for f in report.not_loaded] == ["lxml"] + assert report.observed_hours == 48 + + +def test_analysis_is_idempotent(project): + with RunRepository() as repo: + analyze(repo, project) + analyze(repo, project) + assert repo.connection.execute("SELECT COUNT(*) FROM runtime_findings").fetchone()[0] == 2 + + +def test_recheck_finds_a_newly_published_advisory_against_a_running_version(project): + def query_batch(keys): + published = {("PyPI", "pyyaml", "5.3"): [{"id": "GHSA-8q59-q68h-6hv4"}, {"id": "GHSA-new0-0000-0000"}]} + return [published.get(k, OSV.get(k, [])) for k in keys] + + with RunRepository() as repo: + raised = recheck(repo, project_id=project, query_batch=query_batch) + cached = repo.osv_lookup_map()[("PyPI", "pyyaml", "5.3")].vulns + new = [r for r in raised if r["kind"] == "new_advisory"] + assert [(r["package"], r["advisory"]) for r in new] == [("pyyaml", "GHSA-new0-0000-0000")] + assert {"id": "GHSA-new0-0000-0000"} in cached + + +def test_recheck_without_runtime_data_makes_no_request(tmp_path, monkeypatch): + monkeypatch.setattr("secfoo.storage.repository.DB_PATH", tmp_path / "db.sqlite") + monkeypatch.setattr("secfoo.storage.repository.ensure_store_dirs", lambda: None) + with RunRepository() as repo: + assert recheck(repo, query_batch=lambda keys: pytest.fail("no runtime data, no OSV call")) == [] + + +def test_cyclonedx_runtime_sbom(project): + with RunRepository() as repo: + bom = cyclonedx(analyze(repo, project, app="payments-api", env="prod"), + project="https://github.com/acme/payments", app="payments-api", env="prod") + assert bom["bomFormat"] == "CycloneDX" and bom["specVersion"] == "1.5" + purls = {c["purl"] for c in bom["components"]} + assert {"pkg:pypi/pyyaml@5.3", "pkg:pypi/requests@2.31.0", "pkg:pypi/left-pad-py@1.0"} == purls + statuses = {c["name"]: c["properties"][0]["value"] for c in bom["components"]} + assert statuses == {"pyyaml": "scanned", "requests": "drift", "left-pad-py": "unscanned"} + + +def test_cli_inventory_table_json_and_cyclonedx(project, tmp_path): + runner = CliRunner(env={"COLUMNS": "250"}) + table = runner.invoke(app, ["runtime", "inventory", "--project", "payments"]) + assert table.exit_code == 0, table.output + flat = " ".join(table.output.split()) + assert "drift" in flat and "2.19.0" in flat and "unscanned" in flat and "Not loaded" in flat + as_json = json.loads(runner.invoke(app, ["runtime", "inventory", "--project", "payments", "--format", "json"]).stdout) + assert {p["package"]: p["status"] for p in as_json["packages"]}["requests"] == "drift" + out = tmp_path / "sbom.json" + sbom = runner.invoke(app, ["runtime", "inventory", "--project", "payments", "--format", "cyclonedx", "-o", str(out)]) + assert sbom.exit_code == 0 and json.loads(out.read_text())["bomFormat"] == "CycloneDX" From 284f1fff6d077d7de6a3cf41efdd9987a69d377a Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 14:52:47 +0000 Subject: [PATCH 2/2] test(runtime): keep the conformance-kit test inside CI time on py3.10 The setprofile fallback (3.10/3.11) made the kit's overhead loop exceed the 60 s per-step timeout on a slow runner. Use a smaller loop in the test and give each kit step 180 s. Co-Authored-By: Claude Opus 5.5 --- src/secfoo/runtime/conformance/kit.py | 2 +- tests/test_runtime/test_protocol.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/secfoo/runtime/conformance/kit.py b/src/secfoo/runtime/conformance/kit.py index 4f03fd8..bb0071c 100644 --- a/src/secfoo/runtime/conformance/kit.py +++ b/src/secfoo/runtime/conformance/kit.py @@ -37,7 +37,7 @@ DEFAULT_SYMBOL = "secfoo.runtime.conformance.vulnerable.parse" FINGERPRINT = "conformance-fp" -STEP_TIMEOUT_SECONDS = 60 +STEP_TIMEOUT_SECONDS = 180 # the setprofile fallback (3.10/3.11) on a slow CI runner class FakeSensor: diff --git a/tests/test_runtime/test_protocol.py b/tests/test_runtime/test_protocol.py index 5d0dfb0..09838ae 100644 --- a/tests/test_runtime/test_protocol.py +++ b/tests/test_runtime/test_protocol.py @@ -146,7 +146,7 @@ class Stub(transport_mod.NullTransport): def test_reference_python_probe_passes_the_kit(): - results = {r.name: r for r in run_kit(REFERENCE, loop_calls=50000)} + results = {r.name: r for r in run_kit(REFERENCE, loop_calls=10000)} failed = {name: r.detail for name, r in results.items() if not r.passed} assert not failed, failed assert set(results) == {"inert", "call", "nocall", "privacy", "fail-open", "overhead"}