diff --git a/README.md b/README.md index 54ef984..a9f25bb 100644 --- a/README.md +++ b/README.md @@ -293,9 +293,33 @@ importing `os.system`, a Keras Lambda layer, a Jinja chat template with a sandbox escape. None of those have a CVE, because the file itself is the payload rather than a published component with a patchable defect. The full registry is at [`docs/vex-finding-classes.md`](docs/vex-finding-classes.md) -and each identifier resolves under `https://aisbom.io/vex/`. CVE-keyed -statements about your `requirements.txt` pins are a separate, additive -concern and arrive with OSV mapping. +and each identifier resolves under `https://aisbom.io/vex/`. + +#### CVEs in `requirements.txt` pins + +The same documents also carry CVE-keyed statements for your Python +dependencies. With `--vex`, each exact pin (`requests==2.19.0`) is looked up in +[OSV](https://osv.dev), and every advisory covering that version becomes a +statement keyed on its CVE — or on its GHSA/PYSEC id when it has no CVE, with +the other ids listed as `aliases`. + +- **Exact pins only.** `torch>=2.0` doesn't say which version is installed, so + ranges are skipped and counted in the scan summary rather than guessed at. +- **`affected` means both OSV and AIsbom agree.** OSV names the candidate + advisories; AIsbom re-checks the pinned version against each advisory's + published ranges. When the two disagree, or the ranges can't be evaluated, + the statement is `under_investigation`. +- **No `not_affected` for dependencies.** AIsbom can't see whether your code + reaches the vulnerable function, and "no advisory found" isn't evidence the + vulnerable code is absent, so dependencies never get a negative statement. +- **Cached for 24 hours** in `~/.aisbom/osv_cache.json`, so repeat CI scans of + the same pins make no requests. +- **Never breaks a scan.** If OSV is unreachable, slow or returns something + unexpected, the run prints a warning and the VEX documents simply carry no + CVE statements. Exit codes and model findings are unchanged. + +A plain `aisbom scan` without `--vex` never contacts OSV. To keep `--vex` fully +offline, pass `--no-osv` or set `AISBOM_NO_OSV=1`. #### Remediation evidence (`fixed`) @@ -616,6 +640,10 @@ To detect failure loops (see [Authentication](#authentication-private--gated-hug Each event carries an anonymous `user_id` — a SHA-256 of your machine's MAC address plus an app salt, truncated to 16 hex chars. Stored in `~/.aisbom/config.json`. Lets us see returning users without identifying anyone. +### OSV lookups (`--vex` only) + +When you pass `--vex` and the scan finds exact `requirements.txt` pins, AIsbom sends each pinned **package name and version** to the public OSV API at `https://api.osv.dev`, and fetches the advisories it names. Nothing else is sent: no file paths, model names, hashes, findings, or identifiers. This is a request to a third-party service, not telemetry, so `AISBOM_NO_TELEMETRY` does not affect it; `--no-osv` or `AISBOM_NO_OSV=1` does. Responses are cached locally in `~/.aisbom/osv_cache.json` for 24 hours, and deleting that file is always safe. + ### What's never collected File paths, directory contents, model names, target URLs, file hashes from your SBOMs, exception messages, tracebacks, or anything that could identify you, your project, or your organization. diff --git a/action.yml b/action.yml index afb0318..21e97b1 100644 --- a/action.yml +++ b/action.yml @@ -33,7 +33,7 @@ inputs: required: false default: 'true' token: - description: 'Optional. Per-repo API token for posting the generated SBOM to your hosted inventory dashboard at app.aisbom.io. Setting it also runs the scan with --vex and uploads the resulting VEX documents alongside the SBOM, so the dashboard can show whether a finding is exploitable. Leave unset for purely local PR-comment behavior. Get a token at https://app.aisbom.io/connect.' + description: 'Optional. Per-repo API token for posting the generated SBOM to your hosted inventory dashboard at app.aisbom.io. Setting it also runs the scan with --vex and uploads the resulting VEX documents alongside the SBOM, so the dashboard can show whether a finding is exploitable. With --vex, exact requirements.txt pins (package name and version only) are also looked up in the public OSV database (api.osv.dev) from the runner, adding CVE statements to those documents. Leave unset for purely local PR-comment behavior. Get a token at https://app.aisbom.io/connect.' required: false default: '' platform-url: diff --git a/action/platform_upload.py b/action/platform_upload.py index 538bdcd..669c204 100644 --- a/action/platform_upload.py +++ b/action/platform_upload.py @@ -22,6 +22,11 @@ WEBHOOK_PATH = "/v1/scan-result" REQUEST_TIMEOUT_SEC = 15.0 +# The receiver rejects bodies over 1 MiB with a 413. VEX documents grow with +# dependency CVE statements (one per advisory per pinned package), so an +# envelope can cross that line where the SBOM alone never would. +MAX_BODY_BYTES = 1024 * 1024 + EXIT_OK = 0 EXIT_UPLOAD_FAILED = 3 @@ -117,7 +122,18 @@ def build_request_body(sbom_path: str, vex_documents: List[Dict[str, Any]] | Non if not isinstance(sbom, dict): return raw - return json.dumps({"sbom": sbom, "vex": vex_documents}).encode("utf-8") + envelope = json.dumps({"sbom": sbom, "vex": vex_documents}).encode("utf-8") + # Over the receiver's cap the whole request would be rejected, costing the + # inventory entry to carry supplementary documents — the same trade the + # unreadable-VEX branch above already refuses. Send the SBOM on its own. + if len(envelope) > MAX_BODY_BYTES: + print( + f"[aisbom-action] VEX documents omitted: the upload would be " + f"{len(envelope)} bytes, over the {MAX_BODY_BYTES}-byte limit. " + "The SBOM is uploaded alone; the VEX files remain on the runner." + ) + return raw + return envelope def summarize_response(status: int, body: str) -> str: @@ -167,8 +183,10 @@ def upload( payload = build_request_body(sbom_path, vex_documents) # Part of the same disclosure as the lines above: an opted-in user can # see from the log exactly how many documents left their runner, not - # just that "an upload happened". - print(f"[aisbom-action] vex-documents={len(vex_documents)}") + # just that "an upload happened". Counted from what is actually sent, + # since an oversized envelope falls back to the SBOM alone. + sent = len(vex_documents) if vex_documents and payload.startswith(b'{"sbom"') else 0 + print(f"[aisbom-action] vex-documents={sent}") resp = requests.post( url, data=payload, diff --git a/aisbom/cli.py b/aisbom/cli.py index 8351659..b46305a 100644 --- a/aisbom/cli.py +++ b/aisbom/cli.py @@ -28,6 +28,7 @@ import uuid from .version_check import check_latest_version from . import loop_state +from . import osv from . import telemetry import requests @@ -63,6 +64,9 @@ def main( AISBOM_NO_TELEMETRY=1 Disable all anonymous usage telemetry. Honored on every code path; never overridden. + + AISBOM_NO_OSV=1 Same as `scan --no-osv`: never query OSV for + requirements.txt CVEs when emitting VEX. """ # Order matters: --version wins over the no-args panel so that # `aisbom --version` is short and scriptable. @@ -434,6 +438,8 @@ def _emit_vex( vex_format: "VexFormat", baseline_path: str | None, schema_version: str, + dependencies: list[dict] | None = None, + osv_enabled: bool = False, ) -> None: """Write the VEX document(s) that accompany a just-written SBOM. @@ -458,6 +464,20 @@ def _emit_vex( raise typer.Exit(code=1) statements = derive_statements(artifacts, baseline) + + # CVE-keyed statements for exact requirements.txt pins join the same + # document (#128). Best-effort by contract: a failed lookup costs these + # statements and prints why, and nothing else about the run changes. + osv_result = None + if osv_enabled and dependencies: + osv_result = osv.lookup_dependency_statements(dependencies) + statements = statements + osv_result.statements + if osv_result.error: + err_console.print( + f"[yellow]⚠ {osv_result.error}; VEX carries no CVE statements " + "for requirements.txt pins this run.[/yellow]" + ) + openvex_path, cyclonedx_path = _vex_paths(output) if vex_format in (VexFormat.OPENVEX, VexFormat.BOTH): @@ -484,6 +504,20 @@ def _emit_vex( summary += f", {fixed} fixed since baseline" console.print(summary + ".[/dim]") + if osv_result is not None and osv_result.error is None: + line = ( + f"[dim]OSV: {len(osv_result.statements)} CVE statement(s) across " + f"{osv_result.queried} pinned dependenc" + f"{'y' if osv_result.queried == 1 else 'ies'}" + ) + if osv_result.skipped_unpinned: + line += ( + f"; {osv_result.skipped_unpinned} unpinned dependenc" + f"{'y' if osv_result.skipped_unpinned == 1 else 'ies'} skipped " + "(only exact == pins are looked up)" + ) + console.print(line + ".[/dim]") + def _generate_markdown(results: dict) -> str: """Render a GitHub-flavored Markdown report for CI artifacts.""" @@ -573,6 +607,16 @@ def scan( ), rich_help_panel="Advanced Options", ), + no_osv: bool = typer.Option( + False, + "--no-osv", + help=( + "With --vex, do not query OSV (api.osv.dev) for CVEs affecting " + "exact requirements.txt pins. Use in air-gapped environments; " + "AISBOM_NO_OSV=1 does the same." + ), + rich_help_panel="Advanced Options", + ), ): """ Deep Introspection Scan: Analyzes binary headers and dependency manifests. @@ -859,10 +903,12 @@ def _risk_score(label: str) -> int: _emit_vex( sbom_json=sbom_json, artifacts=results['artifacts'], + dependencies=results.get('dependencies', []), output=output, vex_format=vex_format, baseline_path=vex_baseline, schema_version=schema_version, + osv_enabled=not (no_osv or osv.disabled_by_env()), ) has_content = bool(results.get('artifacts') or results.get('dependencies')) diff --git a/aisbom/cyclonedx_gen.py b/aisbom/cyclonedx_gen.py index f4a4e59..043dd34 100644 --- a/aisbom/cyclonedx_gen.py +++ b/aisbom/cyclonedx_gen.py @@ -22,7 +22,7 @@ from cyclonedx.model.component import Component, ComponentType from cyclonedx.output.json import JsonV1Dot5, JsonV1Dot6, JsonV1Dot7 -from .modelcard import bom_ref_for, inject_model_cards +from .modelcard import bom_ref_for, dependency_bom_ref, inject_model_cards from .properties import build_component_properties from .spdx_gen import _sha256_or_none @@ -101,7 +101,7 @@ def build_bom(results: Dict[str, Any]) -> Bom: bom.components.add(c) - for dep in results.get("dependencies", []): + for dep_index, dep in enumerate(results.get("dependencies", [])): version = dep.get("version") # `version: "unknown"` was a placeholder carrying no more information # than an absent field, and it cost the completeness grade real points @@ -112,6 +112,9 @@ def build_bom(results: Dict[str, Any]) -> Bom: name=dep["name"], version=None if version == "unknown" else version, type=ComponentType.LIBRARY, + # Stable for the same reason as the model components: a CVE-keyed + # VEX statement (#128) addresses this component by bom-ref. + bom_ref=dependency_bom_ref(dep_index, dep), )) return bom diff --git a/aisbom/modelcard.py b/aisbom/modelcard.py index d4bc0f5..363d3ec 100644 --- a/aisbom/modelcard.py +++ b/aisbom/modelcard.py @@ -212,6 +212,18 @@ def bom_ref_for(index: int, art: Dict[str, Any]) -> str: return f"artifact-{index}-{art.get('name', 'unknown')}" +def dependency_bom_ref(index: int, dep: Dict[str, Any]) -> str: + """Stable `bom-ref` for a requirements.txt dependency component. + + Left to the library this is a fresh random string on every run, which is + what `bom_ref_for` exists to avoid for models. Dependencies need the same + once a VEX statement addresses one (#128): the CVE statement and the SBOM + component are written separately and must join on this value. The index + keeps the same package listed in two requirements files distinct. + """ + return f"dependency-{index}-{dep.get('name', 'unknown')}" + + def inject_model_cards( bom_json: str, artifacts: List[Dict[str, Any]], diff --git a/aisbom/osv.py b/aisbom/osv.py new file mode 100644 index 0000000..1c78a88 --- /dev/null +++ b/aisbom/osv.py @@ -0,0 +1,664 @@ +"""OSV lookup for pinned requirements.txt dependencies — CVE-keyed VEX (#128). + +The finding-class statements in :mod:`aisbom.vex` describe content inside model +files, which has no CVE. The ``requirements.txt`` pins in the same scan are the +part where real CVEs do exist, and this module supplies statements for them: +it asks OSV which advisories cover each exact pin and turns the confirmed ones +into :class:`~aisbom.vex.VexStatement` objects the existing emitters serialize +unchanged. + +Contract +-------- + +**Enrichment, never the scan.** Every failure — no network, a timeout, a +malformed response, a record OSV cannot return, an exhausted time budget — +degrades to *no CVE statements* plus a reason string. Nothing here raises into +the CLI, changes an exit code, or touches the model findings. This is the same +asymmetry as the HF model-card fetch (#111), and it is what keeps the +air-gapped workflow working with no flag at all. A partial answer is treated as +no answer: a document listing some of a dependency's advisories reads as the +complete list, which is worse than an honest omission. + +**Only exact pins are looked up.** ``torch>=2.0`` does not say which version is +installed, so any statement about "2.0" might be about a version nobody runs. +Range specifiers are counted and reported, never queried. + +**OSV shortlists, a local check confirms.** ``/v1/querybatch`` names the +advisories OSV believes cover a name + version; each full record is then +re-evaluated here against its ``versions`` list and ``ECOSYSTEM`` ranges using +PEP 440 ordering. Agreement is ``affected``. Where the local check disagrees or +cannot evaluate the record (a ``GIT``-only range, an unparseable version), the +statement is ``under_investigation`` — never silently dropped, and never +asserted as ``affected`` on one opinion. + +**No negative statements.** "No known advisory" is not +``vulnerable_code_not_present``, and AIsbom never observes whether vulnerable +dependency code is reachable, so dependencies only ever receive ``affected`` +or ``under_investigation``. + +**Cached on disk.** A CI job scanning the same ``requirements.txt`` on every +push must not hit OSV every time. Answers (hits *and* misses) live in +``~/.aisbom/osv_cache.json`` for :data:`CACHE_TTL_SECONDS`. An unwritable +config directory just means no cache. + +PyInstaller constraint: ``requests`` and ``packaging`` only, both already +bundled. +""" + +from __future__ import annotations + +import json +import os +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple + +import requests +from packaging.utils import canonicalize_name +from packaging.version import InvalidVersion, Version + +from .modelcard import dependency_bom_ref +from .spdx_gen import _tool_version +from .vex import ( + STATUS_AFFECTED, + STATUS_UNDER_INVESTIGATION, + FindingClass, + VexStatement, +) + +OSV_API = "https://api.osv.dev/v1" +OSV_VULNERABILITY_URL = "https://osv.dev/vulnerability/" +OSV_SOURCE_NAME = "OSV" +ECOSYSTEM = "PyPI" + +CACHE_FILENAME = "osv_cache.json" +CACHE_TTL_SECONDS = 24 * 60 * 60 +_CACHE_SCHEMA = 1 + +# Per request, and for the whole lookup. Capped like the model-card fetch: a +# slow OSV must never be what makes a CI scan hang. +REQUEST_TIMEOUT_SECONDS = 10 +LOOKUP_BUDGET_SECONDS = 30 + +# querybatch accepts up to 1000 queries; a requirements.txt is far smaller, but +# chunking keeps a pathological file from being one rejected request. +_BATCH_SIZE = 500 +# A package with more advisories than fit one page is followed; this bounds it. +_MAX_PAGES = 10 +# Advisory records are fetched concurrently, bounded so a large pin set is +# polite to a free public API rather than a burst of hundreds of requests. +_FETCH_WORKERS = 8 +# Transient-failure retry: three attempts in total, with a short linear pause. +_ATTEMPTS = 3 +_BACKOFF_SECONDS = 0.5 +_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504}) + +DISABLE_ENV_VAR = "AISBOM_NO_OSV" + +_DEFAULT_CACHE_DIR = object() + + +class OsvUnavailable(Exception): + """Raised internally for any failure that degrades the lookup.""" + + +@dataclass +class OsvLookupResult: + statements: List[VexStatement] = field(default_factory=list) + #: Exact pins the lookup covered. + queried: int = 0 + #: Dependencies left out because they are not exact pins. + skipped_unpinned: int = 0 + #: Why the lookup degraded to no statements, or ``None`` if it did not. + error: Optional[str] = None + + +def disabled_by_env(environ: Mapping[str, str] = os.environ) -> bool: + return bool(environ.get(DISABLE_ENV_VAR)) + + +# -------------------------------------------------------------------------- +# Matching +# -------------------------------------------------------------------------- + +def _parse(version: str) -> Version: + return Version(version) + + +_EVENT_KINDS = ("introduced", "fixed", "last_affected") + + +def _ordered_events( + events: Sequence[Mapping[str, Any]], +) -> List[Tuple[str, Optional[Version]]]: + """``(kind, boundary)`` pairs in version order. + + OSV's ``introduced: "0"`` is a sentinel for "from the very first version", + represented as a ``None`` boundary that sorts first and precedes every pin. + It must not become ``Version("0")``: PEP 440 orders pre-releases such as + ``0.dev0`` and ``0rc1`` *below* release 0, so those pins would fall outside + an advisory that covers every version. Raises :class:`InvalidVersion` for + any other boundary that does not parse. + """ + ordered = [] + for event in events: + kind = next((k for k in _EVENT_KINDS if k in event), None) + if kind is None: + continue # `limit` only bounds git ranges + raw = str(event[kind]) + boundary = None if kind == "introduced" and raw == "0" else _parse(raw) + ordered.append((kind, boundary)) + ordered.sort(key=lambda kv: (kv[1] is not None, kv[1] or Version("0"))) + return ordered + + +def _in_ecosystem_range(events: Sequence[Mapping[str, Any]], version: Version) -> bool: + """Evaluate one ``ECOSYSTEM`` range per the OSV schema's algorithm. + + Events are applied in version order, not listed order. ``last_affected`` + is inclusive; ``fixed`` is exclusive. ``limit`` only bounds git ranges and + is ignored. Raises :class:`InvalidVersion` for an event that does not parse, + which the caller reports as undetermined rather than as unaffected. + """ + vulnerable = False + for kind, boundary in _ordered_events(events): + if kind == "introduced" and (boundary is None or version >= boundary): + vulnerable = True + elif kind == "fixed" and version >= boundary: + vulnerable = False + elif kind == "last_affected" and version > boundary: + vulnerable = False + return vulnerable + + +def _closing_fix(events: Sequence[Mapping[str, Any]], version: Version) -> Optional[str]: + """The ``fixed`` boundary that ends the affected interval holding ``version``. + + ``None`` when the pin is not inside this range, or its interval has no fix + (open-ended, or closed by ``last_affected``). An advisory with disjoint + intervals lists older fixes too, and those are not upgrade targets for a + pin in a later interval. + """ + try: + if not _in_ecosystem_range(events, version): + return None + ordered = _ordered_events(events) + except InvalidVersion: + return None + for kind, boundary in ordered: + if boundary is None or boundary <= version or kind == "introduced": + continue + return str(boundary) if kind == "fixed" else None + return None + + +def version_affected(record: Mapping[str, Any], name: str, version: str) -> Optional[bool]: + """Does this OSV record cover ``name == version``? + + ``True`` or ``False`` where a PyPI entry for the package could be evaluated; + ``None`` where nothing could — no entry for this package, only ``GIT`` or + ``SEMVER`` ranges, or a version that does not parse. ``None`` is not a + negative, and callers must not treat it as one. + """ + wanted = canonicalize_name(name) + try: + pin = _parse(version) + except InvalidVersion: + return None + + evaluated = False + for entry in record.get("affected") or []: + package = entry.get("package") or {} + if package.get("ecosystem") != ECOSYSTEM: + continue + if canonicalize_name(str(package.get("name", ""))) != wanted: + continue + + for listed in entry.get("versions") or []: + evaluated = True + try: + if _parse(str(listed)) == pin: + return True + except InvalidVersion: + continue + + for rng in entry.get("ranges") or []: + if rng.get("type") != "ECOSYSTEM": + continue + try: + if _in_ecosystem_range(rng.get("events") or [], pin): + return True + except InvalidVersion: + return None + evaluated = True + + return False if evaluated else None + + +def vulnerability_ids(record: Mapping[str, Any]) -> Tuple[str, Tuple[str, ...]]: + """``(primary, aliases)`` — the CVE if the advisory has one, else its OSV id.""" + osv_id = str(record["id"]) + names = [osv_id] + [str(a) for a in record.get("aliases") or []] + cves = sorted({n for n in names if n.startswith("CVE-")}) + primary = cves[0] if cves else osv_id + aliases = tuple(sorted({n for n in names if n != primary})) + return primary, aliases + + +def _fixed_versions(record: Mapping[str, Any], name: str, version: str) -> List[Version]: + """Fixes that close the pinned version's affected interval, one per range.""" + wanted = canonicalize_name(name) + try: + pin = _parse(version) + except InvalidVersion: + return [] + fixed = [] + for entry in record.get("affected") or []: + package = entry.get("package") or {} + if package.get("ecosystem") != ECOSYSTEM: + continue + if canonicalize_name(str(package.get("name", ""))) != wanted: + continue + for rng in entry.get("ranges") or []: + if rng.get("type") != "ECOSYSTEM": + continue + closing = _closing_fix(rng.get("events") or [], pin) + if closing is not None: + fixed.append(_parse(closing)) + return fixed + + +# -------------------------------------------------------------------------- +# Cache +# -------------------------------------------------------------------------- + +class _Cache: + def __init__(self, directory: Optional[Path], now: float): + self.path = directory / CACHE_FILENAME if directory else None + self.now = now + self.queries: Dict[str, Dict[str, Any]] = {} + self.vulns: Dict[str, Dict[str, Any]] = {} + self._load() + + def _load(self) -> None: + if self.path is None: + return + try: + data = json.loads(self.path.read_text()) + except (OSError, ValueError): + return + if not isinstance(data, dict) or data.get("schema") != _CACHE_SCHEMA: + return + if isinstance(data.get("queries"), dict): + self.queries = data["queries"] + if isinstance(data.get("vulns"), dict): + self.vulns = data["vulns"] + + def _fresh(self, entry: Any) -> bool: + try: + return self.now - float(entry["fetched_at"]) < CACHE_TTL_SECONDS + except (TypeError, KeyError, ValueError): + return False + + def query(self, key: str) -> Optional[List[str]]: + entry = self.queries.get(key) + if self._fresh(entry) and isinstance(entry.get("ids"), list): + return [str(i) for i in entry["ids"]] + return None + + def vuln(self, vuln_id: str) -> Optional[Dict[str, Any]]: + entry = self.vulns.get(vuln_id) + if self._fresh(entry) and isinstance(entry.get("record"), dict): + return entry["record"] + return None + + def put_query(self, key: str, ids: List[str]) -> None: + self.queries[key] = {"fetched_at": self.now, "ids": ids} + + def put_vuln(self, vuln_id: str, record: Dict[str, Any]) -> None: + self.vulns[vuln_id] = {"fetched_at": self.now, "record": record} + + def save(self) -> None: + """Write-tmp-then-rename, dropping expired entries. Never raises.""" + if self.path is None: + return + payload = { + "schema": _CACHE_SCHEMA, + "queries": {k: v for k, v in self.queries.items() if self._fresh(v)}, + "vulns": {k: v for k, v in self.vulns.items() if self._fresh(v)}, + } + tmp = self.path.with_suffix(".json.tmp") + try: + tmp.write_text(json.dumps(payload, separators=(",", ":"))) + tmp.replace(self.path) + except OSError: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + + +def _default_session() -> Any: + """The HTTP client used when none is injected. A seam the test suite stubs. + + A pooled ``Session`` rather than bare ``requests``: an old pin can name well + over a hundred advisories (``django==2.0.0`` plus ``pillow==6.0.0`` is 140), + and a fresh TLS handshake per record both halves throughput and adds + connect stalls that eat the time budget. The pool is sized to the worker + count so parallel fetches do not queue for a connection. + """ + session = requests.Session() + adapter = requests.adapters.HTTPAdapter( + pool_connections=1, pool_maxsize=_FETCH_WORKERS + ) + session.mount("https://", adapter) + return session + + +def _default_cache_dir() -> Optional[Path]: + # Imported lazily so tests that stub telemetry's config dir apply here too. + from . import telemetry + + return telemetry.get_config_dir() + + +# -------------------------------------------------------------------------- +# Network +# -------------------------------------------------------------------------- + +class _Client: + def __init__(self, session: Any, budget_seconds: float): + self.session = session + self.deadline = time.monotonic() + budget_seconds + self.headers = {"User-Agent": f"aisbom-cli/{_tool_version()}"} + + def _timeout(self) -> float: + remaining = self.deadline - time.monotonic() + if remaining <= 0: + raise OsvUnavailable("OSV lookup exceeded its time budget") + return min(REQUEST_TIMEOUT_SECONDS, remaining) + + def _call(self, method: str, url: str, **kwargs: Any) -> Any: + """One request, retried on transient failure, returning parsed JSON. + + OSV sits behind a CDN that occasionally answers a single request with + a 503 — observed live on 1 of 140 record fetches for a real pin set. + Without a retry, that one blip would cost every CVE statement in the + run. Only failures a retry can fix are retried (connection errors, + timeouts, 429, 5xx); a 404 or a malformed body fails at once. Every + attempt and every pause draws on the shared budget. + """ + for attempt in range(_ATTEMPTS): + last = attempt == _ATTEMPTS - 1 + try: + response = getattr(self.session, method)( + url, timeout=self._timeout(), headers=self.headers, **kwargs + ) + except (requests.ConnectionError, requests.Timeout): + if last: + raise + else: + if response.status_code in _RETRYABLE_STATUS and not last: + pass + else: + response.raise_for_status() + return response.json() + self._pause(_BACKOFF_SECONDS * (attempt + 1)) + raise OsvUnavailable("OSV retries exhausted") # pragma: no cover - loop returns or raises + + def _pause(self, seconds: float) -> None: + if self.deadline - time.monotonic() <= seconds: + raise OsvUnavailable("OSV lookup exceeded its time budget") + time.sleep(seconds) + + def querybatch(self, queries: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + payload = self._call( + "post", f"{OSV_API}/querybatch", json={"queries": queries} + ) + results = payload.get("results") if isinstance(payload, dict) else None + if not isinstance(results, list) or len(results) != len(queries): + raise OsvUnavailable("OSV returned a malformed querybatch response") + return results + + def vulnerability(self, vuln_id: str) -> Dict[str, Any]: + record = self._call("get", f"{OSV_API}/vulns/{vuln_id}") + if not isinstance(record, dict) or record.get("id") != vuln_id: + raise OsvUnavailable(f"OSV returned a malformed record for {vuln_id}") + return record + + +def _fetch_records(client: _Client, vuln_ids: List[str]) -> Dict[str, Dict[str, Any]]: + """Fetch full advisory records in parallel. All or nothing. + + The first failure cancels whatever has not started and re-raises, so the + caller degrades exactly as it would for a serial fetch and a partial set of + records never reaches statement building. Every request still draws its + timeout from the shared budget, so parallelism cannot stretch the deadline. + """ + if not vuln_ids: + return {} + records: Dict[str, Dict[str, Any]] = {} + with ThreadPoolExecutor(max_workers=min(_FETCH_WORKERS, len(vuln_ids))) as pool: + futures = {pool.submit(client.vulnerability, v): v for v in vuln_ids} + try: + for future in as_completed(futures): + records[futures[future]] = future.result() + except BaseException: + for future in futures: + future.cancel() + raise + return records + + +def _ids_from(result: Any) -> Tuple[List[str], Optional[str]]: + if not isinstance(result, dict): + raise OsvUnavailable("OSV returned a malformed querybatch result") + ids = [] + for vuln in result.get("vulns") or []: + if not isinstance(vuln, dict) or not isinstance(vuln.get("id"), str): + raise OsvUnavailable("OSV returned a vulnerability with no id") + ids.append(vuln["id"]) + token = result.get("next_page_token") + return ids, token if isinstance(token, str) and token else None + + +def _query_ids(client: _Client, pins: List[Tuple[str, str]]) -> Dict[Tuple[str, str], List[str]]: + """Advisory ids OSV reports for each (canonical name, version).""" + found: Dict[Tuple[str, str], List[str]] = {} + for start in range(0, len(pins), _BATCH_SIZE): + chunk = pins[start:start + _BATCH_SIZE] + pending = [(pin, None) for pin in chunk] + pages = 0 + while pending: + pages += 1 + if pages > _MAX_PAGES: + raise OsvUnavailable("OSV pagination did not terminate") + queries = [] + for (name, version), token in pending: + query: Dict[str, Any] = { + "package": {"name": name, "ecosystem": ECOSYSTEM}, + "version": version, + } + if token: + query["page_token"] = token + queries.append(query) + results = client.querybatch(queries) + next_pending = [] + for (pin, _), result in zip(pending, results): + ids, token = _ids_from(result) + found.setdefault(pin, []).extend(ids) + if token: + next_pending.append((pin, token)) + pending = next_pending + return {pin: list(dict.fromkeys(ids)) for pin, ids in found.items()} + + +# -------------------------------------------------------------------------- +# Statements +# -------------------------------------------------------------------------- + +def _truncate(text: str, limit: int = 1000) -> str: + text = " ".join(text.split()) + return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" + + +def _statement( + name: str, + version: str, + ref: str, + records: List[Dict[str, Any]], +) -> VexStatement: + """One statement for one CVE on one dependency, merging twin OSV records.""" + primary, _ = vulnerability_ids(records[0]) + aliases = sorted({a for r in records for a in vulnerability_ids(r)[1]}) + lead = records[0] + confirmed = [r for r in records if version_affected(r, name, version) is True] + status = STATUS_AFFECTED if confirmed else STATUS_UNDER_INVESTIGATION + source_record = confirmed[0] if confirmed else lead + + summary = str(source_record.get("summary") or "").strip() + details = str(source_record.get("details") or "").strip() + # Deduplicated as versions, not strings: twin records may spell one fix + # as "2.20" and "2.20.0". + fixed = [str(v) for v in sorted( + {v for r in records for v in _fixed_versions(r, name, version)} + )] + + if fixed: + action = ( + f"Upgrade {name} from {version} to a version outside the affected " + f"range (fixed in: {', '.join(fixed)})." + ) + else: + action = ( + f"No fixed version of {name} is published for this advisory. Review " + "it and consider replacing or isolating the dependency." + ) + + if status == STATUS_AFFECTED: + notes = ( + f"{name}=={version} is inside the affected range published by OSV " + f"({source_record['id']}). AIsbom does not observe whether the " + "vulnerable code is reachable from your application." + ) + else: + notes = ( + f"OSV lists {name}=={version} as affected by " + f"{', '.join(r['id'] for r in records)}, but AIsbom could not confirm " + "it against the advisory's published version ranges." + ) + + finding = FindingClass( + id=primary, + title=_truncate(summary or primary, 200), + description=_truncate(details or summary or primary), + action=action, + formats=frozenset(), + aliases=tuple(aliases), + reference_url=f"{OSV_VULNERABILITY_URL}{source_record['id']}", + source_name=OSV_SOURCE_NAME, + ) + return VexStatement( + finding_class=finding, + product_ref=ref, + product_id="", + product_hash=None, + status=status, + status_notes=notes, + action_statement=action if status == STATUS_AFFECTED else None, + ) + + +def lookup_dependency_statements( + dependencies: Sequence[Dict[str, Any]], + *, + session: Any = None, + cache_dir: Any = _DEFAULT_CACHE_DIR, + now: Optional[float] = None, + budget_seconds: float = LOOKUP_BUDGET_SECONDS, +) -> OsvLookupResult: + """CVE-keyed statements for a scan's dependency components. Never raises. + + ``dependencies`` is the scanner's list, in the order the SBOM emits them — + the index is part of each component's ``bom-ref``. + """ + result = OsvLookupResult() + targets: List[Tuple[int, Dict[str, Any], Tuple[str, str]]] = [] + for index, dep in enumerate(dependencies): + if not dep.get("pinned"): + result.skipped_unpinned += 1 + continue + pin = (canonicalize_name(str(dep["name"])), str(dep["version"])) + targets.append((index, dep, pin)) + result.queried = len(targets) + if not targets: + return result + + directory = _default_cache_dir() if cache_dir is _DEFAULT_CACHE_DIR else cache_dir + cache = _Cache(Path(directory) if directory else None, + time.time() if now is None else now) + + try: + client = _Client( + session if session is not None else _default_session(), budget_seconds + ) + + unique_pins = list(dict.fromkeys(pin for _, _, pin in targets)) + ids_by_pin: Dict[Tuple[str, str], List[str]] = {} + uncached = [] + for pin in unique_pins: + cached = cache.query(f"{pin[0]}=={pin[1]}") + if cached is None: + uncached.append(pin) + else: + ids_by_pin[pin] = cached + if uncached: + fetched = _query_ids(client, uncached) + for pin in uncached: + ids_by_pin[pin] = fetched.get(pin, []) + + records: Dict[str, Dict[str, Any]] = {} + missing = [] + for vuln_id in dict.fromkeys(i for ids in ids_by_pin.values() for i in ids): + record = cache.vuln(vuln_id) + if record is None: + missing.append(vuln_id) + else: + records[vuln_id] = record + for vuln_id, record in _fetch_records(client, missing).items(): + cache.put_vuln(vuln_id, record) + records[vuln_id] = record + + # Only now is the answer complete; cache query results together so a + # failure part-way never leaves a pin cached against missing records. + for pin in uncached: + cache.put_query(f"{pin[0]}=={pin[1]}", ids_by_pin[pin]) + + for index, dep, pin in targets: + grouped: Dict[str, List[Dict[str, Any]]] = {} + for vuln_id in ids_by_pin[pin]: + record = records[vuln_id] + if record.get("withdrawn"): + continue + primary, _ = vulnerability_ids(record) + grouped.setdefault(primary, []).append(record) + ref = dependency_bom_ref(index, dep) + for primary in sorted(grouped): + result.statements.append( + _statement(str(dep["name"]), pin[1], ref, grouped[primary]) + ) + except OsvUnavailable as exc: + result.statements = [] + result.error = str(exc) + except requests.RequestException as exc: + result.statements = [] + result.error = f"OSV request failed ({type(exc).__name__})" + except Exception as exc: # noqa: BLE001 - enrichment must never break a scan + result.statements = [] + result.error = f"OSV lookup failed ({type(exc).__name__})" + finally: + cache.save() + + return result diff --git a/aisbom/scanner.py b/aisbom/scanner.py index 5e95939..5d7a894 100644 --- a/aisbom/scanner.py +++ b/aisbom/scanner.py @@ -1928,10 +1928,20 @@ def _parse_requirements(self, path: Path): specs = list(req.specifier) if req.specifier else [] if specs: version = specs[0].version + # Only a single `==`/`===` without a wildcard names the + # version actually installed. `version` above is also set + # for `>=2.0`, so the OSV lookup (#128) keys on this flag + # rather than on whether a version string exists. + pinned = ( + len(specs) == 1 + and specs[0].operator in ("==", "===") + and "*" not in specs[0].version + ) self.dependencies.append({ "name": req.name, "version": version, - "type": "library" + "type": "library", + "pinned": pinned, }) except Exception as e: self.errors.append({"file": str(path), "error": str(e)}) diff --git a/aisbom/vex.py b/aisbom/vex.py index 2a912e9..5a4ab8e 100644 --- a/aisbom/vex.py +++ b/aisbom/vex.py @@ -57,10 +57,12 @@ by its successor. That makes the policy mechanically enforced rather than a comment someone has to remember. -When OSV/CVE mapping for the dependency components lands, those statements join -the same ``statements`` list with no change visible to a consumer — the -emitters take a list of :class:`VexStatement` and do not care where one came -from. +OSV-sourced CVE statements for the pinned dependency components +(:mod:`aisbom.osv`) join the same ``statements`` list with no change visible to +a consumer of these classes — the emitters take a list of :class:`VexStatement` +and do not care where one came from. Such a statement's class carries its own +``reference_url`` and ``source_name``, so it resolves to OSV rather than under +:data:`VEX_NAMESPACE`. Scoping of negative statements ------------------------------ @@ -159,10 +161,16 @@ class cannot be added, renamed or dropped in only one of those places. #: Real CVE/GHSA identifiers for the same issue. Empty for every class #: today — see the module docstring for why. aliases: tuple = () + #: Where the identifier resolves, for a statement AIsbom did not define + #: (an OSV-sourced CVE, see :mod:`aisbom.osv`). Unset for every class in + #: this module, which resolve under :data:`VEX_NAMESPACE`. + reference_url: Optional[str] = None + #: Who published the identifier — the CycloneDX `source.name`. + source_name: str = VEX_AUTHOR @property def iri(self) -> str: - return f"{VEX_NAMESPACE}{self.id}" + return self.reference_url or f"{VEX_NAMESPACE}{self.id}" VEX_FINDING_CLASSES: tuple = ( @@ -810,6 +818,26 @@ def _timestamp(value: Optional[datetime] = None) -> str: return (value or datetime.now(timezone.utc)).isoformat(timespec="seconds") +# Who publishes an alias, by identifier prefix — the CycloneDX +# `references[].source.name`. Every alias was previously attributed to NVD, +# which was only ever true of a CVE: a retired AIsbom class aliased through a +# rename is AIsbom's, and OSV-sourced dependency advisories carry GHSA and +# PYSEC ids alongside their CVE. +_ALIAS_SOURCES = ( + ("AISBOM-", VEX_AUTHOR), + ("CVE-", "NVD"), + ("GHSA-", "GitHub Advisory Database"), + ("PYSEC-", "PyPA Advisory Database"), +) + + +def _alias_source(identifier: str) -> str: + for prefix, name in _ALIAS_SOURCES: + if identifier.startswith(prefix): + return name + return "OSV" + + def _product_id(sbom_serial: str, ref: str) -> str: """Address a component inside the SBOM this scan produced. @@ -922,7 +950,7 @@ def generate_cyclonedx_vex( entry: Dict[str, Any] = { "bom-ref": f"vex-{class_id}-{status}", "id": class_id, - "source": {"name": VEX_AUTHOR, "url": cls.iri}, + "source": {"name": cls.source_name, "url": cls.iri}, "description": cls.description, "detail": cls.title, "analysis": analysis, @@ -930,7 +958,8 @@ def generate_cyclonedx_vex( } if cls.aliases: entry["references"] = [ - {"id": alias, "source": {"name": "NVD"}} for alias in cls.aliases + {"id": alias, "source": {"name": _alias_source(alias)}} + for alias in cls.aliases ] if status == STATUS_AFFECTED: entry["recommendation"] = cls.action @@ -987,9 +1016,10 @@ def finding_classes_markdown() -> str: "the `id` of a CycloneDX VEX entry, and resolve under " f"`{VEX_NAMESPACE}`.", "", - "CVE-keyed statements about a project's Python dependencies are a", - "separate, additive concern and arrive with OSV mapping; they will join", - "the same document without changing anything below.", + "CVE-keyed statements about a project's pinned Python dependencies are", + "a separate, additive concern: they come from OSV, carry the advisory's", + "CVE (or GHSA/PYSEC id) and resolve under `https://osv.dev/vulnerability/`.", + "They join the same document without changing anything below.", "", "## Compatibility policy", "", diff --git a/docs/air-gapped-guide.md b/docs/air-gapped-guide.md index 27858cd..77f7cfa 100644 --- a/docs/air-gapped-guide.md +++ b/docs/air-gapped-guide.md @@ -77,6 +77,8 @@ The tool produces two outputs: 2. **SBOM Report (`sbom.json`):** A CycloneDX JSON file generated in the working directory. * This file is **static plain text**. It is safe to egress back to "Zone A" for ingestion into your central vulnerability dashboard. +> **VEX in Zone B:** `scan --vex` normally looks up `requirements.txt` pins in the public OSV database. On an air-gapped host that lookup fails and the scan carries on without it — you get a warning, and the VEX documents contain the model finding statements but no dependency CVE statements. To skip the attempt entirely, pass `--no-osv` (or set `AISBOM_NO_OSV=1`). + --- ## 3. Why this matters diff --git a/docs/vex-finding-classes.md b/docs/vex-finding-classes.md index c355e3f..2e7efa7 100644 --- a/docs/vex-finding-classes.md +++ b/docs/vex-finding-classes.md @@ -10,9 +10,10 @@ rather than a published component with a patchable defect. They appear as the `vulnerability.name` of an OpenVEX statement and the `id` of a CycloneDX VEX entry, and resolve under `https://aisbom.io/vex/`. -CVE-keyed statements about a project's Python dependencies are a -separate, additive concern and arrive with OSV mapping; they will join -the same document without changing anything below. +CVE-keyed statements about a project's pinned Python dependencies are +a separate, additive concern: they come from OSV, carry the advisory's +CVE (or GHSA/PYSEC id) and resolve under `https://osv.dev/vulnerability/`. +They join the same document without changing anything below. ## Compatibility policy diff --git a/tests/conftest.py b/tests/conftest.py index 40ae941..1b72928 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -40,6 +40,27 @@ def _stub_telemetry(request, monkeypatch): monkeypatch.setattr("aisbom.telemetry.is_ci", lambda: False) +class _OfflineOSV: + """Stands in for `requests` inside aisbom.osv: every call is a failure.""" + + def post(self, *args, **kwargs): + raise ConnectionError("OSV is not reachable from the test suite") + + get = post + + +@pytest.fixture(autouse=True) +def _stub_osv_network(monkeypatch): + """Keep `scan --vex` from ever reaching api.osv.dev during tests. + + Only the *default* session is replaced, so tests that inject a fake OSV + (tests/test_osv.py, the CLI integration tests) are unaffected, and a test + that forgets to inject one sees the documented degraded behaviour rather + than a live request. The air-gap test restores the real client on purpose. + """ + monkeypatch.setattr("aisbom.osv._default_session", lambda: _OfflineOSV()) + + @pytest.fixture(autouse=True) def _stub_version_check(monkeypatch): """Auto-stub the background update check for every test. diff --git a/tests/test_action_platform_upload.py b/tests/test_action_platform_upload.py index b33b277..81b80dd 100644 --- a/tests/test_action_platform_upload.py +++ b/tests/test_action_platform_upload.py @@ -527,3 +527,48 @@ def test_upload_reports_the_vex_document_count(sbom_with_vex: Path, capsys): ) out = capsys.readouterr().out assert "vex-documents=2" in out + + +def _oversized_vex(tmp_path: Path) -> Path: + """VEX siblings whose envelope exceeds the receiver's 1 MiB cap. + + Realistic, not contrived: dependency CVE statements run ~2 KB each across + the two flavors, so a few hundred advisories on old pins crosses the line. + """ + sbom = tmp_path / "sbom.json" + sbom.write_text(json.dumps({"bomFormat": "CycloneDX", "components": []})) + filler = "x" * 2048 + statements = [{"status": "affected", "status_notes": filler} for _ in range(300)] + (tmp_path / "sbom.openvex.json").write_text(json.dumps({"statements": statements})) + (tmp_path / "sbom.vex.cdx.json").write_text(json.dumps({"vulnerabilities": statements})) + return sbom + + +def test_an_oversized_envelope_falls_back_to_the_bare_sbom(tmp_path: Path, capsys): + """A 413 would lose the inventory entry to carry supplementary documents.""" + sbom = _oversized_vex(tmp_path) + body = platform_upload.build_request_body(str(sbom)) + assert body == sbom.read_bytes() + assert "VEX documents omitted" in capsys.readouterr().out + + +def test_upload_reports_zero_vex_documents_when_they_were_omitted(tmp_path: Path, capsys): + sbom = _oversized_vex(tmp_path) + captured = {} + + def fake_post(url, **kwargs): + captured["data"] = kwargs.get("data") + return _mock_response(200, "ok") + + with patch("requests.post", side_effect=fake_post): + rc = platform_upload.upload( + sbom_path=str(sbom), + token="tok", + platform_url="https://app.aisbom.io", + trigger="push", + fail_on_error=True, + env={}, + ) + assert rc == 0 + assert len(captured["data"]) <= platform_upload.MAX_BODY_BYTES + assert "vex-documents=0" in capsys.readouterr().out diff --git a/tests/test_cli_integration.py b/tests/test_cli_integration.py index 0416f9e..8326235 100644 --- a/tests/test_cli_integration.py +++ b/tests/test_cli_integration.py @@ -471,3 +471,143 @@ def test_vex_baseline_turns_a_removed_finding_into_fixed(tmp_path): assert len(fixed) == 1 assert fixed[0]["vulnerability"]["name"] == "AISBOM-PICKLE-RCE" assert "fixed since baseline" in result.output + + +# --------------------------------------------------------------------------- +# #128 — CVE-keyed statements for requirements.txt pins, via OSV. +# --------------------------------------------------------------------------- + + +def _osv_tree(tmp_path, requirements="requests==2.19.0\ntorch>=2.0\n"): + _write_malicious_pt(tmp_path / "mock_malware.pt") + create_mock_gguf(tmp_path) + (tmp_path / "requirements.txt").write_text(requirements) + return tmp_path / "sbom.json" + + +def _fake_osv(monkeypatch): + from tests.test_osv import FakeOSV, _requests_advisory + + fake = FakeOSV( + [_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}, + ) + monkeypatch.setattr("aisbom.osv._default_session", lambda: fake) + return fake + + +def _finding_class_statements(openvex): + return sorted( + (s["vulnerability"]["name"], s["products"][0]["@id"].partition("#")[2], + s["status"]) + for s in openvex["statements"] + if s["vulnerability"]["name"].startswith("AISBOM-") + ) + + +def test_vex_carries_cve_statements_for_pinned_dependencies(tmp_path, monkeypatch): + output_path = _osv_tree(tmp_path) + fake = _fake_osv(monkeypatch) + + result = runner.invoke(app, ["scan", str(tmp_path), "--output", str(output_path), "--vex"]) + assert result.exit_code == 2, result.output + output = " ".join(result.output.split()) # Rich wraps long lines + assert "1 CVE statement(s) across 1 pinned dependency" in output + assert "1 unpinned dependency skipped" in output + + sbom = json.loads(output_path.read_text()) + refs = {c["bom-ref"] for c in sbom["components"]} + openvex = json.loads((tmp_path / "sbom.openvex.json").read_text()) + cdx = json.loads((tmp_path / "sbom.vex.cdx.json").read_text()) + + [cve] = [s for s in openvex["statements"] + if s["vulnerability"]["name"] == "CVE-2018-18074"] + assert cve["status"] == "affected" + assert cve["products"][0]["@id"].partition("#")[2] in refs + # The model findings are still in the same document. + assert any(s["vulnerability"]["name"] == "AISBOM-PICKLE-RCE" + and s["status"] == "affected" for s in openvex["statements"]) + + [entry] = [v for v in cdx["vulnerabilities"] if v["id"] == "CVE-2018-18074"] + assert all(a["ref"] in refs for a in entry["affects"]) + # Only the exact pin was sent to OSV. + assert [q["package"]["name"] for q in fake.posts[0]["queries"]] == ["requests"] + + +def test_no_osv_flag_makes_no_request(tmp_path, monkeypatch): + output_path = _osv_tree(tmp_path) + fake = _fake_osv(monkeypatch) + result = runner.invoke( + app, ["scan", str(tmp_path), "--output", str(output_path), "--vex", "--no-osv"] + ) + assert result.exit_code == 2, result.output + assert fake.calls == 0 + openvex = json.loads((tmp_path / "sbom.openvex.json").read_text()) + assert not any(s["vulnerability"]["name"].startswith("CVE-") + for s in openvex["statements"]) + + +def test_no_osv_env_var_makes_no_request(tmp_path, monkeypatch): + output_path = _osv_tree(tmp_path) + fake = _fake_osv(monkeypatch) + monkeypatch.setenv("AISBOM_NO_OSV", "1") + result = runner.invoke(app, ["scan", str(tmp_path), "--output", str(output_path), "--vex"]) + assert result.exit_code == 2, result.output + assert fake.calls == 0 + + +def test_a_scan_without_vex_never_contacts_osv(tmp_path, monkeypatch): + output_path = _osv_tree(tmp_path) + fake = _fake_osv(monkeypatch) + result = runner.invoke(app, ["scan", str(tmp_path), "--output", str(output_path)]) + assert result.exit_code == 2, result.output + assert fake.calls == 0 + + +def test_air_gapped_scan_still_succeeds_with_every_model_finding(tmp_path, monkeypatch): + """Verified, not assumed: the real HTTP client, with the network cut. + + The conftest stub is undone so `requests` genuinely tries to connect, and + the socket layer refuses. The run must exit exactly as a --no-osv run of + the same tree does, carrying the identical finding-class statements. + """ + import socket + + import requests as real_requests + + offline_dir = tmp_path / "offline" + disabled_dir = tmp_path / "disabled" + offline_dir.mkdir() + disabled_dir.mkdir() + + disabled_out = _osv_tree(disabled_dir) + baseline = runner.invoke( + app, ["scan", str(disabled_dir), "--output", str(disabled_out), "--vex", "--no-osv"] + ) + + attempts = [] + + def refuse(*args, **kwargs): + attempts.append(args) + raise OSError("network is unreachable (air-gapped test)") + + monkeypatch.setattr("aisbom.osv._default_session", lambda: real_requests) + monkeypatch.setattr(socket, "getaddrinfo", refuse) + monkeypatch.setattr(socket.socket, "connect", refuse) + + offline_out = _osv_tree(offline_dir) + offline = runner.invoke( + app, ["scan", str(offline_dir), "--output", str(offline_out), "--vex"] + ) + + assert attempts, "the lookup never tried the network, so nothing was proven" + assert offline.exit_code == baseline.exit_code == 2, offline.output + assert "VEX carries no CVE statements" in " ".join(offline.output.split()) + + offline_vex = json.loads((offline_dir / "sbom.openvex.json").read_text()) + disabled_vex = json.loads((disabled_dir / "sbom.openvex.json").read_text()) + assert _finding_class_statements(offline_vex) == _finding_class_statements(disabled_vex) + assert _finding_class_statements(offline_vex), "expected model findings" + assert not any(s["vulnerability"]["name"].startswith("CVE-") + for s in offline_vex["statements"]) + assert json.loads((offline_dir / "sbom.vex.cdx.json").read_text())["vulnerabilities"] diff --git a/tests/test_cyclonedx_gen.py b/tests/test_cyclonedx_gen.py index 5fab2dd..a1d1026 100644 --- a/tests/test_cyclonedx_gen.py +++ b/tests/test_cyclonedx_gen.py @@ -192,3 +192,22 @@ def test_an_empty_scan_still_produces_a_valid_document(): doc = json.loads(build_cyclonedx_json({"artifacts": [], "dependencies": []})) assert doc["bomFormat"] == "CycloneDX" assert doc["metadata"]["tools"] + + +def test_dependency_bom_refs_are_stable_across_runs(): + """CVE-keyed VEX statements (#128) join on these, so they cannot be random.""" + first = _doc() + second = _doc() + assert _by_name(first, "torch")["bom-ref"] == "dependency-0-torch" + assert _by_name(first, "numpy")["bom-ref"] == "dependency-1-numpy" + assert [c["bom-ref"] for c in first["components"]] == \ + [c["bom-ref"] for c in second["components"]] + + +def test_the_same_package_in_two_requirements_files_keeps_distinct_refs(): + doc = _doc(dependencies=[ + {"name": "torch", "version": "2.1.0"}, + {"name": "torch", "version": "2.1.0"}, + ]) + refs = [c["bom-ref"] for c in doc["components"] if c["name"] == "torch"] + assert sorted(refs) == ["dependency-0-torch", "dependency-1-torch"] diff --git a/tests/test_osv.py b/tests/test_osv.py new file mode 100644 index 0000000..477ce44 --- /dev/null +++ b/tests/test_osv.py @@ -0,0 +1,809 @@ +"""OSV lookup for pinned requirements.txt dependencies — CVE-keyed VEX (#128). + +The OSV API is never contacted from this suite. Every test drives a fake +session whose responses are shaped like real api.osv.dev payloads, so hit, +miss, malformed, failure, pagination and cache reuse are all exercised +deterministically and offline. +""" + +import json +import time +from pathlib import Path + +import pytest +import requests +from cyclonedx.schema import SchemaVersion +from cyclonedx.validation.json import JsonStrictValidator +from jsonschema import Draft202012Validator + +from aisbom import osv +from aisbom.vex import generate_cyclonedx_vex, generate_openvex + +_REAL_DEFAULT_SESSION = osv._default_session + + +@pytest.fixture(autouse=True) +def _no_retry_pause(monkeypatch): + """Retries are exercised here; their real-time backoff is not.""" + monkeypatch.setattr(osv, "_BACKOFF_SECONDS", 0) + +SERIAL = "urn:uuid:3e671687-395b-41f5-a30f-a58921a69b79" +_OPENVEX_SCHEMA = Path(__file__).parent / "schemas" / "openvex-0.2.0.schema.json" + + +# -------------------------------------------------------------------------- +# Real OSV record shapes (trimmed to the fields AIsbom reads) +# -------------------------------------------------------------------------- + +def _requests_advisory(): + """GHSA-x84v-xcm2-53pg as OSV serves it: a GHSA id with a CVE alias.""" + return { + "id": "GHSA-x84v-xcm2-53pg", + "summary": "Insufficiently Protected Credentials in Requests", + "details": "The Requests package before 2.20.0 sends an HTTP " + "Authorization header to an http URI upon receiving a " + "same-hostname https-to-http redirect.", + "aliases": ["CVE-2018-18074"], + "modified": "2024-09-26T20:11:51Z", + "affected": [ + { + "package": {"ecosystem": "PyPI", "name": "requests", + "purl": "pkg:pypi/requests"}, + "ranges": [ + {"type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "2.20.0"}]} + ], + } + ], + } + + +def _pysec_twin(): + """PYSEC-2018-28 — the same CVE published under a second OSV id.""" + return { + "id": "PYSEC-2018-28", + "details": "The Requests package before 2.20.0 ...", + "aliases": ["CVE-2018-18074", "GHSA-x84v-xcm2-53pg"], + "modified": "2021-06-10T06:51:37Z", + "affected": [ + { + "package": {"ecosystem": "PyPI", "name": "requests"}, + "ranges": [ + {"type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "2.20.0"}]} + ], + "versions": ["2.19.0", "2.19.1"], + } + ], + } + + +class FakeResponse: + def __init__(self, payload, status=200): + self._payload = payload + self.status_code = status + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"{self.status_code}") + + def json(self): + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +class FakeOSV: + """A stand-in for api.osv.dev that records every call made to it.""" + + def __init__(self, records=(), matches=None, fail=None, batch_payload=None, + pages=None): + self.records = {r["id"]: r for r in records} + # {(name, version): [ids]} — what querybatch reports as matching. + self.matches = matches or {} + self.fail = fail + self.batch_payload = batch_payload + self.pages = pages or {} + self.posts = [] + self.gets = [] + + def post(self, url, json=None, timeout=None, headers=None): + self.posts.append(json) + if self.fail: + raise self.fail + if self.batch_payload is not None: + return FakeResponse(self.batch_payload) + results = [] + for query in json["queries"]: + token = query.get("page_token") + key = (query["package"]["name"], query["version"]) + if token: + ids = self.pages[token] + results.append({"vulns": [{"id": i} for i in ids]}) + continue + ids = self.matches.get(key, []) + entry = {"vulns": [{"id": i} for i in ids]} if ids else {} + if key in self.pages: + entry["next_page_token"] = f"tok-{key[0]}" + results.append(entry) + return FakeResponse({"results": results}) + + def get(self, url, timeout=None, headers=None): + self.gets.append(url) + if self.fail: + raise self.fail + vuln_id = url.rsplit("/", 1)[-1] + if vuln_id not in self.records: + return FakeResponse({"code": 5, "message": "Bug not found"}, 404) + return FakeResponse(self.records[vuln_id]) + + @property + def calls(self): + return len(self.posts) + len(self.gets) + + +def _dep(name, version, pinned=True): + return {"name": name, "version": version, "type": "library", "pinned": pinned} + + +def _lookup(deps, session, cache_dir=None, **kw): + return osv.lookup_dependency_statements( + deps, session=session, cache_dir=cache_dir, **kw + ) + + +# -------------------------------------------------------------------------- +# Version-range matching — the part that must not be wrong in either direction +# -------------------------------------------------------------------------- + +def _range_record(events, name="demo-pkg", versions=None, ecosystem="PyPI", + range_type="ECOSYSTEM"): + affected = {"package": {"ecosystem": ecosystem, "name": name}, + "ranges": [{"type": range_type, "events": events}]} + if versions is not None: + affected["versions"] = versions + return {"id": "GHSA-demo", "affected": [affected]} + + +@pytest.mark.parametrize("version, expected", [ + ("2.2.9", False), # just below introduced + ("2.3.0", True), # exactly at introduced + ("2.19.1", True), # inside + ("2.20.0rc1", True), # a pre-release of the fix is still before it + ("2.20.0", False), # exactly at fixed + ("2.20.1", False), # after fixed +]) +def test_introduced_fixed_boundaries(version, expected): + record = _range_record([{"introduced": "2.3.0"}, {"fixed": "2.20.0"}]) + assert osv.version_affected(record, "demo-pkg", version) is expected + + +@pytest.mark.parametrize("version, expected", [ + ("1.4.2", True), # exactly at last_affected is still affected + ("1.4.3", False), # one past it is not + ("0.9", True), +]) +def test_last_affected_is_inclusive(version, expected): + record = _range_record([{"introduced": "0"}, {"last_affected": "1.4.2"}]) + assert osv.version_affected(record, "demo-pkg", version) is expected + + +@pytest.mark.parametrize("version, expected", [ + ("1.0", True), ("1.7", False), ("2.1", True), ("2.2", False), ("0.5", False), +]) +def test_disjoint_ranges_leave_the_gap_unaffected(version, expected): + record = _range_record([ + {"introduced": "1.0"}, {"fixed": "1.5"}, + {"introduced": "2.0"}, {"fixed": "2.2"}, + ]) + assert osv.version_affected(record, "demo-pkg", version) is expected + + +def test_events_are_evaluated_in_version_order_not_listed_order(): + record = _range_record([{"fixed": "1.5"}, {"introduced": "1.0"}]) + assert osv.version_affected(record, "demo-pkg", "1.2") is True + assert osv.version_affected(record, "demo-pkg", "1.5") is False + + +@pytest.mark.parametrize("version", ["0.dev0", "0rc1", "0a1", "0"]) +def test_introduced_zero_covers_prereleases_below_release_zero(version): + """OSV's "0" means "from the first version", not PEP 440's release 0.""" + record = _range_record([{"introduced": "0"}, {"fixed": "2.0"}]) + assert osv.version_affected(record, "demo-pkg", version) is True + + +def test_introduced_zero_with_no_fix_affects_every_version(): + record = _range_record([{"introduced": "0"}]) + assert osv.version_affected(record, "demo-pkg", "99.0") is True + + +def test_explicit_versions_list_matches_without_ranges(): + record = {"id": "PYSEC-demo", "affected": [ + {"package": {"ecosystem": "PyPI", "name": "demo-pkg"}, + "versions": ["1.0.0", "1.0.1"]} + ]} + assert osv.version_affected(record, "demo-pkg", "1.0.1") is True + # PEP 440 equality, not string equality. + assert osv.version_affected(record, "demo-pkg", "1.0.1.0") is True + assert osv.version_affected(record, "demo-pkg", "1.0.2") is False + + +def test_package_names_compare_canonically(): + record = _range_record([{"introduced": "0"}, {"fixed": "2.0"}], + name="Demo_Pkg") + assert osv.version_affected(record, "demo.pkg", "1.0") is True + + +def test_an_entry_for_a_different_package_never_matches(): + record = _range_record([{"introduced": "0"}], name="requests-toolbelt") + assert osv.version_affected(record, "requests", "1.0") is None + + +def test_a_non_pypi_entry_is_not_evaluated(): + record = _range_record([{"introduced": "0"}], ecosystem="npm") + assert osv.version_affected(record, "demo-pkg", "1.0") is None + + +def test_git_ranges_cannot_be_evaluated_locally(): + record = _range_record([{"introduced": "abc123"}], range_type="GIT") + assert osv.version_affected(record, "demo-pkg", "1.0") is None + + +def test_an_unparseable_pin_is_undetermined_rather_than_clean(): + record = _range_record([{"introduced": "0"}, {"fixed": "2.0"}]) + assert osv.version_affected(record, "demo-pkg", "not-a-version") is None + + +def test_an_unparseable_event_version_is_undetermined(): + record = _range_record([{"introduced": "garbage!"}, {"fixed": "2.0"}]) + assert osv.version_affected(record, "demo-pkg", "1.0") is None + + +# -------------------------------------------------------------------------- +# Identifiers +# -------------------------------------------------------------------------- + +def test_cve_alias_becomes_the_primary_identifier(): + primary, aliases = osv.vulnerability_ids(_requests_advisory()) + assert primary == "CVE-2018-18074" + assert aliases == ("GHSA-x84v-xcm2-53pg",) + + +def test_advisory_without_cve_keeps_its_osv_id(): + record = {"id": "GHSA-abcd-efgh-ijkl", "aliases": ["PYSEC-2024-1"]} + primary, aliases = osv.vulnerability_ids(record) + assert primary == "GHSA-abcd-efgh-ijkl" + assert aliases == ("PYSEC-2024-1",) + + +# -------------------------------------------------------------------------- +# Lookup — hit, miss, dedupe, pins +# -------------------------------------------------------------------------- + +def test_hit_produces_an_affected_cve_statement(): + fake = FakeOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + + assert result.error is None + [stmt] = result.statements + assert stmt.finding_class.id == "CVE-2018-18074" + assert stmt.status == "affected" + assert stmt.product_ref == "dependency-0-requests" + assert stmt.justification is None + assert "2.20.0" in stmt.action_statement + assert stmt.finding_class.aliases == ("GHSA-x84v-xcm2-53pg",) + assert stmt.finding_class.iri == "https://osv.dev/vulnerability/GHSA-x84v-xcm2-53pg" + assert stmt.finding_class.source_name == "OSV" + + +def test_miss_produces_no_statement_and_no_negative_claim(): + fake = FakeOSV(matches={}) + result = _lookup([_dep("requests", "2.32.3")], fake) + assert result.error is None + assert result.statements == [] + assert result.queried == 1 + assert fake.gets == [] + + +def test_twin_records_for_one_cve_yield_one_statement(): + fake = FakeOSV( + [_requests_advisory(), _pysec_twin()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg", "PYSEC-2018-28"]}, + ) + result = _lookup([_dep("requests", "2.19.0")], fake) + [stmt] = result.statements + assert stmt.finding_class.id == "CVE-2018-18074" + assert set(stmt.finding_class.aliases) == {"GHSA-x84v-xcm2-53pg", "PYSEC-2018-28"} + + +def test_unpinned_dependencies_are_skipped_and_counted(): + fake = FakeOSV() + result = _lookup([_dep("torch", "2.0", pinned=False), + _dep("numpy", "unknown", pinned=False)], fake) + assert result.statements == [] + assert result.skipped_unpinned == 2 + assert result.queried == 0 + assert fake.calls == 0 + + +def test_dependency_refs_follow_the_dependency_index(): + fake = FakeOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + deps = [_dep("torch", "2.1.0"), _dep("requests", "2.19.0")] + result = _lookup(deps, fake) + assert [s.product_ref for s in result.statements] == ["dependency-1-requests"] + + +def test_osv_match_the_local_check_rejects_is_under_investigation(): + """OSV and the range check disagree: never a silent drop, never `affected`.""" + record = _requests_advisory() + record["affected"][0]["ranges"][0]["events"] = [ + {"introduced": "0"}, {"fixed": "2.0.0"} + ] + fake = FakeOSV([record], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert stmt.status == "under_investigation" + assert stmt.action_statement is None + + +def test_osv_match_the_local_check_cannot_evaluate_is_under_investigation(): + record = _requests_advisory() + record["affected"][0]["ranges"][0]["type"] = "GIT" + fake = FakeOSV([record], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert stmt.status == "under_investigation" + + +def test_affected_wins_when_twin_records_disagree(): + twin = _pysec_twin() + twin["affected"][0]["ranges"][0]["type"] = "GIT" + twin["affected"][0].pop("versions") + fake = FakeOSV( + [_requests_advisory(), twin], + matches={("requests", "2.19.0"): ["PYSEC-2018-28", "GHSA-x84v-xcm2-53pg"]}, + ) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert stmt.status == "affected" + + +def test_withdrawn_advisories_are_skipped(): + record = _requests_advisory() + record["withdrawn"] = "2024-01-01T00:00:00Z" + fake = FakeOSV([record], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + assert _lookup([_dep("requests", "2.19.0")], fake).statements == [] + + +def test_action_names_only_the_fix_that_closes_the_pinned_interval(): + """An older interval's fix is not an upgrade target (disjoint ranges).""" + record = _requests_advisory() + record["affected"][0]["ranges"][0]["events"] = [ + {"introduced": "1.0"}, {"fixed": "1.5"}, + {"introduced": "2.0"}, {"fixed": "2.2"}, + ] + fake = FakeOSV([record], + matches={("requests", "2.1"): ["GHSA-x84v-xcm2-53pg"]}) + [stmt] = _lookup([_dep("requests", "2.1")], fake).statements + assert stmt.status == "affected" + assert "fixed in: 2.2)" in stmt.action_statement + assert "1.5" not in stmt.action_statement + + +def test_action_offers_no_fix_when_the_pinned_interval_is_unfixed(): + record = _requests_advisory() + record["affected"][0]["ranges"][0]["events"] = [ + {"introduced": "1.0"}, {"fixed": "1.5"}, {"introduced": "2.0"}, + ] + fake = FakeOSV([record], + matches={("requests", "2.1"): ["GHSA-x84v-xcm2-53pg"]}) + [stmt] = _lookup([_dep("requests", "2.1")], fake).statements + assert "No fixed version" in stmt.action_statement + + +def test_a_fix_spelled_two_ways_is_listed_once(): + twin = _pysec_twin() + twin["affected"][0]["ranges"][0]["events"][1] = {"fixed": "2.20"} + fake = FakeOSV( + [_requests_advisory(), twin], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg", "PYSEC-2018-28"]}, + ) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert "(fixed in: 2.20.0)" in stmt.action_statement + + +def test_a_fix_after_last_affected_is_not_the_pinned_intervals_fix(): + record = _range_record([ + {"introduced": "0"}, {"last_affected": "1.4"}, + {"introduced": "2.0"}, {"fixed": "2.5"}, + ]) + events = record["affected"][0]["ranges"][0]["events"] + assert osv._closing_fix(events, osv.Version("1.2")) is None + assert osv._closing_fix(events, osv.Version("2.1")) == "2.5" + assert osv._closing_fix(events, osv.Version("1.8")) is None + + +def test_twin_records_contribute_one_fix_per_interval(): + twin = _pysec_twin() + fake = FakeOSV( + [_requests_advisory(), twin], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg", "PYSEC-2018-28"]}, + ) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert "(fixed in: 2.20.0)" in stmt.action_statement + + +def test_no_fixed_version_gets_an_honest_action(): + record = _requests_advisory() + record["affected"][0]["ranges"][0]["events"] = [{"introduced": "0"}] + fake = FakeOSV([record], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert "No fixed version" in stmt.action_statement + + +def test_query_uses_canonical_name_and_pypi_ecosystem(): + fake = FakeOSV() + _lookup([_dep("Scikit_Learn", "1.0.0")], fake) + [query] = fake.posts[0]["queries"] + assert query == {"package": {"name": "scikit-learn", "ecosystem": "PyPI"}, + "version": "1.0.0"} + + +def test_paginated_querybatch_results_are_followed(): + fake = FakeOSV( + [_requests_advisory(), _pysec_twin()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}, + pages={("requests", "2.19.0"): True, "tok-requests": ["PYSEC-2018-28"]}, + ) + [stmt] = _lookup([_dep("requests", "2.19.0")], fake).statements + assert len(fake.posts) == 2 + assert fake.posts[1]["queries"][0]["page_token"] == "tok-requests" + assert "PYSEC-2018-28" in stmt.finding_class.aliases + + +# -------------------------------------------------------------------------- +# Degradation — never a changed exit code, never a lost model finding +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("failure", [ + requests.ConnectionError("no route to host"), + requests.Timeout("read timed out"), +]) +def test_network_failure_degrades_to_no_statements(failure): + fake = FakeOSV(fail=failure) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +@pytest.mark.parametrize("payload", [ + ValueError("not json"), + ["not", "an", "object"], + {"results": "nope"}, + {"results": []}, # wrong length + {"results": [{"vulns": [{"no": "id"}]}]}, +]) +def test_malformed_querybatch_degrades_to_no_statements(payload): + fake = FakeOSV(batch_payload=payload) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +def test_a_record_that_is_not_the_one_asked_for_degrades(): + """A proxy or cache returning the wrong body must not become a statement.""" + wrong = _requests_advisory() + wrong["id"] = "GHSA-something-else" + fake = FakeOSV(matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + fake.records["GHSA-x84v-xcm2-53pg"] = wrong + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +def test_a_non_object_querybatch_result_degrades(): + fake = FakeOSV(batch_payload={"results": ["not-an-object"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +def test_pagination_that_never_ends_degrades(): + class Endless: + posts = 0 + + def post(self, url, json=None, timeout=None, headers=None): + Endless.posts += 1 + return FakeResponse({"results": [ + {"vulns": [{"id": "GHSA-x"}], "next_page_token": "again"} + ]}) + + result = _lookup([_dep("requests", "2.19.0")], Endless()) + assert result.statements == [] + assert "pagination" in result.error + assert Endless.posts == osv._MAX_PAGES + + +def test_an_unparseable_listed_version_is_skipped_not_fatal(): + record = {"id": "PYSEC-demo", "affected": [ + {"package": {"ecosystem": "PyPI", "name": "demo-pkg"}, + "versions": ["not a version", "1.0.1"]} + ]} + assert osv.version_affected(record, "demo-pkg", "1.0.1") is True + + +def test_a_cache_file_from_another_schema_is_ignored(tmp_path): + (tmp_path / osv.CACHE_FILENAME).write_text(json.dumps({"schema": 999})) + fake = _hit_fake() + _lookup([_dep("requests", "2.19.0")], fake, cache_dir=tmp_path) + assert fake.calls == 2 + + +def _many_advisories(count): + records = [] + for n in range(count): + record = _requests_advisory() + record["id"] = f"GHSA-many-{n:04d}" + record["aliases"] = [f"CVE-2099-{n:05d}"] + records.append(record) + return records + + +def test_a_pin_with_many_advisories_gets_every_statement(): + """Real pins name 100+ advisories; all must land, in a stable order.""" + records = _many_advisories(140) + fake = FakeOSV(records, matches={("requests", "2.19.0"): [r["id"] for r in records]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.error is None + assert len(fake.gets) == 140 + assert [s.finding_class.id for s in result.statements] == \ + sorted(f"CVE-2099-{n:05d}" for n in range(140)) + + +def test_one_failed_record_among_many_degrades_the_whole_lookup(): + records = _many_advisories(40) + ids = [r["id"] for r in records] + fake = FakeOSV(records[:-1], matches={("requests", "2.19.0"): ids}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +def test_the_default_session_pools_connections_for_every_worker(): + # Bound at import, before conftest swaps the module attribute out. + session = _REAL_DEFAULT_SESSION() + assert isinstance(session, requests.Session) + adapter = session.get_adapter("https://api.osv.dev/v1/querybatch") + assert adapter._pool_maxsize == osv._FETCH_WORKERS + + +class FlakyOSV(FakeOSV): + """Answers the first `flakes` record requests with `status`, then normally.""" + + def __init__(self, *args, flakes=1, status=503, **kwargs): + super().__init__(*args, **kwargs) + self.flakes = flakes + self.status = status + + def get(self, url, timeout=None, headers=None): + if self.flakes: + self.flakes -= 1 + self.gets.append(url) + return FakeResponse("503 Server Error", self.status) + return super().get(url, timeout=timeout, headers=headers) + + +def test_a_transient_503_is_retried_and_the_lookup_succeeds(): + """Observed live: one 503 in 140 record fetches must not cost every statement.""" + fake = FlakyOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.error is None + assert len(result.statements) == 1 + assert len(fake.gets) == 2 + + +@pytest.mark.parametrize("status", [429, 500, 502, 504]) +def test_other_transient_statuses_are_retried(status): + fake = FlakyOSV([_requests_advisory()], status=status, + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + assert len(_lookup([_dep("requests", "2.19.0")], fake).statements) == 1 + + +def test_a_persistent_503_degrades_after_bounded_attempts(): + fake = FlakyOSV([_requests_advisory()], flakes=99, + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + assert len(fake.gets) == osv._ATTEMPTS + + +def test_a_404_is_not_retried(): + fake = FakeOSV([], matches={("requests", "2.19.0"): ["GHSA-gone"]}) + _lookup([_dep("requests", "2.19.0")], fake) + assert len(fake.gets) == 1 + + +def test_a_connection_error_is_retried(): + class DropsFirst(FakeOSV): + dropped = False + + def post(self, url, json=None, timeout=None, headers=None): + if not self.dropped: + self.dropped = True + raise requests.ConnectionError("reset by peer") + return super().post(url, json=json, timeout=timeout, headers=headers) + + fake = DropsFirst([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + assert len(_lookup([_dep("requests", "2.19.0")], fake).statements) == 1 + + +def test_a_retry_pause_that_would_overrun_the_budget_degrades(monkeypatch): + monkeypatch.setattr(osv, "_BACKOFF_SECONDS", 3600) + fake = FlakyOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert "time budget" in result.error + + +def test_a_missing_vulnerability_record_degrades_the_whole_lookup(): + fake = FakeOSV([], matches={("requests", "2.19.0"): ["GHSA-gone"]}) + result = _lookup([_dep("requests", "2.19.0")], fake) + assert result.statements == [] + assert result.error + + +def test_an_exhausted_time_budget_degrades(monkeypatch): + fake = FakeOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + result = _lookup([_dep("requests", "2.19.0")], fake, budget_seconds=0) + assert result.statements == [] + assert result.error + assert fake.calls == 0 + + +def test_unexpected_exception_is_contained(): + class Exploding: + def post(self, *a, **kw): + raise RuntimeError("boom") + result = _lookup([_dep("requests", "2.19.0")], Exploding()) + assert result.statements == [] + assert result.error + + +# -------------------------------------------------------------------------- +# Cache +# -------------------------------------------------------------------------- + +def _hit_fake(): + return FakeOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + + +def test_repeat_lookup_makes_no_network_call(tmp_path): + first = _hit_fake() + r1 = _lookup([_dep("requests", "2.19.0")], first, cache_dir=tmp_path) + assert first.calls == 2 + + second = _hit_fake() + r2 = _lookup([_dep("requests", "2.19.0")], second, cache_dir=tmp_path) + assert second.calls == 0 + assert [s.finding_class.id for s in r2.statements] == \ + [s.finding_class.id for s in r1.statements] + + +def test_a_cached_miss_is_also_reused(tmp_path): + _lookup([_dep("requests", "2.32.3")], FakeOSV(), cache_dir=tmp_path) + second = FakeOSV() + _lookup([_dep("requests", "2.32.3")], second, cache_dir=tmp_path) + assert second.calls == 0 + + +def test_only_the_uncached_pin_is_queried(tmp_path): + _lookup([_dep("requests", "2.19.0")], _hit_fake(), cache_dir=tmp_path) + second = _hit_fake() + _lookup([_dep("requests", "2.19.0"), _dep("flask", "3.0.0")], second, + cache_dir=tmp_path) + assert [q["package"]["name"] for q in second.posts[0]["queries"]] == ["flask"] + assert second.gets == [] + + +def test_stale_cache_entries_are_refetched(tmp_path): + _lookup([_dep("requests", "2.19.0")], _hit_fake(), cache_dir=tmp_path, + now=time.time() - osv.CACHE_TTL_SECONDS - 60) + second = _hit_fake() + _lookup([_dep("requests", "2.19.0")], second, cache_dir=tmp_path) + assert second.calls == 2 + + +def test_a_corrupt_cache_file_is_ignored(tmp_path): + (tmp_path / osv.CACHE_FILENAME).write_text("{not json") + fake = _hit_fake() + result = _lookup([_dep("requests", "2.19.0")], fake, cache_dir=tmp_path) + assert len(result.statements) == 1 + assert fake.calls == 2 + + +def test_a_failed_lookup_does_not_poison_the_cache(tmp_path): + _lookup([_dep("requests", "2.19.0")], + FakeOSV(fail=requests.ConnectionError("down")), cache_dir=tmp_path) + fake = _hit_fake() + result = _lookup([_dep("requests", "2.19.0")], fake, cache_dir=tmp_path) + assert len(result.statements) == 1 + assert fake.calls == 2 + + +def test_unwritable_cache_dir_still_returns_results(tmp_path): + missing = tmp_path / "does-not-exist" / "nested" + fake = _hit_fake() + result = _lookup([_dep("requests", "2.19.0")], fake, cache_dir=missing) + assert len(result.statements) == 1 + + +# -------------------------------------------------------------------------- +# Emitted documents +# -------------------------------------------------------------------------- + +def test_cve_statements_validate_in_both_flavors(): + fake = FakeOSV([_requests_advisory()], + matches={("requests", "2.19.0"): ["GHSA-x84v-xcm2-53pg"]}) + statements = _lookup([_dep("requests", "2.19.0")], fake).statements + + with _OPENVEX_SCHEMA.open() as fh: + Draft202012Validator(json.load(fh)).validate( + json.loads(generate_openvex(statements, sbom_serial=SERIAL)) + ) + ovex = json.loads(generate_openvex(statements, sbom_serial=SERIAL)) + vuln = ovex["statements"][0]["vulnerability"] + assert vuln["name"] == "CVE-2018-18074" + assert vuln["@id"] == "https://osv.dev/vulnerability/GHSA-x84v-xcm2-53pg" + assert vuln["aliases"] == ["GHSA-x84v-xcm2-53pg"] + + cdx_json = generate_cyclonedx_vex(statements, sbom_serial=SERIAL) + assert JsonStrictValidator(SchemaVersion.V1_7).validate_str(cdx_json) is None + entry = json.loads(cdx_json)["vulnerabilities"][0] + assert entry["id"] == "CVE-2018-18074" + assert entry["source"] == { + "name": "OSV", + "url": "https://osv.dev/vulnerability/GHSA-x84v-xcm2-53pg", + } + assert entry["references"] == [ + {"id": "GHSA-x84v-xcm2-53pg", + "source": {"name": "GitHub Advisory Database"}} + ] + + +# -------------------------------------------------------------------------- +# Pin detection in the scanner +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("line, pinned", [ + ("requests==2.19.0", True), + ("requests===2.19.0", True), + ("requests == 2.19.0 ; python_version >= '3.8'", True), + ("requests>=2.19.0", False), + ("requests~=2.19.0", False), + ("requests==2.*", False), + ("requests>=2.0,<3", False), + ("requests", False), +]) +def test_scanner_marks_only_exact_pins(tmp_path, line, pinned): + from aisbom.scanner import DeepScanner + + req = tmp_path / "requirements.txt" + req.write_text(line + "\n") + scanner = DeepScanner(str(tmp_path)) + scanner._parse_requirements(req) + [dep] = scanner.dependencies + assert dep["pinned"] is pinned + + +def test_disabled_env_var_is_read(): + assert osv.disabled_by_env({"AISBOM_NO_OSV": "1"}) is True + assert osv.disabled_by_env({}) is False