#!/usr/bin/env python3 """Create runtime SBOM, provenance, and an unsigned distribution descriptor.""" from __future__ import annotations import argparse from datetime import UTC, datetime, timedelta import hashlib import json from pathlib import Path import re import sys from typing import Any from urllib.parse import urlsplit import uuid SHA256 = re.compile(r"^[0-9a-f]{64}$") DIGEST_IMAGE = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--composition", type=Path, required=True) parser.add_argument("--api-metadata", type=Path, required=True) parser.add_argument("--web-metadata", type=Path, required=True) parser.add_argument("--deployer", type=Path, required=True) parser.add_argument("--deployer-url", required=True) parser.add_argument("--package-lock", type=Path, required=True) parser.add_argument("--artifact-base-url", required=True) parser.add_argument("--source-commit", required=True) parser.add_argument("--version", required=True) parser.add_argument("--channel", default="stable") parser.add_argument("--sequence", type=int, required=True) parser.add_argument("--expires-days", type=int, default=90) parser.add_argument( "--dependency", action="append", default=[], metavar="NAME=IMAGE@SHA256", ) parser.add_argument("--output-directory", type=Path, required=True) parser.add_argument("--descriptor", type=Path, required=True) return parser def finalize(args: argparse.Namespace) -> dict[str, Any]: composition = _json_object(args.composition) _validate_package_lock( _json_object(args.package_lock), version=args.version, composition=composition, ) api = _image_metadata(_json_object(args.api_metadata), "api") web = _image_metadata(_json_object(args.web_metadata), "web") dependencies = dict(_dependency(value) for value in args.dependency) if not dependencies: raise ValueError("at least one --dependency is required") _https_url(args.deployer_url, "deployer URL") artifact_base = _https_url(args.artifact_base_url, "artifact base URL").rstrip("/") if args.sequence < 1 or not 1 <= args.expires_days <= 365: raise ValueError("sequence and expiry window are out of bounds") output = args.output_directory.expanduser().resolve() output.mkdir(parents=True, exist_ok=True) api_sbom = _api_sbom(composition, version=args.version) web_sbom = _web_sbom(composition, version=args.version) api_provenance = _provenance( subject=api["index"], source_commit=args.source_commit, composition=composition, ) web_provenance = _provenance( subject=web["index"], source_commit=args.source_commit, composition=composition, ) artifact_values = { "api-sbom.cdx.json": api_sbom, "web-sbom.cdx.json": web_sbom, "api-provenance.json": api_provenance, "web-provenance.json": web_provenance, } artifacts: dict[str, dict[str, str]] = {} for filename, value in artifact_values.items(): path = output / filename encoded = _canonical_json(value) path.write_bytes(encoded) artifacts[filename] = { "url": f"{artifact_base}/{filename}", "sha256": hashlib.sha256(encoded).hexdigest(), } composition_encoded = _canonical_json(composition) packages = _manifest_packages(composition) issued = datetime.now(UTC).replace(microsecond=0) descriptor: dict[str, Any] = { "schema_version": "1", "channel": args.channel, "sequence": args.sequence, "version": args.version, "issued_at": issued.isoformat(), "expires_at": (issued + timedelta(days=args.expires_days)).isoformat(), "revoked": False, "deployer": { "url": args.deployer_url, "sha256": _sha256_file(args.deployer), }, "package_lock": { "url": f"{artifact_base}/package-artifacts.lock.json", "sha256": _sha256_file(args.package_lock), }, "images": { "api": { **api, "sbom": artifacts["api-sbom.cdx.json"], "provenance": artifacts["api-provenance.json"], }, "web": { **web, "sbom": artifacts["web-sbom.cdx.json"], "provenance": artifacts["web-provenance.json"], }, }, "dependencies": dict(sorted(dependencies.items())), "composition": { "sha256": hashlib.sha256(composition_encoded).hexdigest(), "module_ids": list(composition["python"]["module_ids"]), "packages": packages, }, } args.descriptor.parent.mkdir(parents=True, exist_ok=True) args.descriptor.write_bytes(_canonical_json(descriptor)) return descriptor def _api_sbom(composition: dict[str, Any], *, version: str) -> dict[str, Any]: components = [] for package in composition["python"]["packages"]: components.append( { "type": "library", "name": package["package"], "version": package["version"], "hashes": [{"alg": "SHA-256", "content": package["sha256"]}], "purl": f"pkg:pypi/{package['package']}@{package['version']}", } ) return { "bomFormat": "CycloneDX", "specVersion": "1.6", "serialNumber": f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, 'govoplan-api:' + version)}", "version": 1, "metadata": {"component": {"type": "application", "name": "govoplan-api", "version": version}}, "components": components, } def _web_sbom(composition: dict[str, Any], *, version: str) -> dict[str, Any]: return { "bomFormat": "CycloneDX", "specVersion": "1.6", "serialNumber": f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, 'govoplan-web:' + version)}", "version": 1, "metadata": {"component": {"type": "application", "name": "govoplan-web", "version": version}}, "components": [ { "type": "file", "name": "govoplan-web-dist", "version": version, "hashes": [ { "alg": "SHA-256", "content": composition["web"]["sha256"], } ], } ], } def _provenance( *, subject: str, source_commit: str, composition: dict[str, Any], ) -> dict[str, Any]: digest = subject.rsplit("@sha256:", 1)[1] return { "_type": "https://in-toto.io/Statement/v1", "subject": [{"name": subject.split("@", 1)[0], "digest": {"sha256": digest}}], "predicateType": "https://slsa.dev/provenance/v1", "predicate": { "buildDefinition": { "buildType": "https://govoplan.add-ideas.de/build/runtime-oci/v1", "externalParameters": { "source_commit": source_commit, "network_free_image_assembly": True, }, "resolvedDependencies": [ { "uri": "govoplan:runtime-composition", "digest": { "sha256": hashlib.sha256(_canonical_json(composition)).hexdigest() }, } ], }, "runDetails": { "builder": {"id": "https://git.add-ideas.de/GovOPlaN/govoplan/actions"}, "metadata": {"invocationId": source_commit}, }, }, } def _manifest_packages(composition: dict[str, Any]) -> list[dict[str, str]]: values = [] for package in composition["python"]["packages"]: values.append( { "name": str(package["package"]), "version": str(package["version"]), "wheel_sha256": str(package["sha256"]), } ) return sorted(values, key=lambda item: item["name"]) def _image_metadata(value: dict[str, Any], label: str) -> dict[str, Any]: if set(value) != {"index", "platforms"}: raise ValueError(f"{label} image metadata has invalid fields") if not isinstance(value["index"], str) or DIGEST_IMAGE.fullmatch(value["index"]) is None: raise ValueError(f"{label} index is not digest-pinned") platforms = value["platforms"] if not isinstance(platforms, dict) or set(platforms) != {"linux/amd64", "linux/arm64"}: raise ValueError(f"{label} image does not cover amd64 and arm64") if any(not isinstance(item, str) or DIGEST_IMAGE.fullmatch(item) is None for item in platforms.values()): raise ValueError(f"{label} platform image is not digest-pinned") return {"index": value["index"], "platforms": dict(sorted(platforms.items()))} def _dependency(value: str) -> tuple[str, str]: if "=" not in value: raise ValueError("--dependency must use NAME=IMAGE@SHA256") name, reference = value.split("=", 1) if re.fullmatch(r"[a-z][a-z0-9_]{1,63}", name) is None: raise ValueError(f"invalid dependency name: {name!r}") if DIGEST_IMAGE.fullmatch(reference) is None: raise ValueError(f"dependency {name!r} is not digest-pinned") return name, reference def _json_object(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"JSON root must be an object: {path}") return value def _validate_package_lock( lock: dict[str, Any], *, version: str, composition: dict[str, Any], ) -> None: if lock.get("schema_version") != "1" or lock.get("release_version") != version: raise ValueError("package lock schema or release version does not match") unsigned = dict(lock) expected_hash = unsigned.pop("lock_sha256", None) actual_hash = hashlib.sha256( json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() if expected_hash != actual_hash: raise ValueError("package lock hash does not match its contents") rows = lock.get("python") if not isinstance(rows, list): raise ValueError("package lock Python artifacts are missing") locked = _package_identities(rows, name_key="name") composed = _package_identities( composition["python"]["packages"], name_key="package", ) if locked != composed: raise ValueError("package lock does not match the runtime wheel composition") def _package_identities( rows: list[object], *, name_key: str, ) -> dict[str, tuple[str, str]]: identities: dict[str, tuple[str, str]] = {} for row in rows: if not isinstance(row, dict): raise ValueError("package artifact identity is malformed") name = row.get(name_key) version = row.get("version") digest = row.get("sha256") if ( not isinstance(name, str) or not isinstance(version, str) or not isinstance(digest, str) or SHA256.fullmatch(digest) is None or name in identities ): raise ValueError("package artifact identity is malformed or duplicated") identities[name] = (version, digest) return identities def _https_url(value: str, label: str) -> str: parsed = urlsplit(value) if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password: raise ValueError(f"{label} must be an HTTPS URL without credentials") return value def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _canonical_json(value: object) -> bytes: return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") def main() -> int: args = build_parser().parse_args() try: finalize(args) except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 print(f"Runtime release evidence written below {args.output_directory}") print(f"Unsigned descriptor written to {args.descriptor}") return 0 if __name__ == "__main__": raise SystemExit(main())