diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 7b587eb..e601b4f 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -401,6 +401,52 @@ file may add hits to a session stored earlier. The JSON Schema is app the observed traffic covered. "Not observed" is always shown with this coverage, never as "not reachable". +**Watchlist.** Probes need to know which functions to watch. Build the list +from a project's open SCA findings: + +```bash +secfoo runtime watchlist --project payments -o watchlist.json +secfoo runtime watchlist --project payments --ghsa -o watchlist.json # also ask GitHub's advisory API +``` + +Symbols come from the OSV records SCA runs already cached, so there are no +network calls by default: + +- Go advisories' `ecosystem_specific.imports[].symbols`, joined with the + import path (`golang.org/x/net/html.Parse`). +- RustSec's `affects.functions`. + +`--ghsa` adds GitHub's `vulnerable_functions` when OSV has none. That API is +unauthenticated and rate-limited, so the build stops asking at the first +limit. A finding that names an advisory gets only that advisory; an +`unverified` one gets every advisory OSV has for its package and version. +When no symbols are known, the entry stays in the list at +`"granularity": "package"` and the probe records whether the package was +loaded. Entries are never dropped. Schema: +[`src/secfoo/runtime/watchlist.schema.json`](../src/secfoo/runtime/watchlist.schema.json). + +```json +{"version": 1, "generated_at": "2026-11-08T12:00:00Z", "entries": [ + {"finding_fingerprint": "", "advisory": "GHSA-vvpx-j8f3-3w6h", "ecosystem": "Go", + "package": "golang.org/x/net", "version": "0.6.0", "granularity": "symbol", + "symbols": ["golang.org/x/net/http2/hpack.Decoder.DecodeFull"], "source": "osv"} +]} +``` + +Symbols use one notation per ecosystem, and probes in every runtime match +these exact forms. A symbol that doesn't fit is left out, with a note. + +| Ecosystem | Notation | Example | +|---|---|---| +| PyPI | `module.qualname` | `yaml.load` | +| npm | `package#export.path` | `lodash#template` | +| Maven | `fully.qualified.Class#method` | `org.apache.logging.log4j.core.lookup.JndiLookup#lookup` | +| NuGet | `Namespace.Type#Method` | `System.Text.Json.JsonSerializer#Deserialize` | +| Go | `importpath.Func` / `importpath.Type.Method` | `golang.org/x/net/html.Parse` | +| crates.io | `crate::path::fn` | `hyper::proto::h1::role::parse_headers` | +| RubyGems | `Class#instance_method` / `Class.class_method` | `Nokogiri::XML::Document.parse` | +| Packagist | `Namespace\Class::method` or `function` | `Symfony\Component\Yaml\Yaml::parse` | + **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 diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index 2059ded..407ed46 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -1568,6 +1568,49 @@ def runtime_ingest( raise typer.Exit(code=1) +@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.")], + output: Annotated[ + Path | None, typer.Option("--output", "-o", help="Write the watchlist here instead of stdout.") + ] = None, + ghsa: Annotated[bool, typer.Option( + "--ghsa", + help="Also ask the GitHub advisory API for vulnerable functions when OSV has none " + "(unauthenticated, rate-limited; stops at the first limit).", + )] = False, +) -> None: + """Build the vulnerable-function watchlist runtime probes use, from the project's open SCA findings. + + Symbols come from the OSV records already cached by SCA runs (no network + by default). Findings with no known symbols are kept at package granularity. + """ + from secfoo.runtime.watchlist import build_watchlist, fetch_github_advisory + + 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) + document, notes = build_watchlist(repo, project_id, github_fetch=fetch_github_advisory if ghsa else None) + text = json.dumps(document, indent=2) + "\n" + for note in notes: + err_console.print(f"[yellow]Note:[/] {note}") + entries = document["entries"] + symbol_entries = sum(1 for e in entries if e["granularity"] == "symbol") + if output is None: + sys.stdout.write(text) + else: + output.write_text(text, encoding="utf-8") + err_console.print( + f"{len(entries)} watchlist entr{'y' if len(entries) == 1 else 'ies'}: {symbol_entries} with symbols, " + f"{len(entries) - symbol_entries} at package granularity" + (f". Wrote {output}." if output else ".") + ) + + MERMAID_CDN_URL = "https://cdn.jsdelivr.net/npm/mermaid@11/dist/mermaid.min.js" diff --git a/src/secfoo/runtime/watchlist.py b/src/secfoo/runtime/watchlist.py new file mode 100644 index 0000000..343b606 --- /dev/null +++ b/src/secfoo/runtime/watchlist.py @@ -0,0 +1,204 @@ +"""Vulnerable-function watchlist for runtime probes (#52). + +A probe can only tell whether vulnerable code runs if it knows which +function to watch. `build_watchlist()` turns a project's open SCA findings +into that list, from data secfoo already has: + +1. OSV records cached in `osv_lookups` (no network): + - `affected[].ecosystem_specific.imports[].symbols` -- Go advisories + fill this in well (`{"path": "golang.org/x/net/html", "symbols": ["Parse"]}`); + - `affected[].ecosystem_specific.affects.functions` -- RustSec writes + full `crate::path::fn` names here. +2. Optionally, the GitHub advisory API's `vulnerabilities[].vulnerable_functions` + (often empty; unauthenticated, rate-limited, so off unless asked for). +3. Otherwise the package itself, at `"granularity": "package"`, so the + probe still records whether it was loaded. + +Every symbol is normalised to the per-ecosystem notation the probes match +on (see `NOTATION`). A symbol that doesn't fit the notation is dropped, and +if none fit, the entry falls back to package granularity. An entry is never +dropped. The JSON Schema for the file is watchlist.schema.json. +""" + +from __future__ import annotations + +import json +import re +import urllib.error +import urllib.request +from collections.abc import Callable +from datetime import datetime, timezone + +from secfoo.osv import normalize_ecosystem + +WATCHLIST_VERSION = 1 +GHSA_API_URL = "https://api.github.com/advisories/{ghsa_id}" +GHSA_TIMEOUT_SECONDS = 5 + +_IDENT = r"[A-Za-z_$][\w$]*" +# Per-ecosystem symbol notation (#52). Probes in every runtime match these +# exact forms, so anything that doesn't fit is not emitted as a symbol. +NOTATION: dict[str, re.Pattern[str]] = { + "PyPI": re.compile(rf"^{_IDENT}(\.{_IDENT})+$"), # yaml.load + "npm": re.compile(rf"^(@[\w.-]+/)?[\w.-]+#{_IDENT}(\.{_IDENT})*$"), # lodash#template + "Maven": re.compile(rf"^{_IDENT}(\.{_IDENT})+#{_IDENT}$"), # a.b.Class#method + "NuGet": re.compile(rf"^{_IDENT}(\.{_IDENT})+#{_IDENT}$"), # Ns.Type#Method + "Go": re.compile(r"^[\w.~-]+(/[\w.~-]+)*\.[A-Za-z_]\w*(\.[A-Za-z_]\w*)?$"), # import/path.Func[.Method] + "crates.io": re.compile(rf"^{_IDENT}(::{_IDENT})+$"), # crate::path::fn + "RubyGems": re.compile(rf"^{_IDENT}(::{_IDENT})*[#.]{_IDENT}[?!=]?$"), # Nokogiri::XML::Document.parse + "Packagist": re.compile(rf"^({_IDENT}(\\{_IDENT})*::{_IDENT}|{_IDENT})$"), # Ns\Class::method or function +} + +FetchJson = Callable[[str], dict] + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def normalize_symbol(ecosystem: str, symbol: str) -> str | None: + """`symbol` if it already fits the ecosystem's notation, else None.""" + pattern = NOTATION.get(ecosystem) + symbol = symbol.strip() + return symbol if pattern and pattern.match(symbol) else None + + +def _osv_symbols(vuln: dict, ecosystem: str, package: str) -> list[str]: + """Raw symbols an OSV record lists for this package, before normalising.""" + symbols: list[str] = [] + for affected in vuln.get("affected") or []: + pkg = affected.get("package") or {} + if pkg.get("name") != package or (pkg.get("ecosystem") or "").split(":")[0] != ecosystem: + continue + specific = affected.get("ecosystem_specific") or {} + for imported in specific.get("imports") or []: # Go + path = imported.get("path") or package + symbols.extend(f"{path}.{name}" for name in imported.get("symbols") or []) + symbols.extend((specific.get("affects") or {}).get("functions") or []) # RustSec + return symbols + + +def _advisory_ids(vuln: dict) -> set[str]: + return {vuln.get("id", ""), *(vuln.get("aliases") or [])} - {""} + + +def _ghsa_id(vuln: dict) -> str | None: + return next((i for i in sorted(_advisory_ids(vuln)) if i.startswith("GHSA-")), None) + + +def _matching_vulns(vulns: list[dict], cve: str | None) -> list[dict]: + """The advisories a finding refers to. A finding that names an advisory + (by id or alias) gets only that one; one that doesn't ("unverified") + gets every advisory OSV has for its package and version.""" + wanted = (cve or "").strip() + if wanted and wanted.lower() != "unverified": + named = [v for v in vulns if wanted in _advisory_ids(v)] + if named: + return named + return list(vulns) + + +def fetch_github_advisory(url: str) -> dict: + """GET one advisory from the GitHub REST API (no token: public data).""" + request = urllib.request.Request( # nosec B310 -- fixed https URL from GHSA_API_URL + url, headers={"Accept": "application/vnd.github+json", "User-Agent": "secfoo-watchlist"} + ) + with urllib.request.urlopen(request, timeout=GHSA_TIMEOUT_SECONDS) as response: # nosec B310 + return json.loads(response.read().decode("utf-8")) + + +class _GithubSource: + """Optional GHSA lookups. Stops for the rest of the build at the first + rate-limit (403/429) or network error, instead of retrying.""" + + def __init__(self, fetch: FetchJson | None) -> None: + self.fetch = fetch + self.stopped_reason: str | None = None + self._cache: dict[str, dict | None] = {} + + def symbols(self, ghsa_id: str, ecosystem: str, package: str) -> list[str]: + if self.fetch is None or self.stopped_reason: + return [] + if ghsa_id not in self._cache: + try: + self._cache[ghsa_id] = self.fetch(GHSA_API_URL.format(ghsa_id=ghsa_id)) + except urllib.error.HTTPError as exc: + self._cache[ghsa_id] = None + if exc.code in (403, 429): + self.stopped_reason = f"GitHub advisory API rate limit (HTTP {exc.code}); skipped the rest" + except (urllib.error.URLError, OSError, ValueError) as exc: + self._cache[ghsa_id] = None + self.stopped_reason = f"GitHub advisory API unavailable ({exc.__class__.__name__}); skipped the rest" + advisory = self._cache[ghsa_id] or {} + found: list[str] = [] + for vulnerability in advisory.get("vulnerabilities") or []: + pkg = vulnerability.get("package") or {} + if (pkg.get("name") or "").lower() == package.lower(): + found.extend(vulnerability.get("vulnerable_functions") or []) + return found + + +def _entry(*, fingerprint: str, advisory: str | None, ecosystem: str, package: str, version: str | None, + symbols: list[str], source: str) -> dict: + unique = sorted(set(symbols)) + return { + "finding_fingerprint": fingerprint, + "advisory": advisory, + "ecosystem": ecosystem, + "package": package, + "version": version, + "granularity": "symbol" if unique else "package", + "symbols": unique, + "source": source if unique else "package", + } + + +def build_watchlist(repo, project_id: int, *, github_fetch: FetchJson | None = None) -> tuple[dict, list[str]]: + """(watchlist document, notes) for the project's open SCA findings. + + `github_fetch` enables the optional GHSA source; tests pass a fake. + Notes explain anything a user should know (unrecognised ecosystems, + symbols that didn't fit the notation, a GHSA rate limit). + """ + lookups = repo.osv_lookup_map() + github = _GithubSource(github_fetch) + entries: list[dict] = [] + notes: list[str] = [] + unnormalised = 0 + + def keep(ecosystem: str, raw: list[str]) -> list[str]: + nonlocal unnormalised + kept = [s for s in (normalize_symbol(ecosystem, r) for r in raw) if s] + unnormalised += len(raw) - len(kept) + return kept + + for finding in repo.list_sca_findings(project_id=project_id, status="open"): + ecosystem = normalize_ecosystem(finding.ecosystem) + if ecosystem is None: + notes.append(f"{finding.package}: ecosystem {finding.ecosystem!r} isn't recognised; watching the package") + entries.append(_entry(fingerprint=finding.fingerprint, advisory=None, ecosystem=finding.ecosystem or "unknown", + package=finding.package, version=finding.version, symbols=[], source="package")) + continue + record = lookups.get((ecosystem, finding.package, finding.version or "")) + vulns = _matching_vulns(record.vulns, finding.cve) if record else [] + if not vulns: + named = (finding.cve or "").strip() + advisory = named if named and named.lower() != "unverified" else None + entries.append(_entry(fingerprint=finding.fingerprint, advisory=advisory, ecosystem=ecosystem, + package=finding.package, version=finding.version, symbols=[], source="package")) + continue + for vuln in vulns: + symbols, source = keep(ecosystem, _osv_symbols(vuln, ecosystem, finding.package)), "osv" + ghsa_id = _ghsa_id(vuln) + if not symbols and ghsa_id: + symbols, source = keep(ecosystem, github.symbols(ghsa_id, ecosystem, finding.package)), "ghsa" + entries.append(_entry(fingerprint=finding.fingerprint, advisory=ghsa_id or vuln.get("id"), + ecosystem=ecosystem, package=finding.package, version=finding.version, + symbols=symbols, source=source)) + + if unnormalised: + notes.append(f"{unnormalised} advisory symbol(s) didn't fit their ecosystem's notation and were left out") + if github.stopped_reason: + notes.append(github.stopped_reason) + entries.sort(key=lambda e: (e["finding_fingerprint"], e["advisory"] or "")) + return {"version": WATCHLIST_VERSION, "generated_at": _now(), "entries": entries}, notes diff --git a/src/secfoo/runtime/watchlist.schema.json b/src/secfoo/runtime/watchlist.schema.json new file mode 100644 index 0000000..2ddaf5d --- /dev/null +++ b/src/secfoo/runtime/watchlist.schema.json @@ -0,0 +1,39 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/secfoo-com/secfoo/blob/main/src/secfoo/runtime/watchlist.schema.json", + "title": "Secfoo vulnerable-function watchlist", + "description": "Written by `secfoo runtime watchlist` (#52) and read by runtime probes and the sensor. One entry per open SCA finding and advisory.", + "type": "object", + "required": ["version", "generated_at", "entries"], + "additionalProperties": false, + "properties": { + "version": {"const": 1}, + "generated_at": {"type": "string", "format": "date-time"}, + "entries": { + "type": "array", + "items": { + "type": "object", + "required": ["finding_fingerprint", "advisory", "ecosystem", "package", "version", "granularity", "symbols", "source"], + "additionalProperties": false, + "properties": { + "finding_fingerprint": {"type": "string", "minLength": 1, "description": "sca_findings.fingerprint; runtime evidence hits (#53) carry it back."}, + "advisory": {"type": ["string", "null"], "description": "GHSA id when known, else the OSV id; null when the finding names none."}, + "ecosystem": {"type": "string", "minLength": 1, "description": "OSV ecosystem: PyPI, npm, Maven, NuGet, Go, crates.io, RubyGems, Packagist."}, + "package": {"type": "string", "minLength": 1}, + "version": {"type": ["string", "null"]}, + "granularity": {"enum": ["symbol", "package"]}, + "symbols": { + "type": "array", + "uniqueItems": true, + "items": {"type": "string", "minLength": 1}, + "description": "In the ecosystem's notation (see secfoo/runtime/watchlist.py NOTATION). Empty for package granularity." + }, + "source": {"enum": ["osv", "ghsa", "package"]} + }, + "if": {"properties": {"granularity": {"const": "package"}}}, + "then": {"properties": {"symbols": {"maxItems": 0}, "source": {"const": "package"}}}, + "else": {"properties": {"symbols": {"minItems": 1}, "source": {"enum": ["osv", "ghsa"]}}} + } + } + } +} diff --git a/tests/test_runtime/test_watchlist.py b/tests/test_runtime/test_watchlist.py new file mode 100644 index 0000000..5ce1475 --- /dev/null +++ b/tests/test_runtime/test_watchlist.py @@ -0,0 +1,229 @@ +"""#52: vulnerable-function watchlist built from cached OSV records (no network).""" +from __future__ import annotations + +import io +import json +import urllib.error +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from secfoo.cli import app +from secfoo.runtime import watchlist +from secfoo.runtime.watchlist import build_watchlist, normalize_symbol +from secfoo.storage.repository import RunRepository + +# Fixture OSV records, trimmed to the fields the builder reads. +GO_VULN = { + "id": "GO-2023-1571", "aliases": ["CVE-2022-41723", "GHSA-vvpx-j8f3-3w6h"], + "affected": [{ + "package": {"ecosystem": "Go", "name": "golang.org/x/net"}, + "ecosystem_specific": {"imports": [ + {"path": "golang.org/x/net/http2/hpack", "symbols": ["Decoder.DecodeFull", "Decoder.Write"]}, + {"path": "golang.org/x/net/http2", "symbols": ["Server.ServeConn"]}, + ]}, + }], +} +RUST_VULN = { + "id": "RUSTSEC-2021-0078", "aliases": ["CVE-2021-32715", "GHSA-f3pg-qwvg-p99c"], + "affected": [{ + "package": {"ecosystem": "crates.io", "name": "hyper"}, + "ecosystem_specific": {"affects": {"functions": ["hyper::proto::h1::role::parse_headers"]}}, + }], +} +PYYAML_VULN = { # PyPI advisories rarely carry symbols in OSV + "id": "GHSA-8q59-q68h-6hv4", "aliases": ["CVE-2020-14343"], + "affected": [{"package": {"ecosystem": "PyPI", "name": "pyyaml"}}], +} +PYYAML_OTHER = {"id": "PYSEC-2021-142", "aliases": ["CVE-2020-1747"], "affected": [{"package": {"ecosystem": "PyPI", "name": "pyyaml"}}]} +BAD_SYMBOL_VULN = { + "id": "GO-2099-0001", + "affected": [{"package": {"ecosystem": "Go", "name": "example.com/lib"}, + "ecosystem_specific": {"imports": [{"path": "example.com/lib", "symbols": ["not a symbol!"]}]}}], +} + + +@pytest.fixture +def repo(tmp_path): + with RunRepository(db_path=tmp_path / "db.sqlite") as repo: + yield repo + + +def _finding(repo, pid, run_id, fp, package, ecosystem, version, cve="unverified", status="open"): + repo.upsert_sca_finding(project_id=pid, fingerprint=fp, run_id=run_id, current_ref="D1", package=package, + severity="high", ecosystem=ecosystem, version=version, cve=cve) + if status == "closed": + repo.connection.execute("UPDATE sca_findings SET status='closed' WHERE fingerprint=?", (fp,)) + repo.connection.commit() + + +@pytest.fixture +def project(repo): + pid = repo.upsert_project("https://github.com/acme/app", "app", "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_osv_lookup(ecosystem="Go", package="golang.org/x/net", version="0.6.0", vulns=[GO_VULN]) + repo.upsert_osv_lookup(ecosystem="crates.io", package="hyper", version="0.14.9", vulns=[RUST_VULN]) + repo.upsert_osv_lookup(ecosystem="PyPI", package="pyyaml", version="5.3", vulns=[PYYAML_VULN, PYYAML_OTHER]) + _finding(repo, pid, run_id, "fp-go", "golang.org/x/net", "Go", "0.6.0") + _finding(repo, pid, run_id, "fp-rust", "hyper", "Rust", "0.14.9", cve="CVE-2021-32715") + _finding(repo, pid, run_id, "fp-yaml", "pyyaml", "Python", "5.3", cve="CVE-2020-14343") + _finding(repo, pid, run_id, "fp-nocache", "left-pad", "npm", "1.0.0") + _finding(repo, pid, run_id, "fp-closed", "requests", "Python", "2.0.0", status="closed") + return pid + + +def _by_fp(document): + out = {} + for entry in document["entries"]: + out.setdefault(entry["finding_fingerprint"], []).append(entry) + return out + + +def test_osv_go_import_symbols_become_importpath_qualified_symbols(repo, project): + [entry] = _by_fp(build_watchlist(repo, project)[0])["fp-go"] + assert entry["granularity"] == "symbol" and entry["source"] == "osv" + assert entry["symbols"] == [ + "golang.org/x/net/http2.Server.ServeConn", + "golang.org/x/net/http2/hpack.Decoder.DecodeFull", + "golang.org/x/net/http2/hpack.Decoder.Write", + ] + assert entry["advisory"] == "GHSA-vvpx-j8f3-3w6h" # GHSA id preferred over the OSV id + assert (entry["ecosystem"], entry["package"], entry["version"]) == ("Go", "golang.org/x/net", "0.6.0") + + +def test_rustsec_affected_functions_are_used(repo, project): + [entry] = _by_fp(build_watchlist(repo, project)[0])["fp-rust"] + assert entry["symbols"] == ["hyper::proto::h1::role::parse_headers"] + assert entry["ecosystem"] == "crates.io" # report's "Rust" normalised to OSV's vocabulary + + +def test_no_symbols_falls_back_to_package_granularity_for_the_named_advisory(repo, project): + [entry] = _by_fp(build_watchlist(repo, project)[0])["fp-yaml"] + assert entry["granularity"] == "package" and entry["symbols"] == [] and entry["source"] == "package" + assert entry["advisory"] == "GHSA-8q59-q68h-6hv4" # only the CVE the finding named, not PYSEC-2021-142 + + +def test_finding_without_cached_osv_data_is_kept_at_package_granularity(repo, project): + [entry] = _by_fp(build_watchlist(repo, project)[0])["fp-nocache"] + assert entry["granularity"] == "package" and entry["advisory"] is None + + +def test_only_open_findings_are_listed(repo, project): + assert "fp-closed" not in _by_fp(build_watchlist(repo, project)[0]) + + +def test_unverified_finding_gets_every_advisory_for_its_version(repo, project): + repo.connection.execute("UPDATE sca_findings SET cve='unverified' WHERE fingerprint='fp-yaml'") + repo.connection.commit() + advisories = {e["advisory"] for e in _by_fp(build_watchlist(repo, project)[0])["fp-yaml"]} + assert advisories == {"GHSA-8q59-q68h-6hv4", "PYSEC-2021-142"} + + +def test_symbols_that_dont_fit_the_notation_are_dropped_but_the_entry_is_kept(repo, project): + pid = project + run_id = repo.list_sca_findings(project_id=pid)[0].first_seen_run_id + repo.upsert_osv_lookup(ecosystem="Go", package="example.com/lib", version="1.0.0", vulns=[BAD_SYMBOL_VULN]) + _finding(repo, pid, run_id, "fp-bad", "example.com/lib", "Go", "1.0.0") + document, notes = build_watchlist(repo, pid) + [entry] = _by_fp(document)["fp-bad"] + assert entry["granularity"] == "package" and entry["symbols"] == [] + assert any("didn't fit" in n for n in notes) + + +def test_unrecognised_ecosystem_is_kept_and_noted(repo, project): + run_id = repo.list_sca_findings(project_id=project)[0].first_seen_run_id + _finding(repo, project, run_id, "fp-alpine", "musl", "Alpine", "1.2.3") + document, notes = build_watchlist(repo, project) + [entry] = _by_fp(document)["fp-alpine"] + assert entry["granularity"] == "package" and entry["ecosystem"] == "Alpine" + assert any("isn't recognised" in n for n in notes) + + +# ---------- optional GitHub advisory source ---------- + + +def test_ghsa_source_fills_symbols_when_osv_has_none(repo, project): + calls = [] + + def fake_fetch(url): + calls.append(url) + return {"vulnerabilities": [{"package": {"ecosystem": "pip", "name": "PyYAML"}, + "vulnerable_functions": ["yaml.load", "yaml.full_load"]}]} + + [entry] = _by_fp(build_watchlist(repo, project, github_fetch=fake_fetch)[0])["fp-yaml"] + assert entry["source"] == "ghsa" and entry["symbols"] == ["yaml.full_load", "yaml.load"] + assert calls == ["https://api.github.com/advisories/GHSA-8q59-q68h-6hv4"] # OSV symbols already covered Go/Rust + + +def test_ghsa_rate_limit_stops_further_calls_and_is_noted(repo, project): + repo.connection.execute("UPDATE sca_findings SET cve='unverified' WHERE fingerprint='fp-yaml'") + repo.connection.commit() + calls = [] + + def limited(url): + calls.append(url) + raise urllib.error.HTTPError(url, 403, "rate limited", {}, io.BytesIO()) + + document, notes = build_watchlist(repo, project, github_fetch=limited) + assert len(calls) == 1 # stopped after the first limit + assert all(e["granularity"] == "package" for e in _by_fp(document)["fp-yaml"]) + assert any("rate limit" in n for n in notes) + + +def test_no_network_by_default(repo, project, monkeypatch): + monkeypatch.setattr(watchlist.urllib.request, "urlopen", lambda *a, **k: pytest.fail("no network by default")) + build_watchlist(repo, project) + + +# ---------- notation ---------- + + +@pytest.mark.parametrize("ecosystem, symbol", [ + ("PyPI", "yaml.load"), ("npm", "lodash#template"), ("npm", "@babel/core#transform"), + ("Maven", "org.apache.logging.log4j.core.lookup.JndiLookup#lookup"), + ("NuGet", "System.Text.Json.JsonSerializer#Deserialize"), ("Go", "golang.org/x/net/html.Parse"), + ("Go", "golang.org/x/net/http2.Server.ServeConn"), ("crates.io", "hyper::proto::h1::role::parse_headers"), + ("RubyGems", "Nokogiri::XML::Document.parse"), ("RubyGems", "Rack::Request#params"), + ("Packagist", "Symfony\\Component\\Yaml\\Yaml::parse"), ("Packagist", "unserialize"), +]) +def test_issue_examples_fit_their_notation(ecosystem, symbol): + assert normalize_symbol(ecosystem, symbol) == symbol + + +@pytest.mark.parametrize("ecosystem, symbol", [ + ("PyPI", "load"), ("PyPI", "yaml load"), ("npm", "lodash.template"), ("Maven", "JndiLookup.lookup"), + ("crates.io", "parse_headers"), ("Go", "Parse"), ("Unknown", "a.b"), +]) +def test_symbols_outside_the_notation_are_rejected(ecosystem, symbol): + assert normalize_symbol(ecosystem, symbol) is None + + +# ---------- schema + CLI ---------- + + +def test_output_matches_the_json_schema(repo, project): + jsonschema = pytest.importorskip("jsonschema") + schema = json.loads((Path(watchlist.__file__).parent / "watchlist.schema.json").read_text(encoding="utf-8")) + jsonschema.Draft202012Validator.check_schema(schema) + document = build_watchlist(repo, project)[0] + jsonschema.Draft202012Validator(schema).validate(document) + bad = {**document, "entries": [{**document["entries"][0], "granularity": "package", "symbols": ["x.y"]}]} + assert not jsonschema.Draft202012Validator(schema).is_valid(bad) + + +def test_cli_writes_the_watchlist(tmp_path, monkeypatch, project, repo): + monkeypatch.setattr("secfoo.cli.RunRepository", lambda: RunRepository(db_path=repo.db_path)) + out = tmp_path / "watchlist.json" + result = CliRunner().invoke(app, ["runtime", "watchlist", "--project", "app", "-o", str(out)]) + assert result.exit_code == 0, result.output + document = json.loads(out.read_text(encoding="utf-8")) + assert document["version"] == 1 and len(document["entries"]) == 4 + assert "4 watchlist entries: 2 with symbols, 2 at package granularity" in result.output + + +def test_cli_unknown_project(tmp_path, monkeypatch, repo): + monkeypatch.setattr("secfoo.cli.RunRepository", lambda: RunRepository(db_path=repo.db_path)) + result = CliRunner().invoke(app, ["runtime", "watchlist", "--project", "nope"]) + assert result.exit_code == 1 and "no single project matches" in result.output