diff --git a/yuqei-infra/scripts/build-third-party-materials.py b/yuqei-infra/scripts/build-third-party-materials.py index 081db449..4729fd4f 100644 --- a/yuqei-infra/scripts/build-third-party-materials.py +++ b/yuqei-infra/scripts/build-third-party-materials.py @@ -29,8 +29,8 @@ 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" +SOURCE_ARCHIVE_SCHEMA = "yuqei.corresponding-source-archive.v1" PROFILES = {"internal-hosted", "redistributable"} EXPECTED_ROLES = { "contract-web", @@ -54,6 +54,8 @@ MAX_LICENSE_TEXT = 8 * 1024 * 1024 MAX_ARCHIVE_FILES = 200_000 MAX_ARCHIVE_EXPANDED = 8 * 1024 * 1024 * 1024 MAX_COMPRESSION_RATIO = 2_000 +SOURCE_MANIFEST_NAME = "YUQEI-SOURCE-MANIFEST.json" +SPDX_TOKEN = re.compile(r"[A-Za-z0-9][A-Za-z0-9.+-]*") class MaterialsError(RuntimeError): @@ -198,7 +200,7 @@ def _file_record(root: Path, raw: Any, label: str) -> tuple[dict[str, Any], Path return {"path": relative, "sha256": actual_hash, "size_bytes": actual_size}, path -def _load_json(path: Path, label: str) -> Mapping[str, Any]: +def _loads_json_strict(raw: str, 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: @@ -209,17 +211,25 @@ def _load_json(path: Path, label: str) -> Mapping[str, Any]: try: value = json.loads( - path.read_text(encoding="utf-8"), + raw, 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 + except json.JSONDecodeError as exc: + raise MaterialsError(f"invalid JSON for {label}") from exc return _mapping(value, label) +def _load_json(path: Path, label: str) -> Mapping[str, Any]: + try: + raw = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + raise MaterialsError(f"invalid JSON for {label}: {path}") from exc + return _loads_json_strict(raw, label) + + def _parse_time(value: Any, label: str) -> datetime: text = _text(value, label) if text.endswith("Z"): @@ -233,10 +243,174 @@ def _parse_time(value: Any, label: str) -> datetime: return parsed.astimezone(timezone.utc) -def _validate_archive(path: Path, label: str) -> dict[str, int]: +def _spdx_tokens(value: Any, label: str) -> list[str]: + expression = _text(value, label) + tokens: list[str] = [] + position = 0 + while position < len(expression): + while position < len(expression) and expression[position].isspace(): + position += 1 + if position == len(expression): + break + if expression[position] in "()": + tokens.append(expression[position]) + position += 1 + continue + match = SPDX_TOKEN.match(expression, position) + if not match: + raise MaterialsError(f"invalid SPDX expression token in {label} at offset {position}") + tokens.append(match.group(0)) + position = match.end() + if not tokens: + raise MaterialsError(f"empty SPDX expression for {label}") + return tokens + + +def _normalize_spdx_ast(ast: tuple[Any, ...]) -> tuple[Any, ...]: + operator = ast[0] + if operator not in {"and", "or"}: + return ast + children: list[tuple[Any, ...]] = [] + for child in ast[1]: + normalized = _normalize_spdx_ast(child) + if normalized[0] == operator: + children.extend(normalized[1]) + else: + children.append(normalized) + return (operator, tuple(sorted(children, key=repr))) + + +def _parse_spdx_expression( + value: Any, + label: str, + license_ids: set[str], + exception_ids: set[str], +) -> tuple[Any, ...]: + tokens = _spdx_tokens(value, label) + position = 0 + + def primary() -> tuple[Any, ...]: + nonlocal position + if position >= len(tokens): + raise MaterialsError(f"incomplete SPDX expression for {label}") + token = tokens[position] + if token == "(": + position += 1 + result = parse_or() + if position >= len(tokens) or tokens[position] != ")": + raise MaterialsError(f"unbalanced SPDX expression for {label}") + position += 1 + return result + if token in {"AND", "OR", "WITH", ")"} or token not in license_ids: + raise MaterialsError(f"unknown or non-canonical SPDX license ID in {label}: {token}") + position += 1 + return ("id", token) + + def parse_with() -> tuple[Any, ...]: + nonlocal position + result = primary() + if position < len(tokens) and tokens[position] == "WITH": + if result[0] != "id": + raise MaterialsError(f"SPDX WITH must follow one license ID in {label}") + position += 1 + if position >= len(tokens) or tokens[position] not in exception_ids: + bad = tokens[position] if position < len(tokens) else "" + raise MaterialsError(f"unknown SPDX exception in {label}: {bad}") + result = ("with", result, tokens[position]) + position += 1 + return result + + def parse_and() -> tuple[Any, ...]: + nonlocal position + children = [parse_with()] + while position < len(tokens) and tokens[position] == "AND": + position += 1 + children.append(parse_with()) + return children[0] if len(children) == 1 else ("and", tuple(children)) + + def parse_or() -> tuple[Any, ...]: + nonlocal position + children = [parse_and()] + while position < len(tokens) and tokens[position] == "OR": + position += 1 + children.append(parse_and()) + return children[0] if len(children) == 1 else ("or", tuple(children)) + + parsed = parse_or() + if position != len(tokens): + raise MaterialsError(f"unexpected SPDX token in {label}: {tokens[position]}") + return _normalize_spdx_ast(parsed) + + +def _spdx_terms(ast: tuple[Any, ...]) -> set[str]: + if ast[0] in {"id", "with"}: + return {_spdx_canonical(ast)} + terms: set[str] = set() + for child in ast[1]: + terms.update(_spdx_terms(child)) + return terms + + +def _spdx_canonical(ast: tuple[Any, ...], parent_precedence: int = 0) -> str: + operator = ast[0] + if operator == "id": + return str(ast[1]) + if operator == "with": + return f"{_spdx_canonical(ast[1], 3)} WITH {ast[2]}" + precedence = 2 if operator == "and" else 1 + separator = " AND " if operator == "and" else " OR " + rendered = separator.join(_spdx_canonical(child, precedence) for child in ast[1]) + return f"({rendered})" if precedence < parent_precedence else rendered + + +def _load_spdx_catalog( + root: Path, raw: Any +) -> tuple[set[str], set[str], list[dict[str, Any]]]: + catalog = _mapping(raw, "spdx_catalog") + if set(catalog) != {"licenses", "exceptions"}: + raise MaterialsError("spdx_catalog must contain exactly licenses and exceptions") + license_record, license_path = _file_record(root, catalog["licenses"], "SPDX license catalog") + exception_record, exception_path = _file_record(root, catalog["exceptions"], "SPDX exception catalog") + licenses_payload = _load_json(license_path, "SPDX license catalog") + exceptions_payload = _load_json(exception_path, "SPDX exception catalog") + if licenses_payload.get("licenseListVersion") != exceptions_payload.get("licenseListVersion"): + raise MaterialsError("SPDX license and exception catalog versions differ") + license_ids = [ + _text(item.get("licenseId"), "SPDX license ID") + for item in ( + _mapping(value, "SPDX license") + for value in _list(licenses_payload.get("licenses"), "SPDX licenses") + ) + if not item.get("isDeprecatedLicenseId", False) + ] + exception_ids = [ + _text(item.get("licenseExceptionId"), "SPDX exception ID") + for item in ( + _mapping(value, "SPDX exception") + for value in _list(exceptions_payload.get("exceptions"), "SPDX exceptions") + ) + if not item.get("isDeprecatedLicenseId", False) + ] + _unique(license_ids, "SPDX license ID") + _unique(exception_ids, "SPDX exception ID") + if not license_ids: + raise MaterialsError("SPDX license catalog is empty") + for identifier in [*license_ids, *exception_ids]: + if not SPDX_TOKEN.fullmatch(identifier) or "LicenseRef-" in identifier: + raise MaterialsError(f"invalid SPDX catalog identifier: {identifier}") + records = [ + {"kind": "spdx-license-catalog", **license_record}, + {"kind": "spdx-exception-catalog", **exception_record}, + ] + return set(license_ids), set(exception_ids), records + + +def _inspect_archive(path: Path, label: str) -> dict[str, Any]: file_count = 0 expanded = 0 names: set[str] = set() + members: dict[str, dict[str, Any]] = {} + source_manifest_bytes: bytes | None = None def accept_name(name: str) -> str: safe = _safe_relative(name.rstrip("/"), f"{label} member") @@ -249,7 +423,7 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]: if zipfile.is_zipfile(path): with zipfile.ZipFile(path) as archive: for member in archive.infolist(): - accept_name(member.filename) + safe_name = 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}") @@ -263,10 +437,24 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]: raise MaterialsError(f"excessive compression ratio in {label}: {member.filename}") file_count += 1 expanded += member.file_size + digest = hashlib.sha256() + chunks: list[bytes] = [] + with archive.open(member) as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + if safe_name == SOURCE_MANIFEST_NAME: + chunks.append(chunk) + members[safe_name] = { + "path": safe_name, + "sha256": digest.hexdigest(), + "size_bytes": member.file_size, + } + if safe_name == SOURCE_MANIFEST_NAME: + source_manifest_bytes = b"".join(chunks) else: with tarfile.open(path, "r:*") as archive: for member in archive: - accept_name(member.name) + safe_name = accept_name(member.name) if member.issym() or member.islnk(): raise MaterialsError(f"symlink/hardlink in {label}: {member.name}") if member.isdir(): @@ -275,6 +463,22 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]: raise MaterialsError(f"non-regular member in {label}: {member.name}") file_count += 1 expanded += member.size + handle = archive.extractfile(member) + if handle is None: + raise MaterialsError(f"unreadable regular member in {label}: {member.name}") + digest = hashlib.sha256() + chunks = [] + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + if safe_name == SOURCE_MANIFEST_NAME: + chunks.append(chunk) + members[safe_name] = { + "path": safe_name, + "sha256": digest.hexdigest(), + "size_bytes": member.size, + } + if safe_name == SOURCE_MANIFEST_NAME: + source_manifest_bytes = b"".join(chunks) except (tarfile.TarError, zipfile.BadZipFile, OSError) as exc: raise MaterialsError(f"invalid source archive for {label}: {path}") from exc if file_count < 1: @@ -284,7 +488,12 @@ def _validate_archive(path: Path, label: str) -> dict[str, int]: 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} + return { + "file_count": file_count, + "expanded_size_bytes": expanded, + "members": members, + "source_manifest_bytes": source_manifest_bytes, + } def _validate_sbom(payload: Mapping[str, Any], role: str, manifest_digest: str) -> dict[str, Any]: @@ -295,19 +504,46 @@ def _validate_sbom(payload: Mapping[str, Any], role: str, manifest_digest: str) raise MaterialsError(f"{role} SBOM package inventory is empty") ids: list[str] = [] roots = 0 + components: list[dict[str, Any]] = [] 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 + continue + external_refs = package.get("externalRefs", []) + if external_refs is None: + external_refs = [] + purls = sorted( + { + _text(reference.get("referenceLocator"), f"{role} SBOM purl") + for reference in ( + _mapping(value, f"{role} SBOM external reference") + for value in _list(external_refs, f"{role} SBOM external references") + ) + if reference.get("referenceType") == "purl" + } + ) + components.append( + { + "name": _text(package.get("name"), f"{role} SBOM package name"), + "version": _text(package.get("versionInfo"), f"{role} SBOM package version"), + "spdx_id": spdx_id, + "purls": purls, + } + ) _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} + return { + "package_count": len(packages), + "root_manifest_digest": manifest_digest, + "components": components, + } def _license_is_unresolved(item: Mapping[str, Any]) -> bool: @@ -378,41 +614,158 @@ def _parse_trivy( return occurrences, {"occurrence_count": len(occurrences), "image_id": config_digest} -def _offer_details( +def _validate_source_archive( path: Path, artifact_id: str, candidate: Mapping[str, Any], image_manifest_digests: Mapping[str, str], - profile: str, - as_of: datetime, + occurrences: Mapping[str, Mapping[str, Any]], + sbom_components: Mapping[str, Mapping[str, list[Mapping[str, Any]]]], ) -> 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} + inspected = _inspect_archive(path, f"source material {artifact_id}") + raw_manifest = inspected.pop("source_manifest_bytes") + members = inspected.pop("members") + if raw_manifest is None: + raise MaterialsError( + f"source material {artifact_id} lacks root {SOURCE_MANIFEST_NAME}" + ) + try: + manifest_text = raw_manifest.decode("utf-8") + except UnicodeDecodeError as exc: + raise MaterialsError(f"source material {artifact_id} has invalid embedded manifest") from exc + manifest = _loads_json_strict( + manifest_text, f"source material {artifact_id} embedded manifest" + ) + expected_keys = { + "schema_version", + "source_id", + "candidate_commit", + "source_manifest_sha256", + "package_name", + "package_version", + "binary_coordinate", + "source_coordinate", + "image_manifest_digests", + "covers", + "files", + } + if set(manifest) != expected_keys: + raise MaterialsError( + f"source material {artifact_id} embedded manifest keys must be exactly {sorted(expected_keys)}" + ) + if manifest.get("schema_version") != SOURCE_ARCHIVE_SCHEMA or manifest.get("source_id") != artifact_id: + raise MaterialsError(f"source material {artifact_id} embedded identity mismatch") + if manifest.get("candidate_commit") != candidate["commit"]: + raise MaterialsError(f"source material {artifact_id} candidate mismatch") + if manifest.get("source_manifest_sha256") != candidate["source_manifest_sha256"]: + raise MaterialsError(f"source material {artifact_id} source manifest mismatch") + package_name = _text(manifest.get("package_name"), f"source material {artifact_id} package_name") + package_version = _text(manifest.get("package_version"), f"source material {artifact_id} package_version") + binary_coordinate = _text( + manifest.get("binary_coordinate"), f"source material {artifact_id} binary_coordinate" + ) + source_coordinate = _text( + manifest.get("source_coordinate"), f"source material {artifact_id} source_coordinate" + ) + if source_coordinate.upper() in {"UNKNOWN", "NOASSERTION"} or not re.match( + r"^(?:pkg:|https://|git\+https://)", source_coordinate + ): + raise MaterialsError(f"source material {artifact_id} has an invalid source coordinate") + covers = [ + _text(item, f"source material {artifact_id} covers item") + for item in _list(manifest.get("covers"), f"source material {artifact_id} covers") + ] + if not covers: + raise MaterialsError(f"source material {artifact_id} covers no occurrences") + _unique(covers, f"source material {artifact_id} covers") + covered_roles: set[str] = set() + for occurrence_id in covers: + occurrence = occurrences.get(occurrence_id) + if occurrence is None: + raise MaterialsError(f"source material {artifact_id} covers unknown occurrence {occurrence_id}") + if str(occurrence["category"]).strip().lower() != "restricted": + raise MaterialsError(f"source material {artifact_id} covers non-restricted occurrence {occurrence_id}") + if occurrence["package_name"] != package_name: + raise MaterialsError( + f"source material {artifact_id} package/Trivy mismatch for {occurrence_id}" + ) + role = str(occurrence["role"]) + covered_roles.add(role) + candidates = sbom_components.get(role, {}).get(package_name, []) + identities = { + (str(item["version"]), purl) + for item in candidates + for purl in item.get("purls", []) + if item.get("version") not in {"UNKNOWN", "NOASSERTION", ""} + } + if identities != {(package_version, binary_coordinate)}: + raise MaterialsError( + f"source material {artifact_id} package version/coordinate does not exactly bind " + f"{role}/{package_name}: observed={sorted(identities)}" + ) + expected_image_binding = { + role: image_manifest_digests[role] for role in sorted(covered_roles) + } + if manifest.get("image_manifest_digests") != expected_image_binding: + raise MaterialsError(f"source material {artifact_id} image binding mismatch") + + declared_files: dict[str, dict[str, Any]] = {} + has_substantive_source = False + for index, raw in enumerate(_list(manifest.get("files"), f"source material {artifact_id} files")): + record = _mapping(raw, f"source material {artifact_id} file {index}") + if set(record) != {"path", "sha256", "size_bytes", "kind"}: + raise MaterialsError(f"source material {artifact_id} file records require path/hash/size/kind") + relative = _safe_relative(record.get("path"), f"source material {artifact_id} file path") + if relative == SOURCE_MANIFEST_NAME or relative in declared_files: + raise MaterialsError(f"duplicate/reserved source file in {artifact_id}: {relative}") + digest = _text(record.get("sha256"), f"source material {artifact_id} file hash") + size = record.get("size_bytes") + kind = _text(record.get("kind"), f"source material {artifact_id} file kind") + if not HEX64.fullmatch(digest) or not isinstance(size, int) or isinstance(size, bool) or size < 1: + raise MaterialsError(f"invalid embedded source file record in {artifact_id}: {relative}") + if kind not in {"source", "build", "patch", "license", "metadata", "source-archive"}: + raise MaterialsError(f"invalid embedded source file kind in {artifact_id}: {kind}") + basename = PurePosixPath(relative).name.lower() + documentation_only = basename.startswith( + ("readme", "license", "copying", "notice", "authors", "changelog") + ) + if kind == "source" and not documentation_only: + has_substantive_source = True + declared_files[relative] = { + "path": relative, + "sha256": digest, + "size_bytes": size, + "kind": kind, + } + actual_files = {key: value for key, value in members.items() if key != SOURCE_MANIFEST_NAME} + if set(declared_files) != set(actual_files): + raise MaterialsError( + f"source material {artifact_id} embedded file set mismatch: " + f"missing={sorted(set(declared_files) - set(actual_files))}, " + f"extra={sorted(set(actual_files) - set(declared_files))}" + ) + for relative, declared in declared_files.items(): + actual = actual_files[relative] + if (declared["sha256"], declared["size_bytes"]) != ( + actual["sha256"], + actual["size_bytes"], + ): + raise MaterialsError(f"source material {artifact_id} embedded file hash/size mismatch: {relative}") + if not has_substantive_source: + raise MaterialsError( + f"source material {artifact_id} contains documentation only, not corresponding source" + ) + return { + **inspected, + "package_name": package_name, + "package_version": package_version, + "binary_coordinate": binary_coordinate, + "source_coordinate": source_coordinate, + "image_manifest_digests": expected_image_binding, + "covers": sorted(covers), + "files": [declared_files[key] for key in sorted(declared_files)], + "embedded_manifest_sha256": _sha256_bytes(raw_manifest), + } def _compile(input_manifest: Path) -> dict[str, bytes]: @@ -423,7 +776,16 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: 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"} + expected_top = { + "schema_version", + "profile", + "as_of", + "candidate", + "spdx_catalog", + "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") @@ -452,6 +814,11 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: all_occurrences: list[dict[str, Any]] = [] image_summaries: list[dict[str, Any]] = [] input_records: list[dict[str, Any]] = [] + spdx_license_ids, spdx_exception_ids, spdx_records = _load_spdx_catalog( + root, raw_manifest.get("spdx_catalog") + ) + input_records.extend(spdx_records) + sbom_components_by_role: dict[str, dict[str, list[Mapping[str, Any]]]] = {} roles: list[str] = [] manifest_digests: list[str] = [] config_digests: list[str] = [] @@ -471,6 +838,10 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: 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) + components_by_name: dict[str, list[Mapping[str, Any]]] = {} + for component in sbom_summary["components"]: + components_by_name.setdefault(component["name"], []).append(component) + sbom_components_by_role[role] = components_by_name occurrences, trivy_summary = _parse_trivy( _load_json(trivy_path, f"{role} Trivy report"), role, config_digest, trivy_record["sha256"] ) @@ -505,18 +876,54 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: occurrence_by_id = {item["occurrence_id"]: item for item in all_occurrences} license_defs: dict[str, dict[str, Any]] = {} + license_ast_by_id: dict[str, tuple[Any, ...]] = {} + license_aliases: dict[str, tuple[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") + if set(record) != {"id", "kind", "spdx_expression", "aliases", "file"}: + raise MaterialsError( + "license text records require exactly id, kind, spdx_expression, aliases 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}") + expression_ast = _parse_spdx_expression( + record.get("spdx_expression"), + f"license text {artifact_id} SPDX expression", + spdx_license_ids, + spdx_exception_ids, + ) + aliases = [ + _text(value, f"license text {artifact_id} alias") + for value in _list(record.get("aliases"), f"license text {artifact_id} aliases") + ] + _unique(aliases, f"license text {artifact_id} aliases") + for alias in aliases: + lowered = alias.lower() + if any(marker in lowered for marker in ("licenseref-", "unknown", "unresolved", "noassertion")): + raise MaterialsError(f"unsafe SPDX alias for license text {artifact_id}: {alias}") + previous = license_aliases.get(alias) + if previous is not None and previous != expression_ast: + raise MaterialsError(f"SPDX alias maps to conflicting expressions: {alias}") + try: + parsed_alias = _parse_spdx_expression( + alias, + f"license text {artifact_id} alias", + spdx_license_ids, + spdx_exception_ids, + ) + except MaterialsError: + parsed_alias = None + if parsed_alias is not None and parsed_alias != expression_ast: + raise MaterialsError( + f"canonical SPDX alias has conflicting meaning for license text {artifact_id}: {alias}" + ) + license_aliases[alias] = expression_ast 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}") @@ -532,9 +939,12 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: license_defs[artifact_id] = { "id": artifact_id, "kind": kind, + "spdx_expression": _spdx_canonical(expression_ast), + "aliases": sorted(aliases), "input": file_record, "output_path": output_path, } + license_ast_by_id[artifact_id] = expression_ast license_payloads[output_path] = payload if not license_defs: raise MaterialsError("at least one license text is required") @@ -556,14 +966,20 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: 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 + if kind == "written-source-offer": + raise MaterialsError( + "written source offers are disabled until a pinned Ed25519 root/keyring and " + "authorized-signature policy are implemented; provide a real source archive" ) - output_path = f"source-offers/{artifact_id}--{file_record['sha256'][:16]}.json" + details = _validate_source_archive( + path, + artifact_id, + candidate, + image_manifest_digest_map, + occurrence_by_id, + sbom_components_by_role, + ) + output_path = f"corresponding-source/{artifact_id}--{file_record['sha256'][:16]}.source" source_paths.append(file_record["path"]) source_defs[artifact_id] = { "id": artifact_id, @@ -583,8 +999,16 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: 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") + if set(coverage) != { + "occurrence_id", + "spdx_expression", + "license_text_ids", + "source_material_ids", + }: + raise MaterialsError( + "coverage records require exactly occurrence_id, spdx_expression, " + "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}") @@ -604,18 +1028,66 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: occurrence = occurrence_by_id.get(occurrence_id) if occurrence is None: raise MaterialsError(f"extra coverage record: {occurrence_id}") + raw_detected_expression = str(occurrence["license_name"]).strip() + try: + detected_ast = _parse_spdx_expression( + raw_detected_expression, + f"Trivy occurrence {occurrence_id} license_name", + spdx_license_ids, + spdx_exception_ids, + ) + except MaterialsError as exc: + detected_ast = license_aliases.get(raw_detected_expression) + if detected_ast is None: + raise MaterialsError( + f"Trivy occurrence {occurrence_id} has no exact canonical SPDX or declared alias mapping: " + f"{raw_detected_expression}" + ) from exc + coverage_ast = _parse_spdx_expression( + coverage.get("spdx_expression"), + f"coverage {occurrence_id} SPDX expression", + spdx_license_ids, + spdx_exception_ids, + ) + if coverage_ast != detected_ast: + raise MaterialsError( + f"coverage {occurrence_id} SPDX expression does not exactly match Trivy AND/OR semantics" + ) + selected_license_asts = [ + license_ast_by_id[item] + for item in license_ids + if license_defs[item]["kind"] == "license" + ] + required_terms = _spdx_terms(detected_ast) + selected_terms: set[str] = set() + for selected_ast in selected_license_asts: + terms = _spdx_terms(selected_ast) + if selected_ast != detected_ast and selected_ast[0] not in {"id", "with"}: + raise MaterialsError( + f"coverage {occurrence_id} uses a license text with different SPDX AND/OR semantics" + ) + if not terms.issubset(required_terms): + raise MaterialsError( + f"coverage {occurrence_id} uses a license text for a different SPDX license or exception" + ) + selected_terms.update(terms) + if selected_terms != required_terms: + raise MaterialsError( + f"coverage {occurrence_id} license texts do not exactly cover its SPDX license terms" + ) 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}") + if occurrence_id not in source["details"]["covers"]: + raise MaterialsError(f"source archive {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, + "spdx_expression": _spdx_canonical(detected_ast), "license_text_ids": sorted(license_ids), "source_material_ids": sorted(source_ids), } @@ -629,8 +1101,6 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: 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() @@ -639,7 +1109,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: declared = set(source["details"]["covers"]) if declared != actually_covered: raise MaterialsError( - f"source offer {source_id} coverage mismatch: " + f"source archive {source_id} coverage mismatch: " f"missing={sorted(actually_covered - declared)}, extra={sorted(declared - actually_covered)}" ) @@ -690,6 +1160,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: f"- Image role: `{occurrence['role']}`", f"- Component: `{package}`", f"- Detected license: `{occurrence['license_name']}`", + f"- Canonical SPDX expression: `{coverage['spdx_expression']}`", 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'}`", @@ -742,7 +1213,7 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: "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"])), + "inputs": sorted(input_records, key=lambda item: (item.get("role", ""), item["kind"])), "input_manifest": { "sha256": input_manifest_hash, "size_bytes": input_manifest_size, @@ -750,6 +1221,10 @@ def _compile(input_manifest: Path) -> dict[str, bytes]: "summary": matrix["summary"], "output_files": output_records, "result": "PASS_TECHNICAL_MATERIALS_COMPLETE", + "manifest_payload_hash_contract": ( + "SHA-256 over UTF-8 canonical JSON with sorted keys, compact separators and trailing LF; " + "manifest_payload_sha256 excluded" + ), "distribution_clearance_claimed": False, "claims_not_made": [ "legal opinion", diff --git a/yuqei-infra/tests/test_third_party_materials.py b/yuqei-infra/tests/test_third_party_materials.py index 932aed8e..95a0843a 100644 --- a/yuqei-infra/tests/test_third_party_materials.py +++ b/yuqei-infra/tests/test_third_party_materials.py @@ -46,9 +46,24 @@ def _record(path: Path, root: Path) -> dict[str, object]: } -def _source_tar(path: Path, *, bad_kind: str | None = None) -> None: +def _source_tar( + path: Path, + *, + source_manifest: dict[str, object] | None = None, + files: dict[str, bytes] | None = None, + bad_kind: str | None = None, +) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tarfile.open(path, "w:gz") as archive: + if source_manifest is not None: + manifest_data = tool._canonical_json(source_manifest) + manifest_member = tarfile.TarInfo(tool.SOURCE_MANIFEST_NAME) + manifest_member.size = len(manifest_data) + archive.addfile(manifest_member, io.BytesIO(manifest_data)) + for name, data in sorted((files or {}).items()): + member = tarfile.TarInfo(name) + member.size = len(data) + archive.addfile(member, io.BytesIO(data)) if bad_kind == "traversal": member = tarfile.TarInfo("../escape.txt") data = b"escape" @@ -69,6 +84,8 @@ def _source_tar(path: Path, *, bad_kind: str | None = None) -> None: archive.addfile(member) return else: + if source_manifest is not None: + return member = tarfile.TarInfo("source/README.txt") data = b"corresponding source fixture\n" member.size = len(data) @@ -76,26 +93,50 @@ def _source_tar(path: Path, *, bad_kind: str | None = None) -> None: def _sbom(role: str, manifest_digest: str) -> dict[str, object]: + 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", + "externalRefs": [ + { + "referenceCategory": "PACKAGE-MANAGER", + "referenceType": "purl", + "referenceLocator": f"pkg:generic/fixture-{role}@1.0.0", + } + ], + }, + ] + if role == "contract-api": + packages.append( + { + "name": "fixture-copyleft", + "SPDXID": "SPDXRef-Package-fixture-copyleft", + "versionInfo": "1.0.0", + "licenseDeclared": "GPL-2.0-only", + "externalRefs": [ + { + "referenceCategory": "PACKAGE-MANAGER", + "referenceType": "purl", + "referenceLocator": "pkg:generic/fixture-copyleft@1.0.0", + } + ], + } + ) 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", - }, - ], + "packages": packages, } @@ -144,8 +185,24 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path, 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) + + spdx_licenses = root / "spdx" / "licenses.json" + spdx_exceptions = root / "spdx" / "exceptions.json" + _write_json( + spdx_licenses, + { + "licenseListVersion": "3.29.0", + "licenses": [ + {"licenseId": "Apache-2.0", "isDeprecatedLicenseId": False}, + {"licenseId": "GPL-2.0-only", "isDeprecatedLicenseId": False}, + {"licenseId": "MIT", "isDeprecatedLicenseId": False}, + ], + }, + ) + _write_json( + spdx_exceptions, + {"licenseListVersion": "3.29.0", "exceptions": []}, + ) images: list[dict[str, object]] = [] occurrences: list[dict[str, object]] = [] @@ -176,10 +233,35 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path, coverage.append( { "occurrence_id": occurrence["occurrence_id"], + "spdx_expression": occurrence["license_name"], "license_text_ids": ["GPL-2.0-only" if restricted else "MIT"], "source_material_ids": ["fixture-copyleft-source"] if restricted else [], } ) + restricted_occurrence = next(item for item in occurrences if item["category"] == "restricted") + contract_api_image = next(item for item in images if item["role"] == "contract-api") + source_bytes = b"def copyleft_fixture() -> str:\n return 'source'\n" + source_file_record = { + "path": "src/main.py", + "sha256": tool._sha256_bytes(source_bytes), + "size_bytes": len(source_bytes), + "kind": "source", + } + source_manifest = { + "schema_version": tool.SOURCE_ARCHIVE_SCHEMA, + "source_id": "fixture-copyleft-source", + "candidate_commit": "a" * 40, + "source_manifest_sha256": "b" * 64, + "package_name": "fixture-copyleft", + "package_version": "1.0.0", + "binary_coordinate": "pkg:generic/fixture-copyleft@1.0.0", + "source_coordinate": "pkg:generic/fixture-copyleft-source@1.0.0", + "image_manifest_digests": {"contract-api": contract_api_image["manifest_digest"]}, + "covers": [restricted_occurrence["occurrence_id"]], + "files": [source_file_record], + } + source = root / "sources" / "fixture-copyleft.tar.gz" + _source_tar(source, source_manifest=source_manifest, files={"src/main.py": source_bytes}) manifest: dict[str, object] = { "schema_version": tool.INPUT_SCHEMA, "profile": profile, @@ -190,9 +272,25 @@ def _fixture(tmp_path: Path, *, profile: str = "redistributable") -> tuple[Path, "source_manifest_sha256": "b" * 64, "images": images, }, + "spdx_catalog": { + "licenses": _record(spdx_licenses, root), + "exceptions": _record(spdx_exceptions, root), + }, "license_texts": [ - {"id": "MIT", "kind": "license", "file": _record(mit, root)}, - {"id": "GPL-2.0-only", "kind": "license", "file": _record(gpl, root)}, + { + "id": "MIT", + "kind": "license", + "spdx_expression": "MIT", + "aliases": [], + "file": _record(mit, root), + }, + { + "id": "GPL-2.0-only", + "kind": "license", + "spdx_expression": "GPL-2.0-only", + "aliases": [], + "file": _record(gpl, root), + }, ], "source_materials": [ { @@ -331,7 +429,12 @@ def test_coverage_must_be_an_exact_nonduplicated_set(tmp_path: Path, mode: str) expected = "duplicate coverage record" else: coverage.append( - {"occurrence_id": "occ-" + "f" * 64, "license_text_ids": ["MIT"], "source_material_ids": []} + { + "occurrence_id": "occ-" + "f" * 64, + "spdx_expression": "MIT", + "license_text_ids": ["MIT"], + "source_material_ids": [], + } ) expected = "extra coverage record" _rewrite_manifest(manifest_path, manifest) @@ -424,32 +527,85 @@ def test_restricted_occurrence_requires_source_and_unused_artifacts_fail(tmp_pat tool.build(manifest_path, tmp_path / "materials-extra") -def test_valid_written_offer_is_bound_and_internal_profile_rejects_it(tmp_path: Path) -> None: +def test_mit_text_cannot_masquerade_as_gpl_coverage(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + restricted = next(item for item in manifest["coverage"] if item["source_material_ids"]) + restricted["license_text_ids"] = ["MIT"] + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="different SPDX license|do not exactly cover"): + tool.build(manifest_path, tmp_path / "materials") + + +def test_license_text_and_or_semantics_cannot_masquerade(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + manifest["license_texts"][0]["spdx_expression"] = "MIT OR Apache-2.0" + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="different SPDX AND/OR semantics"): + tool.build(manifest_path, tmp_path / "materials") + + +def test_readme_only_archive_cannot_masquerade_as_corresponding_source(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + root = manifest_path.parent + source_record = manifest["source_materials"][0]["file"] + source_path = root / str(source_record["path"]) + with tarfile.open(source_path, "r:gz") as archive: + member = archive.getmember(tool.SOURCE_MANIFEST_NAME) + extracted = archive.extractfile(member) + assert extracted is not None + source_manifest = json.loads(extracted.read().decode("utf-8")) + readme = b"This is not corresponding source.\n" + source_manifest["files"] = [ + { + "path": "README.md", + "sha256": tool._sha256_bytes(readme), + "size_bytes": len(readme), + "kind": "source", + } + ] + _source_tar(source_path, source_manifest=source_manifest, files={"README.md": readme}) + _update_file_record(root, source_record) + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="documentation only"): + tool.build(manifest_path, tmp_path / "materials") + + +def test_source_archive_package_version_must_match_sbom_identity(tmp_path: Path) -> None: + manifest_path, manifest = _fixture(tmp_path) + root = manifest_path.parent + source_record = manifest["source_materials"][0]["file"] + source_path = root / str(source_record["path"]) + with tarfile.open(source_path, "r:gz") as archive: + embedded = archive.extractfile(archive.getmember(tool.SOURCE_MANIFEST_NAME)) + source_file = archive.extractfile(archive.getmember("src/main.py")) + assert embedded is not None and source_file is not None + source_manifest = json.loads(embedded.read().decode("utf-8")) + source_bytes = source_file.read() + source_manifest["package_version"] = "9.9.9" + _source_tar(source_path, source_manifest=source_manifest, files={"src/main.py": source_bytes}) + _update_file_record(root, source_record) + _rewrite_manifest(manifest_path, manifest) + + with pytest.raises(tool.MaterialsError, match="version/coordinate does not exactly bind"): + tool.build(manifest_path, tmp_path / "materials") + + +def test_self_asserted_written_offer_fails_without_pinned_ed25519_trust(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_id = "self-signed-source-offer" offer = { - "schema_version": tool.OFFER_SCHEMA, + "schema_version": "yuqei.written-source-offer.v1", "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"]], + "public_key": "attacker-controlled-key", + "signature": "attacker-controlled-signature", } offer_path = root / "offers" / "offer.json" _write_json(offer_path, offer) @@ -459,12 +615,8 @@ def test_valid_written_offer_is_bound_and_internal_profile_rejects_it(tmp_path: 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") + with pytest.raises(tool.MaterialsError, match="pinned Ed25519 root/keyring"): + tool.build(manifest_path, tmp_path / "offer-materials") def test_verifier_rejects_extra_or_modified_output(tmp_path: Path) -> None: