#!/usr/bin/env python3 """Run pip-audit with lock-bound provenance and an expiring allowlist. Security M8 offline behaviour is deliberately fail-closed: - a trustworthy online report uses pip-audit's documented exit code 0 or 1; - an offline snapshot must be fresh, not future-dated, and bound to the exact current requirements.lock dependency set and SHA-256; - malformed, incomplete, skipped, or mismatched reports are rejected. Use ``--snapshot-output PATH`` from the scheduled refresh workflow to capture a validated, provenance-bearing online report without applying the allowlist. """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import pathlib import re import subprocess import sys import tempfile from dataclasses import dataclass from typing import Any import yaml # pinned in supply-chain/audit-tooling.lock from packaging.utils import canonicalize_name from packaging.version import InvalidVersion, Version REPO = pathlib.Path(__file__).resolve().parents[1] ALLOWLIST = REPO / "supply-chain" / "pip-audit-allowlist.yml" LOCKFILE = REPO / "requirements.lock" OUTPUT = pathlib.Path(os.environ.get("FLINTTRADE_PIP_AUDIT_REPORT", REPO / "pip-audit-report.json")) SNAPSHOT_DIR = REPO / "supply-chain" MAX_SNAPSHOT_AGE_DAYS = 14 REPORT_SCHEMA = "flinttrade-pip-audit-v1" TRUSTWORTHY_PIP_AUDIT_EXIT_CODES = {0, 1} KNOWN_SEVERITIES = {"LOW", "MEDIUM", "MODERATE", "HIGH", "CRITICAL", "UNKNOWN"} BLOCKING_SEVERITIES = {"HIGH", "CRITICAL", "UNKNOWN"} _LOCK_REQUIREMENT_RE = re.compile( r"^([A-Za-z0-9][A-Za-z0-9._-]*)==([^\s;\\]+)", ) class ReportValidationError(ValueError): """Raised when an audit report cannot prove coverage of the current lock.""" @dataclass(frozen=True) class LockState: """Canonical dependency coverage and digest for requirements.lock.""" path: pathlib.Path dependencies: dict[str, Version] sha256: str def _utc_now() -> dt.datetime: return dt.datetime.now(dt.UTC) def _format_timestamp(value: dt.datetime) -> str: if value.tzinfo is None: raise ValueError("provenance timestamp must be timezone-aware") return value.astimezone(dt.UTC).isoformat().replace("+00:00", "Z") def _parse_timestamp(value: object, field: str) -> dt.datetime: if not isinstance(value, str) or not value.strip(): raise ReportValidationError(f"snapshot provenance {field} must be a timestamp") raw = value.strip() if raw.endswith("Z"): raw = f"{raw[:-1]}+00:00" try: parsed = dt.datetime.fromisoformat(raw) except ValueError as exc: raise ReportValidationError(f"snapshot provenance {field} is invalid") from exc if parsed.tzinfo is None: raise ReportValidationError(f"snapshot provenance {field} must include a timezone") return parsed.astimezone(dt.UTC) def _load_lock_state(path: pathlib.Path = LOCKFILE) -> LockState: """Parse exact PEP 503 names and PEP 440 versions from a hashed uv export.""" try: raw = path.read_bytes() text = raw.decode("utf-8") except (OSError, UnicodeDecodeError) as exc: raise ReportValidationError(f"cannot read {path}: {exc}") from exc dependencies: dict[str, Version] = {} for line_number, line in enumerate(text.splitlines(), start=1): if not line or line.startswith("#") or line[0].isspace(): continue match = _LOCK_REQUIREMENT_RE.match(line) if match is None: raise ReportValidationError(f"unparsed requirement at {path}:{line_number}") name, raw_version = match.groups() remainder = line[match.end() :].strip() if remainder and not remainder.startswith((";", "\\")): raise ReportValidationError(f"unexpected requirement syntax at {path}:{line_number}") normalised_name = str(canonicalize_name(name)) try: version = Version(raw_version) except InvalidVersion as exc: raise ReportValidationError(f"invalid version {raw_version!r} for {normalised_name} in {path}") from exc previous = dependencies.get(normalised_name) if previous is not None and previous != version: raise ReportValidationError(f"conflicting locked versions for {normalised_name}: {previous} and {version}") dependencies[normalised_name] = version if not dependencies: raise ReportValidationError(f"{path} contains no pinned dependencies") return LockState(path=path, dependencies=dependencies, sha256=hashlib.sha256(raw).hexdigest()) def _lock_is_unchanged(lock_state: LockState) -> bool: try: current = _load_lock_state(lock_state.path) except ReportValidationError as exc: print(f"cannot re-read requirements.lock: {exc}", file=sys.stderr) return False if current.sha256 != lock_state.sha256: print("requirements.lock changed while pip-audit was running (failing closed)", file=sys.stderr) return False return True def _normalised_report_coverage(report: object) -> dict[str, Version]: if not isinstance(report, dict): raise ReportValidationError("pip-audit report must be a JSON object") dependencies = report.get("dependencies") if not isinstance(dependencies, list) or not dependencies: raise ReportValidationError("pip-audit report must contain a non-empty dependencies list") if not isinstance(report.get("fixes"), list): raise ReportValidationError("pip-audit report fixes must be a list") coverage: dict[str, Version] = {} for index, dependency in enumerate(dependencies): if not isinstance(dependency, dict): raise ReportValidationError(f"dependency {index} must be an object") name = dependency.get("name") raw_version = dependency.get("version") if not isinstance(name, str) or not name.strip(): raise ReportValidationError(f"dependency {index} has no valid name") if not isinstance(raw_version, str) or not raw_version.strip(): raise ReportValidationError(f"dependency {name!r} has no valid version") if dependency.get("skip_reason"): raise ReportValidationError(f"dependency {name!r} was skipped by pip-audit") vulnerabilities = dependency.get("vulns") if not isinstance(vulnerabilities, list): raise ReportValidationError(f"dependency {name!r} has no valid vulns list") for vulnerability in vulnerabilities: if not isinstance(vulnerability, dict): raise ReportValidationError(f"dependency {name!r} has a malformed vulnerability") vulnerability_id = vulnerability.get("id") if not isinstance(vulnerability_id, str) or not vulnerability_id.strip(): raise ReportValidationError(f"dependency {name!r} has a vulnerability without an id") normalised_name = str(canonicalize_name(name)) try: version = Version(raw_version) except InvalidVersion as exc: raise ReportValidationError(f"dependency {normalised_name!r} has invalid version {raw_version!r}") from exc if normalised_name in coverage: raise ReportValidationError(f"duplicate dependency {normalised_name!r} in pip-audit report") coverage[normalised_name] = version return coverage def _validate_report_coverage(report: object, lock_state: LockState) -> dict[str, Any]: coverage = _normalised_report_coverage(report) expected_names = set(lock_state.dependencies) actual_names = set(coverage) missing = sorted(expected_names - actual_names) extra = sorted(actual_names - expected_names) mismatched = sorted( name for name in expected_names & actual_names if lock_state.dependencies[name] != coverage[name] ) if missing or extra or mismatched: details: list[str] = [] if missing: details.append(f"missing={missing}") if extra: details.append(f"extra={extra}") if mismatched: versions = [f"{name}:{coverage[name]}!={lock_state.dependencies[name]}" for name in mismatched] details.append(f"version_mismatch={versions}") raise ReportValidationError( "pip-audit dependency coverage does not match requirements.lock: " + "; ".join(details) ) assert isinstance(report, dict) return report def _report_has_vulnerabilities(report: dict[str, Any]) -> bool: return any(dependency["vulns"] for dependency in report["dependencies"]) def _with_provenance( report: dict[str, Any], *, source: str, lock_state: LockState, generated_at: dt.datetime, audited_at: dt.datetime, pip_audit_exit_code: int, snapshot: str | None = None, ) -> dict[str, Any]: """Add top-level provenance without changing pip-audit dependency data.""" result = dict(report) provenance: dict[str, Any] = { "schema": REPORT_SCHEMA, "source": source, "generated_at": _format_timestamp(generated_at), "audited_at": _format_timestamp(audited_at), "requirements_lock_sha256": lock_state.sha256, "pip_audit_exit_code": pip_audit_exit_code, } if snapshot is not None: provenance["snapshot"] = snapshot result["_meta"] = provenance return result def _latest_snapshot() -> pathlib.Path | None: snapshots = sorted(SNAPSHOT_DIR.glob("vuln-snapshot-*.json")) return snapshots[-1] if snapshots else None def _run_online_audit( lock_state: LockState, *, source: str = "online-osv", now: dt.datetime | None = None, ) -> dict[str, Any] | None: """Return a validated online report for documented exit code 0 or 1 only.""" try: proc = subprocess.run( [ "pip-audit", "--require-hashes", "-r", str(lock_state.path), "--format", "json", "--vulnerability-service", "osv", "--disable-pip", "--strict", "--progress-spinner", "off", "--timeout", "10", ], capture_output=True, text=True, check=False, timeout=120, ) except FileNotFoundError: print( "pip-audit not installed; run `pip install --require-hashes -r " "supply-chain/audit-tooling.lock` first (failing closed)", file=sys.stderr, ) return None except (OSError, subprocess.TimeoutExpired) as exc: print(f"pip-audit could not complete: {exc} (failing closed)", file=sys.stderr) return None if proc.returncode not in TRUSTWORTHY_PIP_AUDIT_EXIT_CODES: print( f"pip-audit failed operationally with status {proc.returncode}: {proc.stderr.strip()}", file=sys.stderr, ) return None if not proc.stdout.strip(): print("pip-audit emitted empty output (failing closed)", file=sys.stderr) return None if not _lock_is_unchanged(lock_state): return None try: report = json.loads(proc.stdout) report = _validate_report_coverage(report, lock_state) except (json.JSONDecodeError, ReportValidationError) as exc: print(f"pip-audit emitted an untrustworthy report: {exc}", file=sys.stderr) return None has_vulnerabilities = _report_has_vulnerabilities(report) if (proc.returncode == 0 and has_vulnerabilities) or (proc.returncode == 1 and not has_vulnerabilities): print( "pip-audit exit status contradicts its vulnerability report (failing closed)", file=sys.stderr, ) return None audited_at = now or _utc_now() return _with_provenance( report, source=source, lock_state=lock_state, generated_at=audited_at, audited_at=audited_at, pip_audit_exit_code=proc.returncode, ) def _validate_snapshot_provenance( report: dict[str, Any], snapshot: pathlib.Path, lock_state: LockState, now: dt.datetime, ) -> tuple[dt.datetime, int]: metadata = report.get("_meta") if not isinstance(metadata, dict): raise ReportValidationError("snapshot has no provenance metadata") if metadata.get("schema") != REPORT_SCHEMA: raise ReportValidationError("snapshot provenance schema is unsupported") if metadata.get("source") != "online-osv-snapshot": raise ReportValidationError("snapshot provenance source is not an online OSV capture") if metadata.get("requirements_lock_sha256") != lock_state.sha256: raise ReportValidationError("snapshot requirements.lock digest does not match the current lock") return_code = metadata.get("pip_audit_exit_code") if isinstance(return_code, bool) or return_code not in TRUSTWORTHY_PIP_AUDIT_EXIT_CODES: raise ReportValidationError("snapshot has an untrustworthy pip-audit exit code") generated_at = _parse_timestamp(metadata.get("generated_at"), "generated_at") audited_at = _parse_timestamp(metadata.get("audited_at"), "audited_at") if generated_at < audited_at: raise ReportValidationError("snapshot generation time predates its audit time") if generated_at > now or audited_at > now: raise ReportValidationError("snapshot provenance is future-dated") if now - audited_at > dt.timedelta(days=MAX_SNAPSHOT_AGE_DAYS): raise ReportValidationError(f"snapshot is older than {MAX_SNAPSHOT_AGE_DAYS} days") prefix = "vuln-snapshot-" date_text = snapshot.stem[len(prefix) :] if snapshot.stem.startswith(prefix) else "" try: filename_date = dt.date.fromisoformat(date_text) except ValueError as exc: raise ReportValidationError("snapshot filename has an invalid date") from exc if filename_date > now.date(): raise ReportValidationError("snapshot filename is future-dated") if filename_date != audited_at.date(): raise ReportValidationError("snapshot filename date does not match its audit provenance") has_vulnerabilities = _report_has_vulnerabilities(report) if (return_code == 0 and has_vulnerabilities) or (return_code == 1 and not has_vulnerabilities): raise ReportValidationError("snapshot exit status contradicts its vulnerability report") return audited_at, return_code def _load_offline_snapshot( lock_state: LockState, *, now: dt.datetime | None = None, ) -> dict[str, Any] | None: """Load a fresh snapshot bound to the exact current requirements.lock.""" snapshot = _latest_snapshot() if snapshot is None: print("OSV unavailable and no local snapshot exists (failing closed)", file=sys.stderr) return None current_time = now or _utc_now() try: report = json.loads(snapshot.read_text(encoding="utf-8")) report = _validate_report_coverage(report, lock_state) audited_at, return_code = _validate_snapshot_provenance( report, snapshot, lock_state, current_time, ) except (OSError, json.JSONDecodeError, ReportValidationError) as exc: print(f"cached snapshot {snapshot.name} is untrustworthy: {exc}", file=sys.stderr) return None print(f"OSV unavailable; evaluating validated cached snapshot {snapshot.name}") return _with_provenance( report, source="cached-osv-snapshot", lock_state=lock_state, generated_at=current_time, audited_at=audited_at, pip_audit_exit_code=return_code, snapshot=snapshot.name, ) def _active_allowlist() -> set[tuple[str, str]]: if not ALLOWLIST.exists(): return set() allow = yaml.safe_load(ALLOWLIST.read_text(encoding="utf-8")) or {} today = dt.date.today() active: set[tuple[str, str]] = set() for entry in allow.get("allowlist") or []: missing = [key for key in ("vuln_id", "package", "reason", "accepted_by", "expires") if key not in entry] if missing: print(f"WARNING: allowlist entry missing required keys {missing}: {entry}", file=sys.stderr) continue if dt.date.fromisoformat(str(entry["expires"])) >= today: active.add((entry["vuln_id"], str(canonicalize_name(entry["package"])))) else: print( f"NOTE: allowlist entry for {entry['vuln_id']} expired on {entry['expires']}; " "it no longer suppresses findings" ) return active def _normalise_severity(value: object) -> str: if not isinstance(value, str): return "UNKNOWN" normalised = value.strip().upper() return normalised if normalised in KNOWN_SEVERITIES else "UNKNOWN" def _blocking_findings( report: dict[str, Any], active_allow: set[tuple[str, str]], ) -> list[dict[str, str]]: """Return non-allowlisted HIGH, CRITICAL, or unknown-severity findings.""" blocking: list[dict[str, str]] = [] for dependency in report.get("dependencies", []): for vulnerability in dependency.get("vulns", []): severity = _normalise_severity(vulnerability.get("severity")) if severity not in BLOCKING_SEVERITIES: continue key = (vulnerability["id"], str(canonicalize_name(dependency["name"]))) if key in active_allow: print(f"ALLOWLISTED: {vulnerability['id']} in {dependency['name']} {dependency['version']}") continue blocking.append( { "id": vulnerability["id"], "package": dependency["name"], "version": dependency["version"], "severity": severity, } ) return blocking def _remove_stale_output(path: pathlib.Path) -> bool: try: path.unlink(missing_ok=True) except OSError as exc: print(f"cannot remove stale audit output {path}: {exc}", file=sys.stderr) return False return True def _write_json(path: pathlib.Path, report: dict[str, Any]) -> None: """Atomically publish a validated report without leaving a partial artefact.""" temporary: pathlib.Path | None = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False, ) as handle: temporary = pathlib.Path(handle.name) json.dump(report, handle, indent=2) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) temporary = None finally: if temporary is not None: temporary.unlink(missing_ok=True) def _load_current_lock() -> LockState | None: try: return _load_lock_state() except ReportValidationError as exc: print(f"cannot establish requirements.lock coverage: {exc}", file=sys.stderr) return None def _capture_snapshot(path: pathlib.Path) -> int: if not _remove_stale_output(path): return 2 lock_state = _load_current_lock() if lock_state is None: return 2 report = _run_online_audit(lock_state, source="online-osv-snapshot") if report is None: return 2 if not _lock_is_unchanged(lock_state): return 2 try: _write_json(path, report) except OSError as exc: print(f"cannot publish validated snapshot {path}: {exc}", file=sys.stderr) return 2 print(f"wrote validated OSV snapshot {path}") return 0 def _run_gate() -> int: if not _remove_stale_output(OUTPUT): return 2 lock_state = _load_current_lock() if lock_state is None: return 2 report = _run_online_audit(lock_state) if report is None: report = _load_offline_snapshot(lock_state) if report is None: return 2 if not _lock_is_unchanged(lock_state): return 2 try: _write_json(OUTPUT, report) except OSError as exc: print(f"cannot publish validated audit report {OUTPUT}: {exc}", file=sys.stderr) return 2 blocking = _blocking_findings(report, _active_allowlist()) if blocking: print("BLOCKING:", json.dumps(blocking, indent=2)) return 1 print("pip-audit clean (after allowlist filter)") return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--snapshot-output", type=pathlib.Path, help="capture a validated online OSV report for the offline snapshot workflow", ) args = parser.parse_args(argv) if args.snapshot_output is not None: return _capture_snapshot(args.snapshot_output) return _run_gate() if __name__ == "__main__": raise SystemExit(main())