tegwick
d61698ea51
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""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 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
|
|
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)
|
|
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
|
|
|
|
target_snapshots = [] if dry_run else request_json(target_url, "/sbom/snapshots/")
|
|
return {
|
|
"ok": results["missing_repo"] == 0,
|
|
"dry_run": dry_run,
|
|
"source_snapshot_count": len(snapshots),
|
|
"source_entry_count": source_entries,
|
|
"target_snapshot_count": len(target_snapshots) if not dry_run else None,
|
|
"results": dict(results),
|
|
"snapshots_by_repo": dict(sorted(by_repo.items())),
|
|
}
|
|
|
|
|
|
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()
|