"""Idempotently import historical State Hub SBOM snapshots into SBOM Nexus.""" from __future__ import annotations import argparse import json import socket import urllib.error import urllib.parse import urllib.request from collections import Counter from typing import Any def request_json( base_url: str, path: str, *, method: str = "GET", body: dict[str, Any] | None = None, ) -> Any: payload = json.dumps(body).encode() if body is not None else None request = urllib.request.Request( f"{base_url.rstrip('/')}{path}", data=payload, method=method, headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(request, timeout=30) as response: return json.loads(response.read()) def checkout_path(repo: dict[str, Any]) -> str | None: host_paths = repo.get("host_paths") or {} return host_paths.get(socket.gethostname()) or repo.get("local_path") def normalise_licence_groups(groups: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """Compare the report as a set of groups; API ordering is not contractual.""" return { json.dumps(group.get("license_spdx"), sort_keys=True): { "license_spdx": group.get("license_spdx"), "count": group.get("count"), "repos": sorted(group.get("repos") or []), "is_copyleft": bool(group.get("is_copyleft")), } for group in groups } def import_history(source_url: str, target_url: str, *, dry_run: bool) -> dict[str, Any]: repositories = request_json(source_url, "/repos/") repo_by_id = {str(repo["id"]): repo for repo in repositories} snapshots = request_json(source_url, "/sbom/snapshots/") snapshots.sort(key=lambda item: (item["snapshot_at"], item["id"])) results = Counter() by_repo = Counter() source_entries = 0 expected: dict[str, dict[str, Any]] = {} for snapshot in snapshots: repo = repo_by_id.get(str(snapshot["repo_id"])) if repo is None: results["missing_repo"] += 1 continue repo_slug = repo["slug"] detail = request_json(source_url, f"/sbom/snapshots/{snapshot['id']}") entries = [ { "package_name": entry["package_name"], "package_version": entry.get("package_version"), "ecosystem": entry["ecosystem"], "license_spdx": entry.get("license_spdx"), "is_direct": entry.get("is_direct", True), "is_dev": entry.get("is_dev", False), "source_path": entry.get("source_path"), } for entry in detail.get("entries", []) ] source_entries += len(entries) expected[str(snapshot["id"])] = { "repo_slug": repo_slug, "snapshot_at": snapshot["snapshot_at"], "entry_count": len(entries), } if dry_run: results["would_import"] += 1 by_repo[repo_slug] += 1 continue request_json( target_url, f"/repositories/{urllib.parse.quote(repo_slug, safe='')}", method="PUT", body={ "checkout_path": checkout_path(repo), "active": repo.get("status", "active") == "active", }, ) imported = request_json( target_url, "/sbom/import/", method="POST", body={ "repo_slug": repo_slug, "legacy_id": str(snapshot["id"]), "snapshot_at": snapshot["snapshot_at"], "source": f"state-hub:{snapshot.get('source') or 'manual'}", "entries": entries, }, ) results["imported" if imported["imported"] else "already_present"] += 1 by_repo[repo_slug] += 1 reconciliation: dict[str, Any] | None = None licence_reconciliation: dict[str, Any] | None = None if not dry_run: target_snapshots = request_json(target_url, "/sbom/snapshots/") imported = { str(snapshot["legacy_id"]): snapshot for snapshot in target_snapshots if snapshot.get("legacy_id") } missing = sorted(set(expected) - set(imported)) mismatched = [] for legacy_id in sorted(set(expected) & set(imported)): wanted = expected[legacy_id] actual = imported[legacy_id] differences = { field: {"source": wanted[field], "target": actual.get(field)} for field in ("repo_slug", "snapshot_at", "entry_count") if wanted[field] != actual.get(field) } if differences: mismatched.append({"legacy_id": legacy_id, "differences": differences}) reconciliation = { "ok": not missing and not mismatched, "expected_snapshot_count": len(expected), "matched_snapshot_count": len(expected) - len(missing) - len(mismatched), "expected_entry_count": source_entries, "target_imported_entry_count": sum( imported[legacy_id]["entry_count"] for legacy_id in expected if legacy_id in imported ), "missing_legacy_ids": missing, "mismatched_snapshots": mismatched, } source_licences = request_json(source_url, "/sbom/report/licences/") target_licences = request_json(target_url, "/sbom/report/licences/") source_groups = normalise_licence_groups(source_licences.get("groups", [])) target_groups = normalise_licence_groups(target_licences.get("groups", [])) licence_reconciliation = { "ok": source_groups == target_groups and source_licences.get("copyleft_direct_count") == target_licences.get("copyleft_direct_count"), "groups_match": source_groups == target_groups, "source_copyleft_direct_count": source_licences.get("copyleft_direct_count"), "target_copyleft_direct_count": target_licences.get("copyleft_direct_count"), } ok = results["missing_repo"] == 0 if reconciliation is not None: ok = ok and reconciliation["ok"] if licence_reconciliation is not None: ok = ok and licence_reconciliation["ok"] return { "ok": ok, "dry_run": dry_run, "source_repo_count": len(by_repo), "source_snapshot_count": len(snapshots), "source_entry_count": source_entries, "results": dict(results), "snapshots_by_repo": dict(sorted(by_repo.items())), "snapshot_reconciliation": reconciliation, "licence_reconciliation": licence_reconciliation, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source-url", default="http://127.0.0.1:8000") parser.add_argument("--target-url", default="http://127.0.0.1:8010") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() try: result = import_history(args.source_url, args.target_url, dry_run=args.dry_run) except (urllib.error.URLError, json.JSONDecodeError) as exc: raise SystemExit(f"Import failed: {exc}") from exc print(json.dumps(result, indent=2, sort_keys=True)) if not result["ok"]: raise SystemExit(1) if __name__ == "__main__": main()