From e2de02ea5e53ce5341b2c728f26d83b409ef23a7 Mon Sep 17 00:00:00 2001 From: yuqei Date: Sun, 27 Sep 2026 02:15:43 +0800 Subject: [PATCH] build: validate third-party release materials --- .../scripts/build-third-party-materials.py | 880 ++++++++++++++++++ .../tests/test_third_party_materials.py | 482 ++++++++++ 2 files changed, 1362 insertions(+) create mode 100644 yuqei-infra/scripts/build-third-party-materials.py create mode 100644 yuqei-infra/tests/test_third_party_materials.py diff --git a/yuqei-infra/scripts/build-third-party-materials.py b/yuqei-infra/scripts/build-third-party-materials.py new file mode 100644 index 00000000..081db449 --- /dev/null +++ b/yuqei-infra/scripts/build-third-party-materials.py @@ -0,0 +1,880 @@ +#!/usr/bin/env python3 +"""Build and independently verify deterministic third-party materials. + +The tool is deliberately offline and fail closed. It does not download +licenses or source code, resolve an ambiguous license, or grant legal or +distribution approval. Every byte it consumes must be listed in one fixed +input manifest and every Trivy license occurrence must have exactly one +coverage record. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import shutil +import stat +import sys +import tarfile +import tempfile +import zipfile +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Any, Iterable, Mapping, Sequence + + +INPUT_SCHEMA = "yuqei.third-party-materials-input.v1" +OUTPUT_SCHEMA = "yuqei.third-party-materials.v1" +COVERAGE_SCHEMA = "yuqei.third-party-occurrence-coverage.v1" +OFFER_SCHEMA = "yuqei.written-source-offer.v1" +VERIFY_SCHEMA = "yuqei.third-party-materials-verification.v1" +PROFILES = {"internal-hosted", "redistributable"} +EXPECTED_ROLES = { + "contract-web", + "contract-api", + "ai-platform-api", + "ai-worker", + "postgres-runtime", + "postgres-client", + "redis", + "minio", + "minio-client", + "nginx-edge", + "backup-utility", +} +HEX64 = re.compile(r"^[0-9a-f]{64}$") +DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +COMMIT = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") +SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +MAX_INPUT_FILE = 2 * 1024 * 1024 * 1024 +MAX_LICENSE_TEXT = 8 * 1024 * 1024 +MAX_ARCHIVE_FILES = 200_000 +MAX_ARCHIVE_EXPANDED = 8 * 1024 * 1024 * 1024 +MAX_COMPRESSION_RATIO = 2_000 + + +class MaterialsError(RuntimeError): + """A deterministic validation failure.""" + + +def _canonical_json(value: Any) -> bytes: + return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode( + "utf-8" + ) + + +def _pretty_json(value: Any) -> bytes: + return (json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode("utf-8") + + +def _sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _hash_file(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + with path.open("rb") as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + size += len(chunk) + if size > MAX_INPUT_FILE: + raise MaterialsError(f"input exceeds {MAX_INPUT_FILE} bytes: {path}") + return digest.hexdigest(), size + + +def _mapping(value: Any, label: str) -> Mapping[str, Any]: + if not isinstance(value, dict): + raise MaterialsError(f"{label} must be an object") + return value + + +def _list(value: Any, label: str) -> list[Any]: + if not isinstance(value, list): + raise MaterialsError(f"{label} must be an array") + return value + + +def _text(value: Any, label: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise MaterialsError(f"{label} must be a non-empty string") + return value + + +def _unique(values: Sequence[str], label: str) -> None: + if len(values) != len(set(values)): + raise MaterialsError(f"duplicate {label}") + + +def _unique_casefold(values: Sequence[str], label: str) -> None: + folded = [value.casefold() for value in values] + if len(folded) != len(set(folded)): + raise MaterialsError(f"case-colliding or duplicate {label}") + + +def _is_linklike(path: Path, info: os.stat_result | None = None) -> bool: + if path.is_symlink(): + return True + if hasattr(path, "is_junction") and path.is_junction(): + return True + details = info if info is not None else path.lstat() + attributes = getattr(details, "st_file_attributes", 0) + reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + return bool(reparse_flag and attributes & reparse_flag) + + +def _safe_relative(value: Any, label: str) -> str: + raw = _text(value, label) + if "\\" in raw or "\x00" in raw or re.match(r"^[A-Za-z]:", raw): + raise MaterialsError(f"unsafe {label}: {raw!r}") + path = PurePosixPath(raw) + if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts): + raise MaterialsError(f"unsafe {label}: {raw!r}") + return path.as_posix() + + +def _assert_regular_no_links(path: Path, root: Path, label: str) -> None: + try: + root_info = root.lstat() + except OSError as exc: + raise MaterialsError(f"missing input root for {label}: {root}") from exc + if not stat.S_ISDIR(root_info.st_mode) or _is_linklike(root, root_info): + raise MaterialsError(f"input root for {label} must be a real directory: {root}") + try: + resolved_root = root.resolve(strict=True) + resolved = path.resolve(strict=True) + except OSError as exc: + raise MaterialsError(f"missing {label}: {path}") from exc + try: + resolved.relative_to(resolved_root) + except ValueError as exc: + raise MaterialsError(f"{label} escapes input root: {path}") from exc + + relative = path.relative_to(root) + current = root + for part in relative.parts: + current = current / part + try: + info = current.lstat() + except OSError as exc: + raise MaterialsError(f"missing {label}: {path}") from exc + if _is_linklike(current, info): + raise MaterialsError(f"symlink/junction is forbidden for {label}: {path}") + info = path.lstat() + if not stat.S_ISREG(info.st_mode): + raise MaterialsError(f"{label} is not a regular file: {path}") + if info.st_nlink != 1: + raise MaterialsError(f"hardlink is forbidden for {label}: {path}") + + +def _resolve_input(root: Path, relative: Any, label: str) -> tuple[str, Path]: + safe = _safe_relative(relative, label) + path = root.joinpath(*PurePosixPath(safe).parts) + _assert_regular_no_links(path, root, label) + return safe, path + + +def _file_record(root: Path, raw: Any, label: str) -> tuple[dict[str, Any], Path]: + record = _mapping(raw, label) + allowed = {"path", "sha256", "size_bytes"} + if set(record) != allowed: + raise MaterialsError(f"{label} must contain exactly {sorted(allowed)}") + relative, path = _resolve_input(root, record.get("path"), f"{label}.path") + expected_hash = _text(record.get("sha256"), f"{label}.sha256") + if not HEX64.fullmatch(expected_hash): + raise MaterialsError(f"invalid SHA-256 for {label}") + expected_size = record.get("size_bytes") + if not isinstance(expected_size, int) or isinstance(expected_size, bool) or expected_size < 1: + raise MaterialsError(f"invalid size for {label}") + actual_hash, actual_size = _hash_file(path) + if (actual_hash, actual_size) != (expected_hash, expected_size): + raise MaterialsError( + f"{label} size/hash mismatch: expected {expected_size}/{expected_hash}, " + f"actual {actual_size}/{actual_hash}" + ) + return {"path": relative, "sha256": actual_hash, "size_bytes": actual_size}, path + + +def _load_json(path: Path, label: str) -> Mapping[str, Any]: + def no_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise MaterialsError(f"duplicate JSON key in {label}: {key}") + result[key] = value + return result + + try: + value = json.loads( + path.read_text(encoding="utf-8"), + object_pairs_hook=no_duplicate_pairs, + parse_constant=lambda value: (_ for _ in ()).throw( + MaterialsError(f"non-finite JSON value in {label}: {value}") + ), + ) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise MaterialsError(f"invalid JSON for {label}: {path}") from exc + return _mapping(value, label) + + +def _parse_time(value: Any, label: str) -> datetime: + text = _text(value, label) + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise MaterialsError(f"invalid ISO-8601 timestamp for {label}") from exc + if parsed.tzinfo is None: + raise MaterialsError(f"{label} must include a timezone") + return parsed.astimezone(timezone.utc) + + +def _validate_archive(path: Path, label: str) -> dict[str, int]: + file_count = 0 + expanded = 0 + names: set[str] = set() + + def accept_name(name: str) -> str: + safe = _safe_relative(name.rstrip("/"), f"{label} member") + if safe in names: + raise MaterialsError(f"duplicate archive member in {label}: {safe}") + names.add(safe) + return safe + + try: + if zipfile.is_zipfile(path): + with zipfile.ZipFile(path) as archive: + for member in archive.infolist(): + accept_name(member.filename) + mode = (member.external_attr >> 16) & 0xFFFF + if mode and stat.S_ISLNK(mode): + raise MaterialsError(f"symlink in {label}: {member.filename}") + if member.flag_bits & 0x1: + raise MaterialsError(f"encrypted member in {label}: {member.filename}") + if member.is_dir(): + continue + if mode and not stat.S_ISREG(mode): + raise MaterialsError(f"non-regular member in {label}: {member.filename}") + if member.compress_size and member.file_size > member.compress_size * MAX_COMPRESSION_RATIO: + raise MaterialsError(f"excessive compression ratio in {label}: {member.filename}") + file_count += 1 + expanded += member.file_size + else: + with tarfile.open(path, "r:*") as archive: + for member in archive: + accept_name(member.name) + if member.issym() or member.islnk(): + raise MaterialsError(f"symlink/hardlink in {label}: {member.name}") + if member.isdir(): + continue + if not member.isfile(): + raise MaterialsError(f"non-regular member in {label}: {member.name}") + file_count += 1 + expanded += member.size + except (tarfile.TarError, zipfile.BadZipFile, OSError) as exc: + raise MaterialsError(f"invalid source archive for {label}: {path}") from exc + if file_count < 1: + raise MaterialsError(f"empty source archive for {label}") + if file_count > MAX_ARCHIVE_FILES or expanded > MAX_ARCHIVE_EXPANDED: + raise MaterialsError(f"source archive limits exceeded for {label}") + compressed_size = path.stat().st_size + if compressed_size and expanded > compressed_size * MAX_COMPRESSION_RATIO: + raise MaterialsError(f"excessive compression ratio in {label}") + return {"file_count": file_count, "expanded_size_bytes": expanded} + + +def _validate_sbom(payload: Mapping[str, Any], role: str, manifest_digest: str) -> dict[str, Any]: + if payload.get("spdxVersion") != "SPDX-2.3" or payload.get("SPDXID") != "SPDXRef-DOCUMENT": + raise MaterialsError(f"{role} SBOM is not SPDX-2.3") + packages = _list(payload.get("packages"), f"{role} SBOM packages") + if not packages: + raise MaterialsError(f"{role} SBOM package inventory is empty") + ids: list[str] = [] + roots = 0 + for index, raw in enumerate(packages): + package = _mapping(raw, f"{role} SBOM package {index}") + spdx_id = _text(package.get("SPDXID"), f"{role} SBOM package SPDXID") + ids.append(spdx_id) + if package.get("versionInfo") == manifest_digest: + roots += 1 + _unique(ids, f"{role} SBOM SPDXID") + if roots != 1: + raise MaterialsError(f"{role} SBOM must bind exactly one root package to {manifest_digest}") + creators = _mapping(payload.get("creationInfo"), f"{role} SBOM creationInfo").get("creators") + if not isinstance(creators, list) or not any("syft" in str(item).lower() for item in creators): + raise MaterialsError(f"{role} SBOM is not identified as a Syft document") + return {"package_count": len(packages), "root_manifest_digest": manifest_digest} + + +def _license_is_unresolved(item: Mapping[str, Any]) -> bool: + name = str(item.get("Name", "")).strip() + category = str(item.get("Category", "")).strip().lower() + severity = str(item.get("Severity", "")).strip().upper() + lowered = name.lower() + return ( + not name + or category in {"unknown", "unresolved"} + or severity in {"UNKNOWN", "UNRESOLVED"} + or lowered in {"unknown", "unresolved", "noassertion", "none", "noderivs", "verbatim"} + or "unknown" in lowered + or "unresolved" in lowered + or "licenseref-" in lowered + ) + + +def _occurrence_id(record: Mapping[str, Any]) -> str: + return "occ-" + _sha256_bytes(_canonical_json(record)) + + +def _parse_trivy( + payload: Mapping[str, Any], role: str, config_digest: str, report_sha256: str +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + if payload.get("ArtifactType") != "container_image": + raise MaterialsError(f"{role} Trivy report is not a container image report") + metadata = _mapping(payload.get("Metadata"), f"{role} Trivy metadata") + if metadata.get("ImageID") != config_digest: + raise MaterialsError(f"{role} Trivy ImageID/config digest mismatch") + results = _list(payload.get("Results"), f"{role} Trivy results") + occurrences: list[dict[str, Any]] = [] + unresolved: list[str] = [] + for result_index, raw_result in enumerate(results): + result = _mapping(raw_result, f"{role} Trivy result {result_index}") + licenses = result.get("Licenses", []) + if licenses is None: + licenses = [] + for license_index, raw_license in enumerate(_list(licenses, f"{role} Trivy licenses")): + item = _mapping(raw_license, f"{role} Trivy license {result_index}/{license_index}") + occurrence = { + "role": role, + "report_sha256": report_sha256, + "result_index": result_index, + "license_index": license_index, + "target": str(result.get("Target", "")), + "class": str(result.get("Class", "")), + "package_name": str(item.get("PkgName", "")), + "file_path": str(item.get("FilePath", "")), + "license_name": str(item.get("Name", "")), + "category": str(item.get("Category", "")), + "severity": str(item.get("Severity", "")), + "link": str(item.get("Link", "")), + } + occurrence["occurrence_id"] = _occurrence_id(occurrence) + occurrences.append(occurrence) + if _license_is_unresolved(item): + unresolved.append(occurrence["occurrence_id"]) + if not occurrences: + raise MaterialsError(f"{role} Trivy license inventory is empty") + ids = [item["occurrence_id"] for item in occurrences] + _unique(ids, f"{role} Trivy occurrence identity") + if unresolved: + preview = ", ".join(unresolved[:5]) + raise MaterialsError( + f"{role} contains {len(unresolved)} unresolved/unknown/LicenseRef occurrences: {preview}" + ) + return occurrences, {"occurrence_count": len(occurrences), "image_id": config_digest} + + +def _offer_details( + path: Path, + artifact_id: str, + candidate: Mapping[str, Any], + image_manifest_digests: Mapping[str, str], + profile: str, + as_of: datetime, +) -> dict[str, Any]: + if profile != "redistributable": + raise MaterialsError("written source offers are forbidden for internal-hosted profile") + offer = _load_json(path, f"source offer {artifact_id}") + if offer.get("schema_version") != OFFER_SCHEMA or offer.get("offer_id") != artifact_id: + raise MaterialsError(f"invalid source offer identity for {artifact_id}") + if offer.get("status") != "APPROVED" or offer.get("profile") != profile: + raise MaterialsError(f"source offer {artifact_id} is not approved for {profile}") + if offer.get("candidate_commit") != candidate["commit"]: + raise MaterialsError(f"source offer {artifact_id} candidate mismatch") + if offer.get("source_manifest_sha256") != candidate["source_manifest_sha256"]: + raise MaterialsError(f"source offer {artifact_id} source manifest mismatch") + if offer.get("image_manifest_digests") != image_manifest_digests: + raise MaterialsError(f"source offer {artifact_id} image manifest binding mismatch") + issuer = _mapping(offer.get("issuer"), f"source offer {artifact_id} issuer") + for key in ("legal_name", "approver_name", "approver_title", "contact"): + _text(issuer.get(key), f"source offer {artifact_id} issuer.{key}") + _text(offer.get("authorization_id"), f"source offer {artifact_id} authorization_id") + _text(offer.get("approval_provenance_sha256"), f"source offer {artifact_id} approval_provenance_sha256") + if not HEX64.fullmatch(str(offer["approval_provenance_sha256"])): + raise MaterialsError(f"source offer {artifact_id} has invalid approval provenance hash") + valid_from = _parse_time(offer.get("valid_from"), f"source offer {artifact_id} valid_from") + valid_until = _parse_time(offer.get("valid_until"), f"source offer {artifact_id} valid_until") + if not valid_from <= as_of < valid_until: + raise MaterialsError(f"source offer {artifact_id} is not active at manifest as_of") + covers = [_text(item, f"source offer {artifact_id} covers item") for item in _list(offer.get("covers"), "offer covers")] + _unique(covers, f"source offer {artifact_id} covers") + return {"valid_from": valid_from.isoformat(), "valid_until": valid_until.isoformat(), "covers": covers} + + +def _compile(input_manifest: Path) -> dict[str, bytes]: + manifest_path = input_manifest.absolute() + root = manifest_path.parent + _assert_regular_no_links(manifest_path, root, "input manifest") + input_manifest_hash, input_manifest_size = _hash_file(manifest_path) + raw_manifest = _load_json(manifest_path, "input manifest") + if raw_manifest.get("schema_version") != INPUT_SCHEMA: + raise MaterialsError(f"input manifest schema must be {INPUT_SCHEMA}") + expected_top = {"schema_version", "profile", "as_of", "candidate", "license_texts", "source_materials", "coverage"} + if set(raw_manifest) != expected_top: + raise MaterialsError(f"input manifest keys must be exactly {sorted(expected_top)}") + profile = _text(raw_manifest.get("profile"), "profile") + if profile not in PROFILES: + raise MaterialsError(f"unsupported profile: {profile}") + as_of_text = _text(raw_manifest.get("as_of"), "as_of") + as_of = _parse_time(as_of_text, "as_of") + + candidate_raw = _mapping(raw_manifest.get("candidate"), "candidate") + expected_candidate_keys = {"commit", "tag", "source_manifest_sha256", "images"} + if set(candidate_raw) != expected_candidate_keys: + raise MaterialsError(f"candidate keys must be exactly {sorted(expected_candidate_keys)}") + commit = _text(candidate_raw.get("commit"), "candidate.commit") + source_manifest_sha = _text(candidate_raw.get("source_manifest_sha256"), "candidate.source_manifest_sha256") + if not COMMIT.fullmatch(commit) or not HEX64.fullmatch(source_manifest_sha): + raise MaterialsError("candidate commit/source manifest identity is invalid") + candidate = { + "commit": commit, + "tag": _text(candidate_raw.get("tag"), "candidate.tag"), + "source_manifest_sha256": source_manifest_sha, + } + images = _list(candidate_raw.get("images"), "candidate.images") + if len(images) != len(EXPECTED_ROLES): + raise MaterialsError("candidate must bind exactly eleven application and infrastructure images") + + all_occurrences: list[dict[str, Any]] = [] + image_summaries: list[dict[str, Any]] = [] + input_records: list[dict[str, Any]] = [] + roles: list[str] = [] + manifest_digests: list[str] = [] + config_digests: list[str] = [] + for image_index, raw_image in enumerate(images): + image = _mapping(raw_image, f"candidate image {image_index}") + allowed = {"role", "manifest_digest", "config_digest", "sbom", "trivy_license"} + if set(image) != allowed: + raise MaterialsError(f"candidate image {image_index} keys must be exactly {sorted(allowed)}") + role = _text(image.get("role"), f"candidate image {image_index} role") + manifest_digest = _text(image.get("manifest_digest"), f"{role} manifest_digest") + config_digest = _text(image.get("config_digest"), f"{role} config_digest") + if role not in EXPECTED_ROLES or not DIGEST.fullmatch(manifest_digest) or not DIGEST.fullmatch(config_digest): + raise MaterialsError(f"invalid image identity for {role}") + roles.append(role) + manifest_digests.append(manifest_digest) + config_digests.append(config_digest) + sbom_record, sbom_path = _file_record(root, image.get("sbom"), f"{role} SBOM") + trivy_record, trivy_path = _file_record(root, image.get("trivy_license"), f"{role} Trivy report") + sbom_summary = _validate_sbom(_load_json(sbom_path, f"{role} SBOM"), role, manifest_digest) + occurrences, trivy_summary = _parse_trivy( + _load_json(trivy_path, f"{role} Trivy report"), role, config_digest, trivy_record["sha256"] + ) + all_occurrences.extend(occurrences) + input_records.extend( + [ + {"kind": "sbom", "role": role, **sbom_record}, + {"kind": "trivy-license", "role": role, **trivy_record}, + ] + ) + image_summaries.append( + { + "role": role, + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "sbom": sbom_record, + "trivy_license": trivy_record, + "sbom_summary": sbom_summary, + "trivy_summary": trivy_summary, + } + ) + _unique(roles, "image role") + if set(roles) != EXPECTED_ROLES: + raise MaterialsError("candidate image role set mismatch") + _unique(manifest_digests, "image manifest digest") + _unique(config_digests, "image config digest") + image_manifest_digest_map = { + summary["role"]: summary["manifest_digest"] for summary in sorted(image_summaries, key=lambda item: item["role"]) + } + all_occurrences.sort(key=lambda item: item["occurrence_id"]) + _unique([item["occurrence_id"] for item in all_occurrences], "global occurrence identity") + occurrence_by_id = {item["occurrence_id"]: item for item in all_occurrences} + + license_defs: dict[str, dict[str, Any]] = {} + license_payloads: dict[str, bytes] = {} + license_paths: list[str] = [] + for index, raw_license in enumerate(_list(raw_manifest.get("license_texts"), "license_texts")): + record = _mapping(raw_license, f"license_texts[{index}]") + if set(record) != {"id", "kind", "file"}: + raise MaterialsError("license text records require exactly id, kind and file") + artifact_id = _text(record.get("id"), f"license_texts[{index}].id") + if not SAFE_ID.fullmatch(artifact_id) or artifact_id in license_defs: + raise MaterialsError(f"invalid or duplicate license text id: {artifact_id}") + kind = _text(record.get("kind"), f"license_texts[{index}].kind") + if kind not in {"license", "copyright", "notice"}: + raise MaterialsError(f"invalid license text kind: {kind}") + file_record, path = _file_record(root, record.get("file"), f"license text {artifact_id}") + if file_record["size_bytes"] > MAX_LICENSE_TEXT: + raise MaterialsError(f"license text too large: {artifact_id}") + payload = path.read_bytes() + if b"\x00" in payload: + raise MaterialsError(f"license text contains NUL: {artifact_id}") + try: + payload.decode("utf-8") + except UnicodeDecodeError as exc: + raise MaterialsError(f"license text is not UTF-8: {artifact_id}") from exc + output_path = f"license-texts/{artifact_id}--{file_record['sha256'][:16]}.txt" + license_paths.append(file_record["path"]) + license_defs[artifact_id] = { + "id": artifact_id, + "kind": kind, + "input": file_record, + "output_path": output_path, + } + license_payloads[output_path] = payload + if not license_defs: + raise MaterialsError("at least one license text is required") + _unique_casefold(license_paths, "license input path") + _unique_casefold(list(license_defs), "license text id") + _unique([item["input"]["sha256"] for item in license_defs.values()], "license text content") + + source_defs: dict[str, dict[str, Any]] = {} + source_payloads: dict[str, bytes] = {} + source_paths: list[str] = [] + for index, raw_source in enumerate(_list(raw_manifest.get("source_materials"), "source_materials")): + record = _mapping(raw_source, f"source_materials[{index}]") + if set(record) != {"id", "kind", "file"}: + raise MaterialsError("source material records require exactly id, kind and file") + artifact_id = _text(record.get("id"), f"source_materials[{index}].id") + if not SAFE_ID.fullmatch(artifact_id) or artifact_id in source_defs: + raise MaterialsError(f"invalid or duplicate source material id: {artifact_id}") + kind = _text(record.get("kind"), f"source_materials[{index}].kind") + if kind not in {"source-archive", "written-source-offer"}: + raise MaterialsError(f"invalid source material kind: {kind}") + file_record, path = _file_record(root, record.get("file"), f"source material {artifact_id}") + if kind == "source-archive": + details = _validate_archive(path, f"source material {artifact_id}") + output_path = f"corresponding-source/{artifact_id}--{file_record['sha256'][:16]}.source" + else: + details = _offer_details( + path, artifact_id, candidate, image_manifest_digest_map, profile, as_of + ) + output_path = f"source-offers/{artifact_id}--{file_record['sha256'][:16]}.json" + source_paths.append(file_record["path"]) + source_defs[artifact_id] = { + "id": artifact_id, + "kind": kind, + "input": file_record, + "output_path": output_path, + "details": details, + } + source_payloads[output_path] = path.read_bytes() + _unique_casefold(source_paths, "source material input path") + _unique_casefold(list(source_defs), "source material id") + _unique([item["input"]["sha256"] for item in source_defs.values()], "source material content") + + coverage_list = _list(raw_manifest.get("coverage"), "coverage") + coverage_by_id: dict[str, dict[str, Any]] = {} + used_license_ids: set[str] = set() + used_source_ids: set[str] = set() + for index, raw_coverage in enumerate(coverage_list): + coverage = _mapping(raw_coverage, f"coverage[{index}]") + if set(coverage) != {"occurrence_id", "license_text_ids", "source_material_ids"}: + raise MaterialsError("coverage records require exactly occurrence_id, license_text_ids and source_material_ids") + occurrence_id = _text(coverage.get("occurrence_id"), f"coverage[{index}].occurrence_id") + if occurrence_id in coverage_by_id: + raise MaterialsError(f"duplicate coverage record: {occurrence_id}") + license_ids = [_text(item, "coverage license id") for item in _list(coverage.get("license_text_ids"), "license ids")] + source_ids = [_text(item, "coverage source id") for item in _list(coverage.get("source_material_ids"), "source ids")] + if not license_ids: + raise MaterialsError(f"coverage {occurrence_id} has no license/copyright text") + _unique(license_ids, f"coverage {occurrence_id} license ids") + _unique(source_ids, f"coverage {occurrence_id} source ids") + missing_license = set(license_ids) - set(license_defs) + missing_source = set(source_ids) - set(source_defs) + if missing_license or missing_source: + raise MaterialsError( + f"coverage {occurrence_id} references missing artifacts: " + f"licenses={sorted(missing_license)}, sources={sorted(missing_source)}" + ) + occurrence = occurrence_by_id.get(occurrence_id) + if occurrence is None: + raise MaterialsError(f"extra coverage record: {occurrence_id}") + if occurrence["category"].strip().lower() == "restricted" and not source_ids: + raise MaterialsError(f"restricted occurrence lacks corresponding source/source offer: {occurrence_id}") + if not any(license_defs[item]["kind"] in {"license", "copyright"} for item in license_ids): + raise MaterialsError(f"coverage {occurrence_id} lacks a license or copyright text") + for source_id in source_ids: + source = source_defs[source_id] + if source["kind"] == "written-source-offer" and occurrence_id not in source["details"]["covers"]: + raise MaterialsError(f"source offer {source_id} does not cover {occurrence_id}") + used_license_ids.update(license_ids) + used_source_ids.update(source_ids) + coverage_by_id[occurrence_id] = { + "occurrence_id": occurrence_id, + "license_text_ids": sorted(license_ids), + "source_material_ids": sorted(source_ids), + } + missing_coverage = set(occurrence_by_id) - set(coverage_by_id) + if missing_coverage: + raise MaterialsError(f"uncovered Trivy occurrences: {sorted(missing_coverage)[:5]}") + if set(coverage_by_id) != set(occurrence_by_id): + raise MaterialsError("coverage is not an exact occurrence multiset") + if used_license_ids != set(license_defs): + raise MaterialsError(f"unused/extra license artifacts: {sorted(set(license_defs) - used_license_ids)}") + if used_source_ids != set(source_defs): + raise MaterialsError(f"unused/extra source artifacts: {sorted(set(source_defs) - used_source_ids)}") + for source_id, source in source_defs.items(): + if source["kind"] != "written-source-offer": + continue + actually_covered = { + occurrence_id + for occurrence_id, coverage in coverage_by_id.items() + if source_id in coverage["source_material_ids"] + } + declared = set(source["details"]["covers"]) + if declared != actually_covered: + raise MaterialsError( + f"source offer {source_id} coverage mismatch: " + f"missing={sorted(actually_covered - declared)}, extra={sorted(declared - actually_covered)}" + ) + + matrix = { + "schema_version": COVERAGE_SCHEMA, + "profile": profile, + "candidate": candidate, + "occurrences": [ + { + **occurrence, + **coverage_by_id[occurrence["occurrence_id"]], + } + for occurrence in all_occurrences + ], + "license_texts": [license_defs[key] for key in sorted(license_defs)], + "source_materials": [source_defs[key] for key in sorted(source_defs)], + "summary": { + "occurrence_count": len(all_occurrences), + "covered_occurrence_count": len(coverage_by_id), + "unresolved_occurrence_count": 0, + "license_text_count": len(license_defs), + "source_material_count": len(source_defs), + "restricted_occurrence_count": sum( + item["category"].strip().lower() == "restricted" for item in all_occurrences + ), + }, + } + + notices: list[str] = [ + "# THIRD_PARTY_NOTICES", + "", + f"Profile: `{profile}`", + f"Candidate: `{candidate['commit']}` / `{candidate['tag']}`", + "", + "This deterministic technical inventory does not grant legal approval, distribution clearance,", + "third-party attestation, production authorization, or permission to treat unresolved licenses as resolved.", + "", + "## License occurrences", + "", + ] + for occurrence in all_occurrences: + coverage = coverage_by_id[occurrence["occurrence_id"]] + package = occurrence["package_name"] or occurrence["file_path"] or "(unidentified package/file)" + notices.extend( + [ + f"### {occurrence['occurrence_id']}", + "", + f"- Image role: `{occurrence['role']}`", + f"- Component: `{package}`", + f"- Detected license: `{occurrence['license_name']}`", + f"- Category: `{occurrence['category']}`", + f"- License/copyright text IDs: `{', '.join(coverage['license_text_ids'])}`", + f"- Corresponding-source/source-offer IDs: `{', '.join(coverage['source_material_ids']) or 'not-required-by-technical-policy'}`", + "", + ] + ) + notices.extend(["## License, copyright, and notice texts", ""]) + for artifact_id in sorted(license_defs): + definition = license_defs[artifact_id] + text = license_payloads[definition["output_path"]].decode("utf-8") + notices.extend( + [ + f"### {artifact_id}", + "", + f"Kind: `{definition['kind']}`; SHA-256: `{definition['input']['sha256']}`", + "", + text.rstrip("\n"), + "", + ] + ) + payloads: dict[str, bytes] = { + "THIRD_PARTY_NOTICES.md": ("\n".join(notices).rstrip() + "\n").encode("utf-8"), + "occurrence-coverage.json": _pretty_json(matrix), + **license_payloads, + **source_payloads, + } + output_records = [] + for relative in sorted(payloads): + data = payloads[relative] + output_records.append({"path": relative, "sha256": _sha256_bytes(data), "size_bytes": len(data)}) + source_manifest = { + "schema_version": "yuqei.corresponding-source-manifest.v1", + "profile": profile, + "candidate": candidate, + "materials": [source_defs[key] for key in sorted(source_defs)], + "technical_complete": True, + "legal_or_distribution_approval_claimed": False, + } + payloads["corresponding-source-manifest.json"] = _pretty_json(source_manifest) + output_records.append( + { + "path": "corresponding-source-manifest.json", + "sha256": _sha256_bytes(payloads["corresponding-source-manifest.json"]), + "size_bytes": len(payloads["corresponding-source-manifest.json"]), + } + ) + output_records.sort(key=lambda item: item["path"]) + manifest = { + "schema_version": OUTPUT_SCHEMA, + "profile": profile, + "as_of": as_of_text, + "candidate": {**candidate, "images": sorted(image_summaries, key=lambda item: item["role"])}, + "inputs": sorted(input_records, key=lambda item: (item["role"], item["kind"])), + "input_manifest": { + "sha256": input_manifest_hash, + "size_bytes": input_manifest_size, + }, + "summary": matrix["summary"], + "output_files": output_records, + "result": "PASS_TECHNICAL_MATERIALS_COMPLETE", + "distribution_clearance_claimed": False, + "claims_not_made": [ + "legal opinion", + "distribution authorization", + "production authorization", + "resolution of any unknown, unresolved or LicenseRef finding", + ], + } + manifest["manifest_payload_sha256"] = _sha256_bytes(_canonical_json(manifest)) + payloads["materials-manifest.json"] = _pretty_json(manifest) + return payloads + + +def _safe_output_root(path: Path) -> None: + if path.exists(): + if _is_linklike(path) or not path.is_dir(): + raise MaterialsError(f"output path is not a regular directory: {path}") + if any(path.iterdir()): + raise MaterialsError(f"output directory must be empty: {path}") + + +def build(input_manifest: Path, output_dir: Path) -> dict[str, Any]: + payloads = _compile(input_manifest) + output_dir = output_dir.absolute() + _safe_output_root(output_dir) + output_dir.parent.mkdir(parents=True, exist_ok=True) + temporary = Path(tempfile.mkdtemp(prefix=f".{output_dir.name}.partial-", dir=output_dir.parent)) + try: + for relative, data in sorted(payloads.items()): + destination = temporary.joinpath(*PurePosixPath(relative).parts) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(data) + if output_dir.exists(): + output_dir.rmdir() + temporary.replace(output_dir) + except Exception: + shutil.rmtree(temporary, ignore_errors=True) + raise + manifest = json.loads(payloads["materials-manifest.json"]) + return { + "result": manifest["result"], + "profile": manifest["profile"], + "materials_manifest_sha256": _sha256_bytes(payloads["materials-manifest.json"]), + "output_dir": str(output_dir), + } + + +def verify(input_manifest: Path, materials_dir: Path, expected_profile: str) -> dict[str, Any]: + if expected_profile not in PROFILES: + raise MaterialsError(f"unsupported expected profile: {expected_profile}") + expected = _compile(input_manifest) + root = materials_dir.absolute() + if not root.exists() or _is_linklike(root) or not root.is_dir(): + raise MaterialsError("materials root must be a real directory") + actual_paths: list[str] = [] + for path in root.rglob("*"): + if path.is_dir(): + if _is_linklike(path): + raise MaterialsError(f"symlink/junction directory in materials: {path}") + continue + _assert_regular_no_links(path, root, "material output") + actual_paths.append(PurePosixPath(*path.relative_to(root).parts).as_posix()) + if set(actual_paths) != set(expected): + raise MaterialsError( + f"material file set mismatch: missing={sorted(set(expected) - set(actual_paths))}, " + f"extra={sorted(set(actual_paths) - set(expected))}" + ) + for relative, expected_bytes in expected.items(): + actual = root.joinpath(*PurePosixPath(relative).parts).read_bytes() + if actual != expected_bytes: + raise MaterialsError(f"material content mismatch: {relative}") + manifest = json.loads(expected["materials-manifest.json"]) + if manifest["profile"] != expected_profile: + raise MaterialsError( + f"profile mismatch: expected {expected_profile}, materials are {manifest['profile']}" + ) + return { + "schema_version": VERIFY_SCHEMA, + "result": "PASS_INDEPENDENT_RECOMPUTATION", + "profile": expected_profile, + "candidate_commit": manifest["candidate"]["commit"], + "materials_manifest_sha256": _sha256_bytes(expected["materials-manifest.json"]), + "file_count": len(expected), + "occurrence_count": manifest["summary"]["occurrence_count"], + "unresolved_occurrence_count": 0, + "distribution_clearance_claimed": False, + } + + +def _write_receipt(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + raise MaterialsError(f"receipt already exists: {path}") + path.write_bytes(_pretty_json(payload)) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + build_parser = subparsers.add_parser("build", help="build deterministic offline materials") + build_parser.add_argument("--input-manifest", type=Path, required=True) + build_parser.add_argument("--output-dir", type=Path, required=True) + verify_parser = subparsers.add_parser("verify", help="independently recompute and verify materials") + verify_parser.add_argument("--input-manifest", type=Path, required=True) + verify_parser.add_argument("--materials-dir", type=Path, required=True) + verify_parser.add_argument("--expected-profile", choices=sorted(PROFILES), required=True) + verify_parser.add_argument("--receipt", type=Path) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = _parser().parse_args(argv) + try: + if args.command == "build": + result = build(args.input_manifest, args.output_dir) + else: + result = verify(args.input_manifest, args.materials_dir, args.expected_profile) + if args.receipt: + _write_receipt(args.receipt, result) + sys.stdout.buffer.write(_pretty_json(result)) + return 0 + except MaterialsError as exc: + sys.stderr.write(f"FAILED_CLOSED: {exc}\n") + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yuqei-infra/tests/test_third_party_materials.py b/yuqei-infra/tests/test_third_party_materials.py new file mode 100644 index 00000000..932aed8e --- /dev/null +++ b/yuqei-infra/tests/test_third_party_materials.py @@ -0,0 +1,482 @@ +from __future__ import annotations + +import importlib.util +import io +import json +import os +import tarfile +from pathlib import Path + +import pytest + + +SCRIPT = Path(__file__).parents[1] / "scripts" / "build-third-party-materials.py" +SPEC = importlib.util.spec_from_file_location("build_third_party_materials", SCRIPT) +assert SPEC and SPEC.loader +tool = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(tool) + + +ROLES = ( + "contract-web", + "contract-api", + "ai-platform-api", + "ai-worker", + "postgres-runtime", + "postgres-client", + "redis", + "minio", + "minio-client", + "nginx-edge", + "backup-utility", +) + + +def _write_json(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") + + +def _record(path: Path, root: Path) -> dict[str, object]: + digest, size = tool._hash_file(path) + return { + "path": path.relative_to(root).as_posix(), + "sha256": digest, + "size_bytes": size, + } + + +def _source_tar(path: Path, *, bad_kind: str | None = None) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with tarfile.open(path, "w:gz") as archive: + if bad_kind == "traversal": + member = tarfile.TarInfo("../escape.txt") + data = b"escape" + elif bad_kind == "symlink": + member = tarfile.TarInfo("source/link") + member.type = tarfile.SYMTYPE + member.linkname = "../target" + archive.addfile(member) + return + elif bad_kind == "hardlink": + target = tarfile.TarInfo("source/target.txt") + data = b"target" + target.size = len(data) + archive.addfile(target, io.BytesIO(data)) + member = tarfile.TarInfo("source/hardlink.txt") + member.type = tarfile.LNKTYPE + member.linkname = "source/target.txt" + archive.addfile(member) + return + else: + member = tarfile.TarInfo("source/README.txt") + data = b"corresponding source fixture\n" + member.size = len(data) + archive.addfile(member, io.BytesIO(data)) + + +def _sbom(role: str, manifest_digest: str) -> dict[str, object]: + return { + "spdxVersion": "SPDX-2.3", + "SPDXID": "SPDXRef-DOCUMENT", + "name": f"{role}.oci.tar", + "documentNamespace": f"https://example.invalid/sbom/{role}", + "creationInfo": {"creators": ["Tool: syft-1.52.0"], "created": "2026-09-27T00:00:00Z"}, + "packages": [ + { + "name": f"{role}.oci.tar", + "SPDXID": f"SPDXRef-Root-{role}", + "versionInfo": manifest_digest, + "licenseDeclared": "NOASSERTION", + }, + { + "name": f"fixture-{role}", + "SPDXID": f"SPDXRef-Package-{role}", + "versionInfo": "1.0.0", + "licenseDeclared": "MIT", + }, + ], + } + + +def _trivy(role: str, config_digest: str, *, bad_license: dict[str, str] | None = None) -> dict[str, object]: + licenses: list[dict[str, object]] = [ + { + "Severity": "LOW", + "Category": "notice", + "PkgName": f"fixture-{role}", + "FilePath": f"usr/share/doc/{role}/copyright", + "Name": "MIT", + "Text": "", + "Confidence": 1, + "Link": "https://spdx.org/licenses/MIT.html", + } + ] + if role == "contract-api": + licenses.append( + { + "Severity": "HIGH", + "Category": "restricted", + "PkgName": "fixture-copyleft", + "FilePath": "usr/share/doc/fixture-copyleft/copyright", + "Name": "GPL-2.0-only", + "Text": "", + "Confidence": 1, + "Link": "https://spdx.org/licenses/GPL-2.0-only.html", + } + ) + if bad_license: + licenses[0].update(bad_license) + return { + "SchemaVersion": 2, + "ArtifactName": f"{role}.docker.tar", + "ArtifactType": "container_image", + "Metadata": {"ImageID": config_digest, "OS": {"Family": "fixture", "Name": "1"}}, + "Results": [{"Target": "OS Packages", "Class": "license", "Licenses": licenses}], + } + + +def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path, dict[str, object]]: + root = tmp_path / "input" + root.mkdir(parents=True) + mit = root / "texts" / "MIT.txt" + gpl = root / "texts" / "GPL-2.0-only.txt" + mit.parent.mkdir() + mit.write_text("MIT fixture license text\n", encoding="utf-8") + gpl.write_text("GPL-2.0-only fixture license text\n", encoding="utf-8") + source = root / "sources" / "fixture-copyleft.tar.gz" + _source_tar(source) + + images: list[dict[str, object]] = [] + occurrences: list[dict[str, object]] = [] + for index, role in enumerate(ROLES, start=1): + manifest_digest = f"sha256:{index:064x}" + config_digest = f"sha256:{index + 10:064x}" + sbom_path = root / "reports" / f"{role}.spdx.json" + trivy_path = root / "reports" / f"{role}.trivy-license.json" + trivy_payload = _trivy(role, config_digest) + _write_json(sbom_path, _sbom(role, manifest_digest)) + _write_json(trivy_path, trivy_payload) + trivy_record = _record(trivy_path, root) + parsed, _ = tool._parse_trivy(trivy_payload, role, config_digest, str(trivy_record["sha256"])) + occurrences.extend(parsed) + images.append( + { + "role": role, + "manifest_digest": manifest_digest, + "config_digest": config_digest, + "sbom": _record(sbom_path, root), + "trivy_license": trivy_record, + } + ) + + coverage = [] + for occurrence in occurrences: + restricted = occurrence["category"] == "restricted" + coverage.append( + { + "occurrence_id": occurrence["occurrence_id"], + "license_text_ids": ["GPL-2.0-only" if restricted else "MIT"], + "source_material_ids": ["fixture-copyleft-source"] if restricted else [], + } + ) + manifest: dict[str, object] = { + "schema_version": tool.INPUT_SCHEMA, + "profile": profile, + "as_of": "2026-09-27T00:00:00+00:00", + "candidate": { + "commit": "a" * 40, + "tag": "v-test-third-party-materials", + "source_manifest_sha256": "b" * 64, + "images": images, + }, + "license_texts": [ + {"id": "MIT", "kind": "license", "file": _record(mit, root)}, + {"id": "GPL-2.0-only", "kind": "license", "file": _record(gpl, root)}, + ], + "source_materials": [ + { + "id": "fixture-copyleft-source", + "kind": "source-archive", + "file": _record(source, root), + } + ], + "coverage": coverage, + } + manifest_path = root / "input-manifest.json" + _write_json(manifest_path, manifest) + return manifest_path, manifest + + +def _rewrite_manifest(path: Path, manifest: dict[str, object]) -> None: + _write_json(path, manifest) + + +def _role_image(manifest: dict[str, object], role: str) -> dict[str, object]: + candidate = manifest["candidate"] + assert isinstance(candidate, dict) + images = candidate["images"] + assert isinstance(images, list) + return next(image for image in images if image["role"] == role) + + +def _update_file_record(root: Path, record: dict[str, object]) -> None: + path = root / str(record["path"]) + record.update(_record(path, root)) + + +def test_build_verify_and_determinism(tmp_path: Path) -> None: + manifest_path, _ = _fixture(tmp_path) + output_a = tmp_path / "materials-a" + output_b = tmp_path / "materials-b" + + built_a = tool.build(manifest_path, output_a) + built_b = tool.build(manifest_path, output_b) + verified = tool.verify(manifest_path, output_a, "redistributable") + + assert built_a["result"] == "PASS_TECHNICAL_MATERIALS_COMPLETE" + assert built_a["materials_manifest_sha256"] == built_b["materials_manifest_sha256"] + assert verified["result"] == "PASS_INDEPENDENT_RECOMPUTATION" + assert verified["occurrence_count"] == 12 + assert (output_a / "THIRD_PARTY_NOTICES.md").read_bytes() == ( + output_b / "THIRD_PARTY_NOTICES.md" + ).read_bytes() + matrix = json.loads((output_a / "occurrence-coverage.json").read_text(encoding="utf-8")) + assert matrix["summary"] == { + "covered_occurrence_count": 12, + "license_text_count": 2, + "occurrence_count": 12, + "restricted_occurrence_count": 1, + "source_material_count": 1, + "unresolved_occurrence_count": 0, + } + assert json.loads((output_a / "materials-manifest.json").read_text())["distribution_clearance_claimed"] is False + output_manifest = json.loads((output_a / "materials-manifest.json").read_text(encoding="utf-8")) + recorded_payload_hash = output_manifest.pop("manifest_payload_sha256") + assert recorded_payload_hash == tool._sha256_bytes(tool._canonical_json(output_manifest)) + + +def test_all_eleven_delivery_roles_are_mandatory(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + manifest["candidate"]["images"].pop() + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="exactly eleven"): + tool.build(manifest_path, tmp_path / "materials") + + +@pytest.mark.parametrize( + "raw,expected", + [ + ('{"schema_version":"one","schema_version":"two"}\n', "duplicate JSON key"), + ('{"schema_version":NaN}\n', "non-finite JSON value"), + ('{"schema_version":Infinity}\n', "non-finite JSON value"), + ], +) +def test_duplicate_json_keys_and_nonfinite_values_are_rejected( + tmp_path: Path, raw: str, expected: str +) -> None: + manifest_path = tmp_path / "input-manifest.json" + manifest_path.write_text(raw, encoding="utf-8") + with pytest.raises(tool.MaterialsError, match=expected): + tool.build(manifest_path, tmp_path / "materials") + + +def test_internal_and_redistributable_profiles_cannot_masquerade(tmp_path: Path) -> None: + manifest_path, _ = _fixture(tmp_path) + output = tmp_path / "materials" + tool.build(manifest_path, output) + + with pytest.raises(tool.MaterialsError, match="profile mismatch"): + tool.verify(manifest_path, output, "internal-hosted") + + +@pytest.mark.parametrize( + "change", + [ + {"Name": "LicenseRef-fixture"}, + {"Name": "MIT AND LicenseRef-fixture"}, + {"Name": "UNKNOWN"}, + {"Category": "unknown"}, + {"Severity": "UNKNOWN"}, + {"Name": "noderivs"}, + ], +) +def test_unknown_unresolved_and_licenseref_fail_closed(tmp_path: Path, change: dict[str, str]) -> None: + manifest_path, manifest = _fixture(tmp_path) + image = _role_image(manifest, "contract-web") + record = image["trivy_license"] + assert isinstance(record, dict) + report = manifest_path.parent / str(record["path"]) + payload = json.loads(report.read_text(encoding="utf-8")) + payload["Results"][0]["Licenses"][0].update(change) + _write_json(report, payload) + _update_file_record(manifest_path.parent, record) + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="unresolved/unknown/LicenseRef"): + tool.build(manifest_path, tmp_path / "materials") + + +@pytest.mark.parametrize("mode", ["missing", "duplicate", "extra"]) +def test_coverage_must_be_an_exact_nonduplicated_set(tmp_path: Path, mode: str) -> None: + manifest_path, manifest = _fixture(tmp_path) + coverage = manifest["coverage"] + assert isinstance(coverage, list) + if mode == "missing": + coverage.pop() + expected = "uncovered Trivy occurrences" + elif mode == "duplicate": + coverage.append(dict(coverage[0])) + expected = "duplicate coverage record" + else: + coverage.append( + {"occurrence_id": "occ-" + "f" * 64, "license_text_ids": ["MIT"], "source_material_ids": []} + ) + expected = "extra coverage record" + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match=expected): + tool.build(manifest_path, tmp_path / "materials") + + +def test_hash_and_size_drift_fail_closed(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + image = _role_image(manifest, "contract-web") + record = image["sbom"] + assert isinstance(record, dict) + record["sha256"] = "0" * 64 + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="size/hash mismatch"): + tool.build(manifest_path, tmp_path / "materials") + + +@pytest.mark.parametrize("kind", ["sbom", "trivy"]) +def test_sbom_and_trivy_image_binding_drift_fail_closed(tmp_path: Path, kind: str) -> None: + manifest_path, manifest = _fixture(tmp_path) + image = _role_image(manifest, "contract-web") + key = "sbom" if kind == "sbom" else "trivy_license" + record = image[key] + assert isinstance(record, dict) + report = manifest_path.parent / str(record["path"]) + payload = json.loads(report.read_text(encoding="utf-8")) + if kind == "sbom": + payload["packages"][0]["versionInfo"] = "sha256:" + "f" * 64 + expected = "bind exactly one root package" + else: + payload["Metadata"]["ImageID"] = "sha256:" + "f" * 64 + expected = "ImageID/config digest mismatch" + _write_json(report, payload) + _update_file_record(manifest_path.parent, record) + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match=expected): + tool.build(manifest_path, tmp_path / "materials") + + +def test_relative_path_escape_and_hardlink_are_rejected(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + license_record = manifest["license_texts"][0]["file"] + license_record["path"] = "../outside.txt" + _rewrite_manifest(manifest_path, manifest) + with pytest.raises(tool.MaterialsError, match="unsafe"): + tool.build(manifest_path, tmp_path / "escape-materials") + + manifest_path, manifest = _fixture(tmp_path / "hardlink") + root = manifest_path.parent + original = root / str(manifest["license_texts"][0]["file"]["path"]) + linked = root / "texts" / "MIT-hardlink.txt" + os.link(original, linked) + manifest["license_texts"][0]["file"] = _record(linked, root) + _rewrite_manifest(manifest_path, manifest) + with pytest.raises(tool.MaterialsError, match="hardlink is forbidden"): + tool.build(manifest_path, tmp_path / "hardlink-materials") + + +@pytest.mark.parametrize("bad_kind", ["traversal", "symlink", "hardlink"]) +def test_unsafe_source_archive_members_are_rejected(tmp_path: Path, bad_kind: str) -> None: + manifest_path, manifest = _fixture(tmp_path) + source_record = manifest["source_materials"][0]["file"] + source_path = manifest_path.parent / str(source_record["path"]) + _source_tar(source_path, bad_kind=bad_kind) + _update_file_record(manifest_path.parent, source_record) + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="unsafe|archive member|symlink/hardlink"): + tool.build(manifest_path, tmp_path / "materials") + + +def test_restricted_occurrence_requires_source_and_unused_artifacts_fail(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + coverage = manifest["coverage"] + assert isinstance(coverage, list) + restricted = next(item for item in coverage if item["source_material_ids"]) + restricted["source_material_ids"] = [] + _rewrite_manifest(manifest_path, manifest) + with pytest.raises(tool.MaterialsError, match="restricted occurrence lacks"): + tool.build(manifest_path, tmp_path / "materials-no-source") + + manifest_path, manifest = _fixture(tmp_path / "extra") + manifest["license_texts"].append(dict(manifest["license_texts"][0], id="UNUSED")) + _rewrite_manifest(manifest_path, manifest) + with pytest.raises(tool.MaterialsError, match="duplicate license input path|unused/extra"): + tool.build(manifest_path, tmp_path / "materials-extra") + + +def test_valid_written_offer_is_bound_and_internal_profile_rejects_it(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + root = manifest_path.parent + restricted = next(item for item in manifest["coverage"] if item["source_material_ids"]) + offer_id = "approved-source-offer" + offer = { + "schema_version": tool.OFFER_SCHEMA, + "offer_id": offer_id, + "status": "APPROVED", + "profile": "redistributable", + "candidate_commit": manifest["candidate"]["commit"], + "source_manifest_sha256": manifest["candidate"]["source_manifest_sha256"], + "image_manifest_digests": { + image["role"]: image["manifest_digest"] for image in manifest["candidate"]["images"] + }, + "issuer": { + "legal_name": "Example Fixture Organization", + "approver_name": "Fixture Approver", + "approver_title": "Release Owner", + "contact": "source-offer@example.invalid", + }, + "authorization_id": "FIXTURE-OFFER-001", + "approval_provenance_sha256": "c" * 64, + "valid_from": "2026-09-01T00:00:00Z", + "valid_until": "2027-09-01T00:00:00Z", + "covers": [restricted["occurrence_id"]], + } + offer_path = root / "offers" / "offer.json" + _write_json(offer_path, offer) + manifest["source_materials"] = [ + {"id": offer_id, "kind": "written-source-offer", "file": _record(offer_path, root)} + ] + restricted["source_material_ids"] = [offer_id] + _rewrite_manifest(manifest_path, manifest) + + tool.build(manifest_path, tmp_path / "offer-materials") + + manifest["profile"] = "internal-hosted" + _rewrite_manifest(manifest_path, manifest) + with pytest.raises(tool.MaterialsError, match="forbidden for internal-hosted"): + tool.build(manifest_path, tmp_path / "internal-materials") + + +def test_verifier_rejects_extra_or_modified_output(tmp_path: Path) -> None: + manifest_path, _ = _fixture(tmp_path) + output = tmp_path / "materials" + tool.build(manifest_path, output) + (output / "unexpected.txt").write_text("extra\n", encoding="utf-8") + with pytest.raises(tool.MaterialsError, match="file set mismatch"): + tool.verify(manifest_path, output, "redistributable") + + (output / "unexpected.txt").unlink() + notices = output / "THIRD_PARTY_NOTICES.md" + notices.write_text(notices.read_text(encoding="utf-8") + "tampered\n", encoding="utf-8") + with pytest.raises(tool.MaterialsError, match="content mismatch"): + tool.verify(manifest_path, output, "redistributable")