build: validate third-party release materials
This commit is contained in:
@@ -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())
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user