feat: finish register receiving and authority routing

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
tegwick 2026-08-21 23:15:48 +02:00
parent 6d134425df
commit d103955217
28 changed files with 970 additions and 48 deletions

View file

@ -295,6 +295,25 @@ def main(argv: list[str] | None = None) -> int:
p_id_plan.add_argument("--output", default=None, help="Write the provenance mapping as JSON")
p_id_plan.add_argument("--force", action="store_true", help="Replace an existing --output file")
p_sbom = sub.add_parser("sbom", help="Derive SBOM snapshots and licence reports from repository files")
sbom_sub = p_sbom.add_subparsers(dest="sbom_command")
p_sbom_scan = sbom_sub.add_parser("scan", help="Scan recognised lockfiles and tool manifests")
p_sbom_scan.add_argument("--path", default=".")
p_sbom_scan.add_argument("--slug", default=None)
p_sbom_scan.add_argument("--output", default=None, help="Write the derived snapshot as JSON")
p_sbom_scan.add_argument("--force", action="store_true", help="Replace an existing --output file")
p_sbom_report = sbom_sub.add_parser("licence-report", help="Report licences from a fresh file scan")
p_sbom_report.add_argument("--path", default=".")
p_sbom_report.add_argument("--slug", default=None)
p_authority = sub.add_parser("authority", help="Resolve the one authoritative record owner")
authority_sub = p_authority.add_subparsers(dest="authority_command")
p_authority_route = authority_sub.add_parser("route", help="Resolve or verify an authority route")
p_authority_route.add_argument("--record-type", required=True)
p_authority_route.add_argument("--repo-slug", default=None)
p_authority_route.add_argument("--domain-slug", default=None)
p_authority_route.add_argument("--claimed-owner", default=None)
args = parser.parse_args(argv)
if args.version or args.command in (None, "version"):
@ -618,6 +637,56 @@ def main(argv: list[str] | None = None) -> int:
print(json.dumps(result, indent=2))
return 0 if result.get("ok") else 1
if args.command == "sbom":
if not args.sbom_command:
p_sbom.print_help()
return 2
from repo_manager.sbom import scan_repository
snapshot = scan_repository(Path(args.path), slug=args.slug)
result = snapshot if args.sbom_command == "scan" else {
"ok": snapshot["ok"],
"repo_slug": snapshot["repo_slug"],
"source_revision": snapshot["source_revision"],
"generated_at": snapshot["generated_at"],
"entry_count": snapshot["entry_count"],
"licence_report": snapshot["licence_report"],
"errors": snapshot["errors"],
}
if args.sbom_command == "scan" and args.output:
output = Path(args.output)
if output.exists() and not args.force:
print(
json.dumps(
{"ok": False, "error": f"output exists: {output}; use --force to replace"},
indent=2,
)
)
return 1
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8")
print(json.dumps(result, indent=2))
return 0 if result.get("ok") else 1
if args.command == "authority":
if not args.authority_command:
p_authority.print_help()
return 2
from repo_manager.authority import AuthorityError, resolve_record_authority
try:
result = resolve_record_authority(
args.record_type,
repo_slug=args.repo_slug,
domain_slug=args.domain_slug,
claimed_owner=args.claimed_owner,
)
except AuthorityError as exc:
print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
return 1
print(json.dumps({"ok": True, **result}, indent=2))
return 0
parser.print_help()
return 0