#!/usr/bin/env python3 """Fail on RustSec findings reachable from nigig-site production dependencies. Cargo.lock is workspace-wide, so cargo-audit can report packages that are not in this application's normal/build graph. This script consumes unmodified `cargo metadata` and `cargo audit --json` output, traverses every non-dev edge for all target predicates, and makes that containment explicit rather than ignoring advisory IDs globally. """ from __future__ import annotations import json import sys from collections import deque from pathlib import Path from typing import Any, NoReturn def fail(message: str) -> NoReturn: print(f"ERROR: {message}", file=sys.stderr) raise SystemExit(1) def load_json(path: str) -> dict[str, Any]: try: value = json.loads(Path(path).read_text()) except (OSError, UnicodeError, json.JSONDecodeError) as error: fail(f"cannot parse {path}: {error}") if not isinstance(value, dict): fail(f"expected a JSON object in {path}") return value def main() -> None: if len(sys.argv) != 4: fail("usage: audit-production-deps.py METADATA.json AUDIT.json REPORT.txt") metadata = load_json(sys.argv[1]) audit = load_json(sys.argv[2]) packages = metadata.get("packages") resolve = metadata.get("resolve") if not isinstance(packages, list) or not isinstance(resolve, dict): fail("cargo metadata is missing packages/resolve") nodes = resolve.get("nodes") if not isinstance(nodes, list): fail("cargo metadata is missing resolve.nodes") roots = [ package for package in packages if package.get("name") == "nigig-site" and str(package.get("manifest_path", "")).replace("\\", "/").endswith( "/crates/apps/nigig-site/Cargo.toml" ) ] if len(roots) != 1: fail(f"expected exactly one nigig-site package, found {len(roots)}") root_id = roots[0].get("id") node_by_id = {node.get("id"): node for node in nodes} package_by_id = {package.get("id"): package for package in packages} if root_id not in node_by_id or root_id not in package_by_id: fail("nigig-site is absent from the resolved graph") reachable: set[str] = set() queue: deque[str] = deque([root_id]) while queue: package_id = queue.popleft() if package_id in reachable: continue reachable.add(package_id) node = node_by_id.get(package_id) if not isinstance(node, dict): fail(f"resolved node is missing for {package_id}") dependencies = node.get("deps") if not isinstance(dependencies, list): fail(f"resolved dependencies are malformed for {package_id}") for dependency in dependencies: kinds = dependency.get("dep_kinds") if not isinstance(kinds, list) or not kinds: fail(f"dependency kinds are missing for {dependency!r}") # `kind: null` is a normal dependency. Build dependencies execute in # the production build trust boundary. Dev-only edges are excluded. if not any(kind.get("kind") in (None, "build") for kind in kinds): continue child = dependency.get("pkg") if child not in node_by_id: fail(f"resolved child node is missing for {child}") queue.append(child) reachable_versions = { (str(package_by_id[item].get("name")), str(package_by_id[item].get("version"))) for item in reachable } vulnerabilities = audit.get("vulnerabilities") warnings = audit.get("warnings") database = audit.get("database") settings = audit.get("settings") if not isinstance(vulnerabilities, dict) or not isinstance(warnings, dict): fail("cargo-audit JSON is missing vulnerabilities/warnings") if not isinstance(database, dict) or not database.get("last-commit"): fail("cargo-audit JSON is missing database provenance") if not isinstance(settings, dict): fail("cargo-audit JSON is missing settings") if settings.get("ignore") != []: fail("cargo-audit advisory ignores are forbidden") if settings.get("target_arch") != [] or settings.get("target_os") != []: fail("cargo-audit target filters are forbidden") informational = settings.get("informational_warnings") if not isinstance(informational, list) or not { "unmaintained", "unsound", "notice", }.issubset(informational): fail("cargo-audit informational warnings are not fully enabled") vulnerability_list = vulnerabilities.get("list") if not isinstance(vulnerability_list, list): fail("cargo-audit vulnerability list is malformed") if vulnerabilities.get("count") != len(vulnerability_list): fail("cargo-audit vulnerability count does not match its list") if vulnerabilities.get("found") is not bool(vulnerability_list): fail("cargo-audit vulnerability flag does not match its list") findings: list[tuple[str, str, str, str]] = [] excluded: list[tuple[str, str, str, str]] = [] def classify(kind: str, item: dict[str, Any]) -> None: package = item.get("package") advisory = item.get("advisory") if not isinstance(package, dict) or not isinstance(advisory, dict): fail(f"malformed cargo-audit {kind} item") name = str(package.get("name")) version = str(package.get("version")) advisory_id = str(advisory.get("id")) record = (kind, advisory_id, name, version) if (name, version) in reachable_versions: findings.append(record) else: excluded.append(record) for item in vulnerability_list: if not isinstance(item, dict): fail("malformed cargo-audit vulnerability item") classify("vulnerability", item) for warning_kind, items in warnings.items(): if not isinstance(items, list): fail(f"cargo-audit warning list is malformed: {warning_kind}") for item in items: if not isinstance(item, dict): fail(f"malformed cargo-audit warning item: {warning_kind}") classify(f"warning:{warning_kind}", item) report_lines = [ "nigig-site production RustSec containment report", f"reachable normal/build packages (all target predicates): {len(reachable)}", f"workspace findings reported by cargo-audit: {len(findings) + len(excluded)}", f"reachable findings: {len(findings)}", f"excluded dev/unrelated-workspace findings: {len(excluded)}", "", ] for prefix, records in (("REACHABLE", findings), ("EXCLUDED", excluded)): for kind, advisory_id, name, version in sorted(records): report_lines.append( f"{prefix} {kind} {advisory_id} {name} {version}" ) report = "\n".join(report_lines) + "\n" Path(sys.argv[3]).write_text(report) print(report, end="") if findings: fail("RustSec findings are reachable from nigig-site production dependencies") if __name__ == "__main__": main()