#!/usr/bin/env python3 """Verify a BuildKit SPDX attestation transitively bound to a signed OCI index.""" from __future__ import annotations import argparse import hashlib import json import re from pathlib import Path from typing import Any DIGEST = re.compile(r"^sha256:([0-9a-f]{64})$") SPDX_VERSION = re.compile(r"^SPDX-2\.[0-9]+$") STATEMENT_TYPES = { "https://in-toto.io/Statement/v0.1", "https://in-toto.io/Statement/v1", } PREDICATE_TYPE = "https://spdx.dev/Document" MANIFEST_MEDIA_TYPE = "application/vnd.oci.image.manifest.v1+json" INTOTO_MEDIA_TYPE = "application/vnd.in-toto+json" class VerificationFailure(RuntimeError): pass def load_unique(path: Path) -> Any: def unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]: value: dict[str, Any] = {} for key, item in pairs: if key in value: raise VerificationFailure(f"duplicate JSON field {key!r} in {path}") value[key] = item return value try: return json.loads(path.read_text(), object_pairs_hook=unique) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise VerificationFailure(f"cannot read {path}: {exc}") from exc def file_digest(path: Path) -> str: return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() def exact_descriptor(items: Any, predicate, label: str) -> dict[str, Any]: if not isinstance(items, list): raise VerificationFailure(f"{label} collection is not an array") matches = [item for item in items if isinstance(item, dict) and predicate(item)] if len(matches) != 1: raise VerificationFailure(f"expected exactly one {label}, got {len(matches)}") descriptor = matches[0] digest = descriptor.get("digest") size = descriptor.get("size") if not isinstance(digest, str) or not DIGEST.fullmatch(digest): raise VerificationFailure(f"{label} has no canonical digest") if not isinstance(size, int) or isinstance(size, bool) or size <= 0: raise VerificationFailure(f"{label} has no positive size") return descriptor def locate(index: dict[str, Any], subject_digest: str) -> tuple[dict[str, Any], dict[str, Any]]: if file_schema(index) != "index": raise VerificationFailure("signed subject is not an OCI image index") if not DIGEST.fullmatch(subject_digest): raise VerificationFailure("subject digest is not canonical") platform = exact_descriptor( index.get("manifests"), lambda item: item.get("mediaType") == MANIFEST_MEDIA_TYPE and item.get("platform") == {"architecture": "amd64", "os": "linux"} and item.get("annotations", {}).get("vnd.docker.reference.type") != "attestation-manifest", "linux/amd64 platform manifest", ) platform_digest = platform["digest"] attestation = exact_descriptor( index.get("manifests"), lambda item: item.get("mediaType") == MANIFEST_MEDIA_TYPE and item.get("platform") == {"architecture": "unknown", "os": "unknown"} and item.get("annotations", {}).get("vnd.docker.reference.type") == "attestation-manifest" and item.get("annotations", {}).get("vnd.docker.reference.digest") == platform_digest, "BuildKit attestation manifest", ) return platform, attestation def file_schema(value: Any) -> str: if not isinstance(value, dict) or value.get("schemaVersion") != 2: raise VerificationFailure("OCI document must be a schemaVersion 2 object") media_type = value.get("mediaType") if media_type in ( "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.list.v2+json", ): return "index" if media_type == MANIFEST_MEDIA_TYPE: return "manifest" raise VerificationFailure(f"unsupported OCI mediaType {media_type!r}") def verify( index_path: Path, subject_digest: str, attestation_path: Path, sbom_path: Path, ) -> dict[str, Any]: if file_digest(index_path) != subject_digest: raise VerificationFailure("OCI index bytes do not match the owner-signed subject digest") index = load_unique(index_path) platform, attestation_descriptor = locate(index, subject_digest) if file_digest(attestation_path) != attestation_descriptor["digest"]: raise VerificationFailure("attestation manifest bytes do not match the signed index descriptor") if attestation_path.stat().st_size != attestation_descriptor["size"]: raise VerificationFailure("attestation manifest size does not match the signed index descriptor") attestation = load_unique(attestation_path) if file_schema(attestation) != "manifest": raise VerificationFailure("attestation descriptor does not resolve to an OCI manifest") sbom_descriptor = exact_descriptor( attestation.get("layers"), lambda item: item.get("mediaType") == INTOTO_MEDIA_TYPE and item.get("annotations", {}).get("in-toto.io/predicate-type") == PREDICATE_TYPE, "SPDX in-toto layer", ) raw_sbom = sbom_path.read_bytes() if len(raw_sbom) != sbom_descriptor["size"]: raise VerificationFailure("SBOM blob size does not match its attestation descriptor") if file_digest(sbom_path) != sbom_descriptor["digest"]: raise VerificationFailure("SBOM blob bytes do not match their attestation descriptor") statement = load_unique(sbom_path) if not isinstance(statement, dict) or statement.get("_type") not in STATEMENT_TYPES: raise VerificationFailure("SBOM blob is not a supported in-toto statement") if statement.get("predicateType") != PREDICATE_TYPE: raise VerificationFailure("SBOM predicateType does not match its descriptor") subjects = statement.get("subject") if not isinstance(subjects, list) or not subjects or not all(isinstance(item, dict) for item in subjects): raise VerificationFailure("SBOM must have at least one platform subject") expected_hex = platform["digest"].removeprefix("sha256:") for subject in subjects: if not isinstance(subject.get("name"), str) or not subject["name"]: raise VerificationFailure("SBOM subject name is empty") if subject.get("digest") != {"sha256": expected_hex}: raise VerificationFailure("SBOM subject does not match the indexed linux/amd64 manifest") spdx = statement.get("predicate") if not isinstance(spdx, dict): raise VerificationFailure("SBOM predicate is not an SPDX document") required_strings = { "SPDXID": "SPDXRef-DOCUMENT", "dataLicense": "CC0-1.0", } for field, expected in required_strings.items(): if spdx.get(field) != expected: raise VerificationFailure(f"SPDX {field} must be {expected}") if not isinstance(spdx.get("spdxVersion"), str) or not SPDX_VERSION.fullmatch(spdx["spdxVersion"]): raise VerificationFailure("SPDX version is missing or unsupported") for field in ("name", "documentNamespace"): if not isinstance(spdx.get(field), str) or not spdx[field]: raise VerificationFailure(f"SPDX {field} is empty") creation = spdx.get("creationInfo") if not isinstance(creation, dict): raise VerificationFailure("SPDX creationInfo is missing") if not isinstance(creation.get("created"), str) or not creation["created"]: raise VerificationFailure("SPDX creation timestamp is empty") creators = creation.get("creators") if not isinstance(creators, list) or not creators or not all(isinstance(item, str) and item for item in creators): raise VerificationFailure("SPDX creators are missing") return { "schema_version": "nexuscrm.buildkit-sbom-verification/v1", "subject_index_digest": subject_digest, "platform_manifest_digest": platform["digest"], "attestation_manifest_digest": attestation_descriptor["digest"], "sbom_blob_digest": sbom_descriptor["digest"], "predicate_type": PREDICATE_TYPE, "spdx_version": spdx["spdxVersion"], } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--index", required=True) parser.add_argument("--subject-digest", required=True) parser.add_argument("--attestation-manifest", required=True) parser.add_argument("--sbom", required=True) parser.add_argument("--output", required=True) args = parser.parse_args() try: result = verify( Path(args.index).resolve(), args.subject_digest, Path(args.attestation_manifest).resolve(), Path(args.sbom).resolve(), ) except (OSError, VerificationFailure) as exc: print(f"BuildKit SBOM verification failed: {exc}") return 1 output = Path(args.output).resolve() output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())