"""Bind the collected runtime corresponding-source archive and notices into the release package manifest and distribution. ``tools/collect_runtime_sources.py`` collects the actual source of the third-party runtime that the shipped C64 PRGs link, plus the license and notice text, into a byte-reproducible tarball and a JSON report. This tool is the packaging side of that work: it verifies the collected archive against the collector's exact tree evidence, binds the result to a digest and provenance record, and stages the tree the release source and binary archives reference. It fails closed on every axis: * The archive is read only through ``tools/package_release.py``'s bounded, no-follow archive reader (``open_bounded_tar`` / ``audit_tar``); this tool carries no second tar parser. Member payloads are read in chunks the shared reader's single-read bound accepts. * The archive is bound to the collector report's configured SHA-256, size, member count and tree digest. * Coverage is required and mapped, never inferred from the presence of an archive. Each required component is bound to the collector's **exact tree evidence**: the selected members under its archive prefix must hash to the report's pinned per-input ``tree_digest``, its declared ``required_paths`` must all be present, its dependencies must carry a locked version, and its notices and external artifacts must match the report's recorded SHA-256 and size byte for byte -- presence or a member count is never enough. * The built-in coverage is a **trusted minimum**. ``--coverage`` may only add requirements; a supplied mapping that omits a required component, path, notice, artifact, license, commit or dependency disposition is rejected. * Every obligation the collector records must be accounted for by the mapping; an unaccounted obligation is a hard failure, not a report line. * Staged files are created exclusively and atomically. ``--force`` is removed, so an existing file or a hardlink to an unrelated file is never truncated. * Nothing is fetched and nothing is published. All work is local file I/O. Source completeness and release readiness are separate: ``source_complete`` is ``true`` only when the mapped coverage is satisfied and reviewed, while ``release_ready`` stays ``false`` because human legal approval and the clean extracted-source rebuild are still outstanding. A public license file alone never marks completeness. ``evaluate_bundle_from_archived_evidence`` is the reusable entry point for the release checker; it takes the current report, archive and coverage metadata and returns the same record the CLI emits. ``tools/package_release.py`` accepts the same two integration options as this tool: ``--runtime-sources-report PATH`` and ``--runtime-sources-dir PATH``. They attach the verified runtime-source tree and notices to the release without modifying the original collector evidence or silently approving publication. Exit codes: 0 success, 1 a check failed or a policy blocked the run, 2 usage error, 3 internal error. """ from __future__ import annotations import argparse import json import os import sys import time from dataclasses import asdict, replace from pathlib import Path from typing import Any, Mapping, Optional, Sequence TOOLS_DIR = Path(__file__).resolve().parent if str(TOOLS_DIR) not in sys.path: sys.path.insert(0, str(TOOLS_DIR)) import package_release as pr # noqa: E402 (shared bounded archive reader and policy) SCHEMA = 1 RECORD_KIND = "release_runtime_source_bundle" COVERAGE_RECORD_KIND = "release_runtime_source_coverage" REPORT_RECORD_KIND = "release_runtime_sources" ARCHIVE_NAME = "runtime-corresponding-sources.tar.gz" INTERNAL_MANIFEST = "RUNTIME-SOURCES-MANIFEST.txt" NOTICE_MEMBER_DIR = "notices" BUNDLE_RECORD_NAME = "RUNTIME-SOURCE-BUNDLE.json" FRAGMENT_NAME = "SOURCE-AND-LICENSES.runtime.md" STAGED_SOURCE_DIR = "runtime-sources" STAGED_SOURCE_NOTICE_DIR = "NOTICES" STAGED_BINARY_NOTICE_DIR = "runtime-notices" EXIT_OK = 0 EXIT_FAILED = 1 EXIT_USAGE = 2 EXIT_INTERNAL = 3 REPO_ROOT = Path(__file__).resolve().parent.parent DEFAULT_REPORT = REPO_ROOT / "target/release-runtime-sources/runtime-sources-report.json" DEFAULT_COLLECTED = REPO_ROOT / "target/release-runtime-sources/collected" DEFAULT_SOURCE_MANIFEST = REPO_ROOT / "tools/release_source_manifest.json" COLLECTOR_TOOL = "tools/collect_runtime_sources.py" SELF_TOOL = "tools/runtime_source_bundle.py" MAX_REPORT_BYTES = 8 << 20 MAX_NOTICE_BYTES = 8 << 20 MAX_STAGED_FILE_BYTES = 1 << 30 # A dependency is only satisfied by a disposition at least this strong. A # stronger disposition may replace a weaker one in a supplied mapping; the # reverse is a weakening and is rejected. DISPOSITION_STRENGTH = { "covered": 3, "notice_covered": 2, "external_prerequisite": 1, "not_distributed": 1, "unresolved": 0, } LICENSE_FINDING_KINDS = ("license_mismatch", "no_obligation", "unaccounted_obligation") # --------------------------------------------------------------------------- # trusted minimum coverage # # This is the trusted floor the bundle must satisfy. Each component is bound to # one collector source input by id, to the input's archive prefix, and to the # exact ``required_paths`` the collector must have verified. ``--coverage`` can # only add requirements to this floor, never remove or relax one. The license, # commit, notices, artifacts and dependency dispositions are the hardened # review's required values. # --------------------------------------------------------------------------- def _component( key: str, component: str, *, repository: str, commit: Optional[str], scope: str, license: str, license_accepts: Sequence[str], source_input: Optional[str], archive_prefix: str, required_paths: Sequence[str], required_notices: Sequence[str], required_collected_artifacts: Sequence[str] = (), dependencies: Sequence[dict] = (), min_members: int = 0, obligation_match: Optional[str] = None, note: Optional[str] = None, ) -> dict: entry = { "key": key, "component": component, "repository": repository, "commit": commit, "scope": scope, "license": license, "license_accepts": list(license_accepts), "source_input": source_input, "archive_prefix": archive_prefix, "min_members": min_members, "required_paths": list(required_paths), "required_source_prefixes": [archive_prefix] if archive_prefix else [], "required_notices": list(required_notices), "required_collected_artifacts": list(required_collected_artifacts), "dependencies": [dict(dependency) for dependency in dependencies], } if obligation_match: entry["obligation_match"] = obligation_match if note: entry["note"] = note return entry REQUIRED_COVERAGE: dict = { "schema": SCHEMA, "record_kind": COVERAGE_RECORD_KIND, "reviewed": True, "maintainer": "runtime-source hardening review", "legal_review": "pending", "role": ( "the trusted minimum runtime source, notice and dependency coverage the bundle must satisfy; a " "supplied mapping may only strengthen it" ), "components": [ _component( "rust-mos-runtime-library", "rust-mos fork Rust runtime libraries (core, alloc, compiler_builtins, panic_abort, ...)", repository="https://github.com/koalabuttz/rust-mos", commit="4aaa40e160fcabfcc5b0c30740a518084f00ab5d", scope="linked", license="Apache-2.0 OR MIT", license_accepts=[ "Apache-2.0 OR MIT", "Apache-2.0 OR MIT, with the Rust project's COPYRIGHT attribution", ], source_input="rust-mos-library", archive_prefix="rust-mos/library/", min_members=2200, required_paths=[ "Cargo.lock", "core/src/lib.rs", "alloc/src/lib.rs", "panic_abort/src/lib.rs", "compiler-builtins/compiler-builtins/src/lib.rs", "backtrace/src/lib.rs", ], required_notices=[ "notices/rust-mos-fork/COPYRIGHT", "notices/rust-mos-fork/LICENSE-APACHE", "notices/rust-mos-fork/LICENSE-MIT", ], dependencies=[ { "name": "compiler_builtins", "requires_source": True, "disposition": "covered", "locked_version": "0.1.160", "note": "the submodule tree is archived under rust-mos/library/compiler-builtins/", }, { "name": "backtrace", "requires_source": True, "disposition": "covered", "note": "the submodule tree is archived under rust-mos/library/backtrace/", }, ], obligation_match="rust-mos C64 runtime", ), _component( "rust-mos-toolchain-image", "rust-mos C64 toolchain image ghcr.io/koalabuttz/rust-mos:ac2fb2277-4537158-4aaa40e16", repository="https://github.com/koalabuttz/rust-mos", commit="4aaa40e160fcabfcc5b0c30740a518084f00ab5d", scope="prerequisite", license="Apache-2.0 OR MIT", license_accepts=["Apache-2.0 OR MIT"], source_input="rust-mos-library", archive_prefix="rust-mos/library/", min_members=2200, required_paths=["Cargo.lock"], required_notices=[], required_collected_artifacts=[ "rust-mos/targets/mos-c64-none.json", "rust-mos/rust-src/Cargo.lock", "probe/rust-mos/rustc-vV.txt", "probe/rust-mos/mos-clang-version.txt", ], obligation_match="rust-mos C64 toolchain image", ), _component( "llvm-mos-sdk", "llvm-mos-sdk platform (CRT, C library, linker scripts, C64 target files)", repository="https://github.com/llvm-mos/llvm-mos-sdk", commit="45371581201110552a51b3b1de2aae1a0b0af89c", scope="linked", license="Apache-2.0 WITH LLVM-exception", license_accepts=["Apache-2.0 WITH LLVM-exception"], source_input="llvm-mos-sdk-tree", archive_prefix="llvm-mos-sdk/", min_members=800, required_paths=[ "LICENSE", "CMakeLists.txt", "mos-platform/common/CMakeLists.txt", "mos-platform/common/c/string.c", "mos-platform/common/crt0/crt0.S", "mos-platform/common/ldscripts/imag-regs.ld", "mos-platform/c64/CMakeLists.txt", "mos-platform/c64/link.ld", ], required_notices=["notices/llvm-mos-sdk/LICENSE"], required_collected_artifacts=["mos-sysroot/common/imag-regs.ld", "mos-sysroot/c64/link.ld"], dependencies=[ { "name": "cc65 (shared platform files)", "requires_source": False, "disposition": "notice_covered", "note": ( "the llvm-mos-sdk LICENSE carries an appended cc65 license section for the files " "the SDK and cc65 share; cc65's compiler and runtime are not linked" ), } ], obligation_match="MOS sysroot", ), _component( "llvm-mos-compiler", "llvm-mos compiler (Clang/LLVM MOS backend)", repository="https://github.com/llvm-mos/llvm-mos", commit="ac2fb2277e6903a1f687714f5e5047c8caeae7c6", scope="notice_only", license="Apache-2.0 WITH LLVM-exception", license_accepts=["Apache-2.0 WITH LLVM-exception"], source_input=None, archive_prefix="", required_paths=[], required_notices=["notices/llvm-mos/LICENSE.TXT"], obligation_match="Clang/LLVM MOS backend", ), _component( "rust-std-1.93.0", "Rust standard library (core, std) for x86_64-unknown-linux-gnu, official stable 1.93.0", repository="https://github.com/rust-lang/rust", commit="254b59607d4417e9dffbc307138ae5c86280fe4c", scope="linked", license="Apache-2.0 OR MIT", license_accepts=[ "Apache-2.0 OR MIT", "Apache-2.0 OR MIT, with the Rust project's COPYRIGHT attribution", ], source_input="rust-1.93.0-library", archive_prefix="rust-1.93.0/library/", min_members=2100, required_paths=[ "library/Cargo.lock", "library/core/src/lib.rs", "library/alloc/src/lib.rs", "library/panic_abort/src/lib.rs", "library/compiler-builtins/compiler-builtins/src/lib.rs", ], required_notices=[ "notices/rust-1.93.0-src/COPYRIGHT", "notices/rust-1.93.0-src/LICENSE-APACHE", "notices/rust-1.93.0-src/LICENSE-MIT", ], required_collected_artifacts=[ "rust-1.93.0/COPYRIGHT-library.html", "rust-1.93.0/licenses/Apache-2.0.txt", "rust-1.93.0/licenses/MIT.txt", ], dependencies=[ { "name": "compiler_builtins", "requires_source": True, "disposition": "covered", "locked_version": "0.1.160", "note": "the submodule tree is archived under rust-1.93.0/library/compiler-builtins/", } ], obligation_match="Rust standard library", ), ], } OBLIGATION_MATCH = {spec["key"]: spec["obligation_match"] for spec in REQUIRED_COVERAGE["components"] if spec.get("obligation_match")} class BundleError(Exception): """An operator-facing check or policy failure. Nothing partial is kept.""" class UsageError(Exception): """Invalid invocation.""" # --------------------------------------------------------------------------- # small helpers # --------------------------------------------------------------------------- def sha256_hex(data: bytes) -> str: return pr.sha256_bytes(data) def utc_now() -> str: epoch = os.environ.get("SOURCE_DATE_EPOCH") if epoch: try: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(epoch))) except ValueError as error: raise UsageError(f"SOURCE_DATE_EPOCH is not an integer: {epoch}") from error return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def display_path(path: Path, repo_root: Path = REPO_ROOT) -> str: """A repo-relative or bare name; never an absolute maintainer path.""" try: return str(path.resolve().relative_to(repo_root.resolve())) except (ValueError, OSError): return path.name def _read_bounded(path: Path, *, max_bytes: int, description: str) -> bytes: try: return pr.read_regular_bytes(path, max_bytes=max_bytes, description=description) except pr.CheckFailure as error: raise BundleError(str(error)) from error def _load_json(path: Path, *, max_bytes: int, description: str) -> tuple[Any, str]: data = _read_bounded(path, max_bytes=max_bytes, description=description) try: payload = json.loads(data) except json.JSONDecodeError as error: raise BundleError(f"{path} is not valid JSON: {error}") from error return payload, sha256_hex(data) def _read_member_bytes(extracted, limits: pr.Limits) -> bytes: """Read a member in chunks the shared bounded reader accepts.""" chunk_size = max(1, getattr(limits, "max_single_read", 64 << 10)) chunks: list[bytes] = [] while True: chunk = extracted.read(chunk_size) if not chunk: break chunks.append(chunk) return b"".join(chunks) # --------------------------------------------------------------------------- # archive audit through the shared bounded reader # --------------------------------------------------------------------------- def tree_digest(member_hashes: Mapping[str, str]) -> str: """Aggregate digest over (path, content-hash) pairs; order independent.""" lines = [f"{member_hashes[name]} {name}" for name in sorted(member_hashes)] return sha256_hex("\n".join(lines).encode("utf-8")) def internal_manifest_lines(member_hashes: Mapping[str, str]) -> bytes: return "".join(f"{member_hashes[name]} {name}\n" for name in sorted(member_hashes)).encode("utf-8") def is_notice_member(name: str) -> bool: return name.startswith(NOTICE_MEMBER_DIR + "/") def verify_internal_manifest(manifest_bytes: bytes, member_hashes: Mapping[str, str]) -> None: """The archive's own member list must equal the recomputed member hashes.""" expected = internal_manifest_lines(member_hashes) if manifest_bytes == expected: return listed: dict[str, str] = {} for line in manifest_bytes.decode("utf-8", "replace").splitlines(): parts = line.split(" ", 1) if len(parts) == 2: listed[parts[1]] = parts[0] missing = sorted(set(member_hashes) - set(listed)) extra = sorted(set(listed) - set(member_hashes)) mismatched = sorted( name for name in set(listed) & set(member_hashes) if listed[name] != member_hashes[name] ) detail = [] if missing: detail.append(f"{len(missing)} member(s) unlisted, first {missing[0]!r}") if extra: detail.append(f"{len(extra)} listed path(s) absent, first {extra[0]!r}") if mismatched: detail.append(f"{len(mismatched)} hash mismatch(es), first {mismatched[0]!r}") raise BundleError( f"{INTERNAL_MANIFEST} does not match the recomputed member hashes" + (": " + "; ".join(detail) if detail else "") ) def audit_archive(path: Path, *, limits: pr.Limits = pr.DEFAULT_LIMITS) -> dict: """Verify the runtime source archive with the shared bounded reader.""" try: safety = pr.audit_tar(path, limits) except pr.CheckFailure as error: raise BundleError(str(error)) from error member_hashes: dict[str, str] = {} member_sizes: dict[str, int] = {} manifest_bytes: Optional[bytes] = None try: with pr.open_bounded_tar(path, limits) as tar: for member in tar: if not member.isreg(): continue extracted = tar.extractfile(member) if extracted is None: raise BundleError(f"member {member.name!r} could not be read") payload = _read_member_bytes(extracted, limits) if len(payload) != member.size: raise BundleError( f"member {member.name!r} read back {len(payload)} bytes, expected {member.size}" ) if member.name == INTERNAL_MANIFEST: manifest_bytes = payload continue member_hashes[member.name] = sha256_hex(payload) member_sizes[member.name] = len(payload) except pr.CheckFailure as error: raise BundleError(str(error)) from error if manifest_bytes is None: raise BundleError(f"the runtime source archive does not carry {INTERNAL_MANIFEST}") verify_internal_manifest(manifest_bytes, member_hashes) notice_names = sorted(name for name in member_hashes if is_notice_member(name)) return { "name": path.name, "sha256": safety["sha256"], "size_bytes": safety["size"], "members": len(member_hashes) + 1, "source_members": len(member_hashes), "tree_digest": tree_digest(member_hashes), "internal_manifest": INTERNAL_MANIFEST, "internal_manifest_sha256": sha256_hex(manifest_bytes), "internal_manifest_verified": True, "tar_safety": safety.get("tar_safety", "verified"), "notice_members": notice_names, "member_hashes": member_hashes, "member_sizes": member_sizes, } def extract_members(path: Path, names: Sequence[str], *, limits: pr.Limits = pr.DEFAULT_LIMITS) -> dict: """Read the named regular members out of the archive through the shared reader.""" wanted = set(names) found: dict[str, bytes] = {} try: with pr.open_bounded_tar(path, limits) as tar: for member in tar: if member.name not in wanted or not member.isreg(): continue extracted = tar.extractfile(member) if extracted is None: continue payload = _read_member_bytes(extracted, limits) if len(payload) != member.size: raise BundleError( f"member {member.name!r} read back {len(payload)} bytes, expected {member.size}" ) found[member.name] = payload except pr.CheckFailure as error: raise BundleError(str(error)) from error missing = sorted(wanted - set(found)) if missing: raise BundleError("the archive is missing expected members: " + ", ".join(missing)) return found # --------------------------------------------------------------------------- # report loading and binding # --------------------------------------------------------------------------- def load_report(path: Path) -> tuple[dict, str]: report, digest = _load_json(path, max_bytes=MAX_REPORT_BYTES, description="collector report") if not isinstance(report, dict): raise BundleError(f"{path} is not a JSON object") if report.get("schema") != SCHEMA: raise BundleError( f"the collector report schema {report.get('schema')!r} is not the supported schema {SCHEMA}" ) if report.get("record_kind") != REPORT_RECORD_KIND: raise BundleError( f"the collector report record_kind {report.get('record_kind')!r} is not {REPORT_RECORD_KIND!r}" ) return report, digest def archive_filename(report: dict) -> str: """The configured archive file name, refused if it is not a plain name.""" source_archive = report.get("source_archive") if not isinstance(source_archive, dict): raise BundleError("the collector report carries no source_archive record") name = source_archive.get("name") if not isinstance(name, str) or not name: raise BundleError("the collector report's source_archive has no name") if name != os.path.basename(name) or pr.unsafe_member_name(name): raise BundleError(f"the configured archive name is not a plain file name: {name!r}") return name def bind_report(report: dict, audit: dict) -> dict: """Cross-check the collector report's configured identity against the bytes.""" source_archive = report["source_archive"] if not source_archive.get("written"): raise BundleError( "the collector report records the source archive as not written; run the collector " "without --dry-run before wiring the archive into a package" ) problems: list[str] = [] if source_archive.get("name") != audit["name"]: problems.append( f"the report names {source_archive.get('name')!r} but the archive on disk is {audit['name']!r}" ) if source_archive.get("sha256") != audit["sha256"]: problems.append( "the archive SHA-256 does not match the digest the collector report configured " f"(report {source_archive.get('sha256')!r}, archive {audit['sha256']})" ) if source_archive.get("size_bytes") != audit["size_bytes"]: problems.append( f"the archive is {audit['size_bytes']} bytes, the report records {source_archive.get('size_bytes')!r}" ) if source_archive.get("members") != audit["members"]: problems.append( f"the archive has {audit['members']} members, the report records {source_archive.get('members')!r}" ) if source_archive.get("source_members") != audit["source_members"]: problems.append( f"the archive has {audit['source_members']} source members, the report records " f"{source_archive.get('source_members')!r}" ) if source_archive.get("tree_digest") != audit["tree_digest"]: problems.append( "the archive tree digest does not match the report " f"(report {source_archive.get('tree_digest')!r}, archive {audit['tree_digest']})" ) listed = source_archive.get("notices_included") if not isinstance(listed, list) or not listed: problems.append("the collector report records no included notices") else: missing = [name for name in listed if name not in audit["notice_members"]] if missing: problems.append("the report lists notices that are not members of the archive: " + ", ".join(missing)) if problems: raise BundleError( "the runtime source archive does not bind the collector report:\n " + "\n ".join(problems) ) return {"verified": True, "notices_listed": list(listed)} def report_input_index(report: dict) -> dict: index = {} for source_input in report.get("source_archive", {}).get("inputs", []): if isinstance(source_input, dict) and source_input.get("id"): index[source_input["id"]] = source_input return index def report_file_index(report: dict) -> dict: """Every collected/fetched file the report records, keyed by output_name.""" index: dict[str, dict] = {} for image in report.get("images", []): for record in (image or {}).get("files", []) or []: if isinstance(record, dict) and record.get("output_name"): index[record["output_name"]] = record for record in (image or {}).get("probes", []) or []: if isinstance(record, dict) and record.get("output_name"): index[record["output_name"]] = record for fetch in report.get("fetches", []): for record in (fetch or {}).get("files", []) or []: if isinstance(record, dict) and record.get("output_name"): index[record["output_name"]] = record return index # --------------------------------------------------------------------------- # coverage # --------------------------------------------------------------------------- def validate_coverage(coverage: Any) -> dict: if not isinstance(coverage, dict): raise BundleError("the coverage mapping must be a JSON object") if coverage.get("record_kind") != COVERAGE_RECORD_KIND: raise BundleError( f"the coverage mapping record_kind {coverage.get('record_kind')!r} is not {COVERAGE_RECORD_KIND!r}" ) if coverage.get("schema") != SCHEMA: raise BundleError( f"the coverage mapping schema {coverage.get('schema')!r} is not the supported schema {SCHEMA}" ) components = coverage.get("components") if not isinstance(components, list) or not components: raise BundleError("the coverage mapping names no components") seen: set[str] = set() for entry in components: if not isinstance(entry, dict) or not entry.get("key"): raise BundleError("every coverage component needs a string key") if entry["key"] in seen: raise BundleError(f"the coverage mapping repeats the component key {entry['key']!r}") seen.add(entry["key"]) for field in ( "required_source_prefixes", "required_notices", "required_collected_artifacts", "required_paths", "dependencies", ): if field in entry and not isinstance(entry[field], list): raise BundleError(f"coverage component {entry['key']!r} field {field!r} must be a list") return coverage def load_coverage(path: Path) -> tuple[dict, str]: coverage, digest = _load_json(path, max_bytes=MAX_REPORT_BYTES, description="coverage mapping") return validate_coverage(coverage), digest def validate_coverage_against_minimum(supplied: dict, minimum: dict = REQUIRED_COVERAGE) -> None: """A supplied mapping may only strengthen the trusted minimum. Any omitted component, required path, notice, artifact, license, commit or dependency disposition, and any weakening of a dependency disposition, is rejected. Adding components or requirements is allowed. """ problems: list[str] = [] if supplied.get("schema") != minimum.get("schema"): problems.append( f"schema {supplied.get('schema')!r} does not match the trusted minimum schema {minimum.get('schema')!r}" ) supplied_by_key = {entry["key"]: entry for entry in supplied.get("components", [])} def superset(minimum_values, supplied_values, label, key): supplied_values = supplied_values or [] missing = [value for value in minimum_values or [] if value not in supplied_values] if missing: problems.append(f"{key}: supplied coverage drops required {label}: {missing}") for spec in minimum.get("components", []): key = spec["key"] candidate = supplied_by_key.get(key) if candidate is None: problems.append(f"{key}: the supplied coverage omits a required component") continue for field in ("scope", "license", "commit", "source_input", "archive_prefix"): if field in spec and candidate.get(field) != spec.get(field): problems.append( f"{key}: supplied {field} {candidate.get(field)!r} is not the required {spec.get(field)!r}" ) if spec.get("min_members"): supplied_floor = candidate.get("min_members") if not isinstance(supplied_floor, int) or supplied_floor < spec["min_members"]: problems.append( f"{key}: supplied min_members {supplied_floor!r} is below the trusted floor {spec['min_members']}" ) superset(spec.get("required_source_prefixes"), candidate.get("required_source_prefixes"), "source prefixes", key) superset(spec.get("required_paths"), candidate.get("required_paths"), "paths", key) superset(spec.get("required_notices"), candidate.get("required_notices"), "notices", key) superset( spec.get("required_collected_artifacts"), candidate.get("required_collected_artifacts"), "collected artifacts", key, ) supplied_deps = {dep.get("name"): dep for dep in candidate.get("dependencies", []) or []} for dep in spec.get("dependencies", []) or []: name = dep.get("name") if name not in supplied_deps: problems.append(f"{key}: supplied coverage drops dependency {name!r}") continue supplied_dep = supplied_deps[name] if DISPOSITION_STRENGTH.get(supplied_dep.get("disposition"), 0) < DISPOSITION_STRENGTH.get( dep.get("disposition"), 0 ): problems.append( f"{key}: dependency {name!r} disposition {supplied_dep.get('disposition')!r} is weaker than " f"the required {dep.get('disposition')!r}" ) if dep.get("locked_version") and supplied_dep.get("locked_version") != dep.get("locked_version"): problems.append( f"{key}: dependency {name!r} locked_version {supplied_dep.get('locked_version')!r} is not " f"the required {dep.get('locked_version')!r}" ) if problems: raise BundleError( "the supplied coverage mapping weakens the trusted minimum:\n " + "\n ".join(problems) ) def member_for_required_path(archive_prefix: str, path: str, members: Mapping[str, Any]) -> Optional[str]: """Resolve a collector ``required_paths`` entry to an archive member. The collector reports paths relative to the extracted tree root; the archive may carry the tree under an arcname that already includes the tree's own leading directory, so the arcname tail is tried as a fallback. """ candidates = [] if archive_prefix and path: candidates.append(f"{archive_prefix}{path}") tail = archive_prefix.rstrip("/").split("/")[-1] if tail and path.startswith(tail + "/"): parent = archive_prefix.rstrip("/")[: -len(tail)] candidates.append(f"{parent}{path}") for candidate in candidates: if candidate in members: return candidate return None def component_selection(audit: dict, archive_prefix: str) -> dict: if not archive_prefix: return {} return {name: digest for name, digest in audit["member_hashes"].items() if name.startswith(archive_prefix)} def read_collected_artifacts( collected_dir: Path, names: Sequence[str], limits: pr.Limits, file_index: Mapping[str, dict] ) -> tuple[dict, list]: """Read named collected artifacts and compare them to the report's records.""" found: dict[str, dict] = {} missing: list[str] = [] for name in names: reason = pr.unsafe_member_name(name) if reason: missing.append(f"{name} (unsafe name: {reason})") continue try: data = _read_bounded( collected_dir / name, max_bytes=limits.max_file_bytes, description=f"collected artifact {name}" ) except BundleError: missing.append(name) continue digest = sha256_hex(data) record = file_index.get(name) if record is None: missing.append(f"{name} (the collector report records no artifact with this name)") continue if record.get("sha256") != digest: missing.append( f"{name} (sha256 {digest} does not match the report's recorded {record.get('sha256')!r})" ) continue recorded_size = record.get("size_bytes") if not isinstance(recorded_size, int): missing.append(f"{name} (the collector report records no size for this artifact)") continue if recorded_size != len(data): missing.append( f"{name} (size {len(data)} does not match the report's recorded {recorded_size})" ) continue if len(data) == 0: missing.append(f"{name} (the collected artifact is empty)") continue found[name] = {"sha256": digest, "size_bytes": len(data), "status": record.get("status")} return found, missing def evaluate_notice_mapping(audit: dict, report: dict) -> tuple[list, list]: """Bind each archive notice to the collector's fetched artifact byte for byte.""" file_index = report_file_index(report) mapping: list[dict] = [] problems: list[str] = [] for member in audit["notice_members"]: output_name = member[len(NOTICE_MEMBER_DIR) + 1:] observed = audit["member_sizes"].get(member) digest = audit["member_hashes"].get(member) record = file_index.get(output_name) if record is None: problems.append(f"{member}: the collector report records no fetched artifact for this notice") continue if not observed: problems.append(f"{member}: the archive notice is empty") continue if observed > MAX_NOTICE_BYTES: problems.append( f"{member}: the archive notice is {observed} bytes, above the {MAX_NOTICE_BYTES} byte notice bound" ) continue recorded_size = record.get("size_bytes") if not isinstance(recorded_size, int): problems.append(f"{member}: the fetched artifact records no size") continue if recorded_size != observed: problems.append( f"{member}: the archive notice is {observed} bytes but the fetched artifact records {recorded_size}" ) continue if record.get("sha256") != digest: problems.append( f"{member}: the archive notice hash {digest} does not match the fetched artifact " f"{record.get('sha256')!r}" ) continue mapping.append( { "member": member, "output_name": output_name, "sha256": digest, # The observed archive size is authoritative; the fetched size # is retained only as provenance. "size_bytes": observed, "fetched_size_bytes": recorded_size, "url": record.get("url"), "status": record.get("status"), } ) return mapping, problems def _pin_present(report: dict, spec: dict) -> bool: commit = spec.get("commit") repository = spec.get("repository") if not commit: return False for pin in report.get("pins", []): if not isinstance(pin, dict) or pin.get("commit") != commit: continue if repository is None or pin.get("repository") == repository: return True return False def check_locked_dependency(dependency: dict, report_dependencies: Sequence[dict]) -> list[str]: """Require the report's locked dependency to match by name *and* version. A missing entry, an entry with no locked version, an entry whose version differs from the required one, and duplicate entries that disagree on the version are each a gap; a crate name match alone is not enough. """ name = dependency.get("name") required_version = dependency.get("locked_version") entries = [entry for entry in report_dependencies if isinstance(entry, dict) and entry.get("crate") == name] if not entries: return [f"the collector report records no dependency entry for {name!r}"] problems: list[str] = [] versions = [entry.get("locked_version") for entry in entries] resolved = sorted({version for version in versions if version}) if len(resolved) > 1: problems.append(f"dependency {name!r} records conflicting locked versions {resolved}") if required_version: if any(version is None for version in versions): problems.append(f"dependency {name!r} is recorded without a locked version") elif resolved != [required_version]: problems.append( f"dependency {name!r} locked version {resolved} is not the required {required_version!r}" ) return problems def evaluate_coverage( coverage: dict, audit: dict, report: dict, collected_dir: Path, limits: pr.Limits = pr.DEFAULT_LIMITS, ) -> dict: """Require mapped, exact-evidence coverage. An archive alone is never enough.""" problems: list[str] = [] components: list[dict] = [] inputs = report_input_index(report) file_index = report_file_index(report) for spec in coverage["components"]: key = spec["key"] missing_source: list[str] = [] missing_notices: list[str] = [] missing_artifacts: list[str] = [] evidence: dict = {} scope = spec.get("scope") classified = scope in ("linked", "notice_only", "prerequisite", "not_distributed") if not classified: problems.append(f"{key}: scope {scope!r} is unclassified; the review must classify it") source_input = spec.get("source_input") archive_prefix = spec.get("archive_prefix") or "" if source_input: record = inputs.get(source_input) if record is None: missing_source.append(f"the collector report records no source input {source_input!r}") else: selection = component_selection(audit, archive_prefix) if not selection: missing_source.append(f"the archive holds no member under {archive_prefix!r}") pinned = record.get("tree_digest") if not isinstance(pinned, str) or len(pinned) != 64: missing_source.append(f"source input {source_input!r} records no pinned tree digest") elif selection and tree_digest(selection) != pinned: missing_source.append( f"the archive selection under {archive_prefix!r} does not match the pinned tree digest " f"{pinned!r}" ) required_paths = record.get("required_paths") or [] verified = record.get("required_paths_verified") if verified is not None and verified != len(required_paths): missing_source.append( f"source input {source_input!r} verified {verified} of {len(required_paths)} required paths" ) for path in spec.get("required_paths", []): if path not in required_paths: missing_source.append(f"the collector report does not require {path!r}") continue if member_for_required_path(archive_prefix, path, audit["member_hashes"]) is None: missing_source.append(f"the archive is missing required path {path!r}") floor = spec.get("min_members") record_floor = record.get("min_members") if floor: if not isinstance(record_floor, int) or record_floor < floor: missing_source.append( f"the collector report's min_members {record_floor!r} is below the trusted floor {floor}" ) if len(selection) < floor: missing_source.append( f"the archive selection has {len(selection)} members, below the trusted floor {floor}" ) if isinstance(record_floor, int) and len(selection) < record_floor: missing_source.append( f"the archive selection has {len(selection)} members, below the report's own min_members {record_floor}" ) evidence = { "source_input": source_input, "archive_prefix": archive_prefix, "members": len(selection), "min_members": record_floor, "min_members_floor": floor, "tree_digest": pinned, "required_paths": required_paths, "required_paths_verified": verified, "locked_dependencies": record.get("dependencies", []), } elif spec.get("required_paths"): for path in spec["required_paths"]: if member_for_required_path(archive_prefix, path, audit["member_hashes"]) is None: missing_source.append(f"the archive is missing required path {archive_prefix}{path!r}") for notice in spec.get("required_notices", []): if notice not in audit["member_hashes"]: missing_notices.append(notice) artifact_evidence, artifact_missing = read_collected_artifacts( collected_dir, spec.get("required_collected_artifacts", []), limits, file_index ) missing_artifacts.extend(artifact_missing) dependencies: list[dict] = [] for dependency in spec.get("dependencies", []): disposition = dependency.get("disposition") if dependency.get("requires_source") and DISPOSITION_STRENGTH.get(disposition, 0) < DISPOSITION_STRENGTH["covered"]: missing_source.append( f"dependency {dependency.get('name')!r} is not covered (disposition {disposition!r})" ) if dependency.get("requires_source"): report_deps = [d for d in evidence.get("locked_dependencies", []) if isinstance(d, dict)] missing_source.extend(check_locked_dependency(dependency, report_deps)) dependencies.append(dict(dependency)) if spec.get("commit") and not _pin_present(report, spec): problems.append(f"{key}: commit {spec['commit']} is not a lineage pin in the collector report") if missing_source: problems.append(f"{key}: source evidence incomplete: {missing_source}") if missing_notices: problems.append(f"{key}: archive is missing notice members {missing_notices}") if missing_artifacts: problems.append(f"{key}: collected artifacts incomplete: {missing_artifacts}") satisfied = not (missing_source or missing_notices or missing_artifacts or not classified) components.append( { "key": key, "component": spec.get("component"), "scope": scope, "status": "satisfied" if satisfied else "gap", "missing_source": missing_source, "missing_notices": missing_notices, "missing_artifacts": missing_artifacts, "artifact_evidence": artifact_evidence, "dependencies": dependencies, "evidence": evidence, } ) reviewed = coverage.get("reviewed") is True complete = reviewed and not problems if not reviewed: problems.append("the coverage mapping is not marked reviewed; the runtime-source hardening pass must confirm it") return {"reviewed": reviewed, "complete": complete, "problems": problems, "components": components} def license_findings(coverage: dict, report: dict) -> list[dict]: """Compare the collector report's obligation labels against the coverage mapping.""" findings: list[dict] = [] obligations = [item for item in report.get("obligations", []) if isinstance(item, dict)] claimed: set[int] = set() for spec in coverage["components"]: accepts = spec.get("license_accepts") or [] needle = spec.get("obligation_match") or OBLIGATION_MATCH.get(spec.get("key")) index = None if needle: for position, obligation in enumerate(obligations): if position in claimed: continue if needle.lower() in (obligation.get("component") or "").lower(): index = position break if index is None: if spec.get("license"): findings.append( { "component": spec.get("component"), "kind": "no_obligation", "declared_license": None, "expected_license": spec.get("license"), "finding": "the collector report records no obligation for this component", } ) continue claimed.add(index) obligation = obligations[index] declared = obligation.get("license") or "" if accepts and not any(declared == value or declared.startswith(value) for value in accepts): findings.append( { "component": obligation.get("component"), "kind": "license_mismatch", "declared_license": declared, "expected_license": spec.get("license"), "finding": "the collector report labels this component with a license the coverage mapping does not accept", } ) for position, obligation in enumerate(obligations): if position not in claimed: findings.append( { "component": obligation.get("component"), "kind": "unaccounted_obligation", "declared_license": obligation.get("license"), "expected_license": None, "finding": "the collector report records an obligation the coverage mapping does not account for", } ) return findings # --------------------------------------------------------------------------- # bundle record and distribution text # --------------------------------------------------------------------------- def staged_source_notice_path(member: str) -> str: rest = member[len(NOTICE_MEMBER_DIR) + 1:] if is_notice_member(member) else member return f"{STAGED_SOURCE_DIR}/{STAGED_SOURCE_NOTICE_DIR}/{rest}" def staged_binary_notice_path(member: str) -> str: rest = member[len(NOTICE_MEMBER_DIR) + 1:] if is_notice_member(member) else member return f"{STAGED_BINARY_NOTICE_DIR}/{rest}" def _coverage_requirements(coverage: dict) -> list[dict]: return [ { "key": spec.get("key"), "component": spec.get("component"), "scope": spec.get("scope"), "license": spec.get("license"), "commit": spec.get("commit"), "source_input": spec.get("source_input"), "archive_prefix": spec.get("archive_prefix"), "min_members": spec.get("min_members", 0), "required_paths": spec.get("required_paths", []), "required_source_prefixes": spec.get("required_source_prefixes", []), "required_notices": spec.get("required_notices", []), "required_collected_artifacts": spec.get("required_collected_artifacts", []), "dependencies": spec.get("dependencies", []), } for spec in coverage.get("components", []) ] def build_bundle_record( *, report: dict, report_name: str, report_sha256: str, coverage: dict, coverage_name: Optional[str], coverage_sha256: Optional[str], coverage_state: dict, findings: list, notice_mapping: list, audit: dict, created_at: str, limits: pr.Limits = pr.DEFAULT_LIMITS, ) -> dict: """The digest-bound record shipped inside the source archive.""" source_archive = report["source_archive"] notices = [ { "member": member, "sha256": audit["member_hashes"][member], "staged_source_path": staged_source_notice_path(member), "staged_binary_path": staged_binary_notice_path(member), } for member in audit["notice_members"] ] status = "collected" if coverage_state["complete"] else "incomplete_coverage" return { "schema": SCHEMA, "record_kind": RECORD_KIND, "role": ( "digest-binds the collected third-party runtime corresponding source and notices, and records " "the mapped required coverage and exact tree evidence the release relies on" ), "created_at": created_at, "generated_by": SELF_TOOL, "archive": { "name": audit["name"], "sha256": audit["sha256"], "size_bytes": audit["size_bytes"], "members": audit["members"], "source_members": audit["source_members"], "tree_digest": audit["tree_digest"], "internal_manifest": audit["internal_manifest"], "internal_manifest_sha256": audit["internal_manifest_sha256"], "internal_manifest_verified": audit["internal_manifest_verified"], "tar_safety": audit["tar_safety"], "bounds": asdict(limits), }, "notices": notices, "notice_mapping": notice_mapping, "notices_listed_by_collector": list(source_archive.get("notices_included", [])), "provenance": { "collector_report": { "path": report_name, "sha256": report_sha256, "record_kind": report.get("record_kind"), "generated_by": report.get("generated_by"), }, "coverage_mapping": { "path": coverage_name, "sha256": coverage_sha256, "reviewed": coverage_state["reviewed"], "legal_review": coverage.get("legal_review"), "maintainer": coverage.get("maintainer"), "source": "built-in trusted minimum" if coverage_name is None else "supplied mapping", }, "pins": report.get("pins", []), "source_inputs": source_archive.get("inputs", []), "collector_limits": report.get("limits", {}), }, "coverage": { "reviewed": coverage_state["reviewed"], "complete": coverage_state["complete"], "problems": coverage_state["problems"], "components": coverage_state["components"], "requirements": _coverage_requirements(coverage), }, "license_findings": findings, "obligations": report.get("obligations", []), "source_completeness": { "status": status, "source_complete": coverage_state["complete"], "release_ready": False, "note": ( "source completeness is claimed only when the mapped required source, notices, artifacts and " "dependency coverage is satisfied and reviewed; release readiness is separate and stays false " "while human legal approval and the clean extracted-source rebuild are outstanding" ), }, "distribution": { "source_archive_path": f"{STAGED_SOURCE_DIR}/{audit['name']}", "bundle_record_path": f"{STAGED_SOURCE_DIR}/{BUNDLE_RECORD_NAME}", "source_notice_root": f"{STAGED_SOURCE_DIR}/{STAGED_SOURCE_NOTICE_DIR}", "binary_notice_root": STAGED_BINARY_NOTICE_DIR, "referenced_by": "SOURCE-AND-LICENSES.md", }, } def render_fragment(record: dict) -> bytes: """The SOURCE-AND-LICENSES.md fragment the packager merges or references.""" archive = record["archive"] coverage = record["coverage"] lines = [ "## Third-party runtime corresponding source and notices", "", "The shipped C64 PRGs link third-party runtime code. Its actual", "corresponding source ships with this release, not just a notice:", "", f"- Archive: `{archive['name']}` (inside `{STAGED_SOURCE_DIR}/`)", f"- SHA-256: `{archive['sha256']}`", f"- Bytes: {archive['size_bytes']}", f"- Members: {archive['members']} " f"({archive['source_members']} source files plus `{archive['internal_manifest']}`)", f"- Tree digest: `{archive['tree_digest']}`", f"- Binding record: `{STAGED_SOURCE_DIR}/{BUNDLE_RECORD_NAME}`", "", "Every member's SHA-256 is listed inside the archive in", f"`{archive['internal_manifest']}`, and each source input is bound to the", "collector's pinned tree digest, so the full selection can be recomputed.", "", "### Required coverage", "", f"Coverage status: **{'satisfied' if coverage['complete'] else 'incomplete'}** " f"(mapping reviewed: {'yes' if coverage['reviewed'] else 'no'}).", "", "| Component | Scope | Status |", "| --- | --- | --- |", ] for component in coverage["components"]: lines.append(f"| {component['component']} | {component['scope']} | {component['status']} |") if coverage["problems"]: lines += ["", "Outstanding coverage items:", ""] lines.extend(f"- {problem}" for problem in coverage["problems"]) lines += [ "", "### Notices bound to their fetched artifacts", "", "Each notice below is byte-identical to the collector's fetched artifact:", "", "| Notice | SHA-256 | Bytes |", "| --- | --- | --- |", ] for entry in record.get("notice_mapping", []): lines.append(f"| `{entry['member']}` | `{entry['sha256']}` | {entry.get('size_bytes')} |") lines += [ "", f"The notice text is also extracted under `{STAGED_BINARY_NOTICE_DIR}/` beside the", "binaries and inside the source archive.", "", "### Obligations", "", "| Component | License | Status |", "| --- | --- | --- |", ] for obligation in record["obligations"]: lines.append( f"| {obligation.get('component')} | {obligation.get('license')} | {obligation.get('status')} |" ) lines += [ "", f"Source completeness: **{record['source_completeness']['status']}**. Release readiness is", "**false**: human legal approval and the clean extracted-source rebuild are still outstanding.", "", "Nothing here is a compiler-reproducibility claim. The pinned toolchain images are", "compiler prerequisites; the release states their identity and licenses, not that the", "compiler output can be reproduced byte for byte.", "", ] return ("\n".join(lines) + "\n").encode("utf-8") # --------------------------------------------------------------------------- # staging (exclusive, atomic, no-follow, owner-checked) # --------------------------------------------------------------------------- def _require_owned(path: Path) -> None: try: info = os.lstat(path) except OSError as error: raise BundleError(f"cannot inspect {path}: {error}") from error if info.st_uid != os.geteuid(): raise BundleError(f"refusing to write into {path}: it is not owned by the current user") def write_staged_file(root: Path, relative: str, data: bytes) -> dict: """Write one staged file exclusively and atomically; never overwrite. The bytes are staged in a uniquely named sibling and linked into place, so an existing file -- including a hardlink to an unrelated file -- is refused rather than truncated. No ``O_TRUNC`` is ever used. """ reason = pr.unsafe_member_name(relative) if reason: raise BundleError(f"refusing to stage an unsafe name ({reason}): {relative!r}") if len(data) > MAX_STAGED_FILE_BYTES: raise BundleError(f"refusing to stage {relative}: {len(data)} bytes is above the staged-file bound") destination = root / relative pr.check_ancestors(destination.parent) destination.parent.mkdir(parents=True, exist_ok=True) pr.check_ancestors(destination.parent) _require_owned(root) _require_owned(destination.parent) staged = destination.parent / f".rtsb-part-{os.getpid()}-{abs(hash(relative)) & 0xFFFFFFFF:08x}" descriptor = os.open( staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o644 ) try: view = memoryview(data) written = 0 while written < len(view): written += os.write(descriptor, view[written:]) os.fsync(descriptor) finally: os.close(descriptor) try: try: os.link(staged, destination) except FileExistsError as error: raise BundleError(f"refusing to overwrite the existing file {destination}") from error except OSError as error: raise BundleError(f"cannot place {destination}: {error}") from error finally: try: os.unlink(staged) except FileNotFoundError: pass return {"path": relative, "size_bytes": len(data), "sha256": sha256_hex(data)} def stage_bundle( stage_dir: Path, *, archive_path: Path, audit: dict, record: dict, fragment: bytes, notice_payloads: Mapping[str, bytes], limits: pr.Limits = pr.DEFAULT_LIMITS, ) -> dict: """Write the ``runtime-sources/`` and ``runtime-notices/`` staging trees.""" pr.check_ancestors(stage_dir) stage_dir = Path(os.path.abspath(stage_dir)) stage_dir.mkdir(parents=True, exist_ok=True) if not stage_dir.is_dir(): raise BundleError(f"the staging path {stage_dir} is not a directory") archive_data = _read_bounded( archive_path, max_bytes=limits.max_archive_bytes, description="runtime source archive" ) if sha256_hex(archive_data) != audit["sha256"]: raise BundleError("the runtime source archive changed between the audit and the staging copy") written: list[dict] = [] written.append(write_staged_file(stage_dir, f"{STAGED_SOURCE_DIR}/{audit['name']}", archive_data)) record_bytes = json.dumps(record, indent=2, sort_keys=True).encode("utf-8") + b"\n" written.append(write_staged_file(stage_dir, f"{STAGED_SOURCE_DIR}/{BUNDLE_RECORD_NAME}", record_bytes)) written.append(write_staged_file(stage_dir, f"{STAGED_SOURCE_DIR}/{FRAGMENT_NAME}", fragment)) for member, payload in sorted(notice_payloads.items()): if len(payload) > MAX_NOTICE_BYTES: raise BundleError(f"notice {member} is {len(payload)} bytes, above the notice bound") if not payload: raise BundleError(f"notice {member} is empty") written.append(write_staged_file(stage_dir, staged_source_notice_path(member), payload)) written.append(write_staged_file(stage_dir, staged_binary_notice_path(member), payload)) return {"stage_dir": str(stage_dir), "files": written} # --------------------------------------------------------------------------- # the manifest patch (proposed; never applied here) # --------------------------------------------------------------------------- def load_source_manifest(path: Path) -> dict: manifest, _ = _load_json(path, max_bytes=MAX_REPORT_BYTES, description="release source manifest") if not isinstance(manifest, dict): raise BundleError(f"{path} is not a JSON object") return manifest def find_inventory_entry(manifest: dict, needle: str) -> Optional[int]: for index, entry in enumerate(manifest.get("inventory", [])): if needle in (entry.get("component") or ""): return index return None _KEYS_FRAGMENTS = { "rust-mos-runtime-library": "Rust-MOS", "rust-mos-toolchain-image": "Rust-MOS C64 toolchain image", "llvm-mos-sdk": "MOS sysroot", "llvm-mos-compiler": "llvm-mos compiler", "rust-std-1.93.0": "Rust standard library", } def _match_by_fragment(manifest: dict, key: str) -> Optional[int]: fragment = _KEYS_FRAGMENTS.get(key) if not fragment: return None for index, entry in enumerate(manifest.get("inventory", [])): if fragment.lower() in (entry.get("component") or "").lower(): return index return None def build_inventory_patch(manifest: dict, record: dict, *, report_name: str) -> dict: """The exact manifest patch a human applies after review and hardening.""" archive = record["archive"] coverage = record["coverage"] ready = coverage["complete"] common = [ f"{STAGED_SOURCE_DIR}/{archive['name']}", f"{STAGED_SOURCE_DIR}/{BUNDLE_RECORD_NAME}", f"{STAGED_SOURCE_DIR}/{FRAGMENT_NAME}", ] retrieval = ( "python3 tools/collect_runtime_sources.py " "--input target/release-runtime-sources/runtime-sources-input.json" ) digest_binding = { "archive": archive["name"], "sha256": archive["sha256"], "size_bytes": archive["size_bytes"], "members": archive["members"], "source_members": archive["source_members"], "tree_digest": archive["tree_digest"], "internal_manifest": archive["internal_manifest"], "internal_manifest_sha256": archive["internal_manifest_sha256"], "collector_report": report_name, } coverage_field = { "reviewed": coverage["reviewed"], "complete": coverage["complete"], "problems": coverage["problems"], } operations: list[dict] = [] replacement_entries: dict[str, dict] = {} added_entries: list[str] = [] bundled_by = ( "python3 tools/runtime_source_bundle.py " "--runtime-sources-report target/release-runtime-sources/runtime-sources-report.json " "--runtime-sources-dir target/release-runtime-sources/collected --stage " ) requirements = coverage.get("requirements") or [] matched: dict[int, list] = {} unmatched: list[dict] = [] for component in requirements: index = find_inventory_entry(manifest, component["component"]) if index is None: index = _match_by_fragment(manifest, component["key"]) if index is None: unmatched.append(component) else: matched.setdefault(index, []).append(component) def _component_detail(component: dict) -> dict: return { "key": component.get("key"), "component": component.get("component"), "scope": component.get("scope"), "license": component.get("license"), "source_input": component.get("source_input"), "required_paths": component.get("required_paths", []), "required_notices": component.get("required_notices", []), } def _license_source_for(components: list) -> str: sources = [] for component in components: notices = component.get("required_notices", []) source = ( staged_source_notice_path(notices[0]) if notices else "runtime-sources/" ) if source not in sources: sources.append(source) return "; ".join(sources) for index in sorted(matched): components = matched[index] entry = dict(manifest["inventory"][index]) paths = set(entry.get("collected_paths") or []) for component in components: paths.update(staged_source_notice_path(m) for m in component.get("required_notices", [])) licenses = [] for component in components: if component.get("license") and component["license"] not in licenses: licenses.append(component["license"]) entry.update( { "collected_paths": sorted(paths), "bundled_by": bundled_by, "retrieval": retrieval, "digest_binding": digest_binding, "runtime_source_coverage": {**coverage_field, "components": [_component_detail(c) for c in components]}, "source_complete": ready, "release_ready": False, "review": entry.get("review", "pending_human_license_approval"), "license": " | ".join(licenses) if licenses else entry.get("license"), "license_source": _license_source_for(components) or entry.get("license_source"), } ) if ready: entry["status"] = "collected" entry["collected_paths"] = sorted(set(entry["collected_paths"]) | set(common)) else: entry["status"] = entry.get("status", "pending_external") replacement_entries[entry["component"]] = entry operations.append({"op": "replace", "path": f"/inventory/{index}", "value": entry}) for component in unmatched: notices = component.get("required_notices", []) paths = {staged_source_notice_path(m) for m in notices} entry = { "component": component.get("component"), "scope": component.get("scope"), "license": component.get("license"), "license_source": _license_source_for([component]), "status": "collected" if ready else "pending_external", "collected_paths": sorted(paths | (set(common) if ready else set())), "bundled_by": bundled_by, "retrieval": retrieval, "digest_binding": digest_binding, "runtime_source_coverage": {**coverage_field, "components": [_component_detail(component)]}, "source_complete": ready, "release_ready": False, "review": "pending_human_license_approval", } added_entries.append(entry["component"]) replacement_entries[entry["component"]] = entry operations.append({"op": "add", "path": "/inventory/-", "value": entry}) include_entry = { "path": SELF_TOOL, "kind": "release-script", "note": "binds the collected runtime corresponding source and notices into the package", } include_paths = [entry.get("path") for entry in manifest.get("source_paths", {}).get("include", [])] include_addition = None if SELF_TOOL in include_paths else include_entry if include_addition is not None: operations.append({"op": "add", "path": "/source_paths/include/-", "value": include_addition}) return { "schema": SCHEMA, "patch_kind": "release_source_manifest_inventory", "target": "tools/release_source_manifest.json", "generated_by": SELF_TOOL, "applied_by": "the manifest owner, after the runtime-source coverage review; this tool does not edit the manifest", "ready_to_apply": ready, "bound_archive": { key: archive[key] for key in ("name", "sha256", "size_bytes", "members", "source_members", "tree_digest") }, "coverage": coverage_field, "operations": operations, "replacement_entries": replacement_entries, "added_entries": added_entries, "source_paths_include_addition": include_addition, "packager_requirement": { "narrow_options": { "--runtime-sources-report": "path to runtime-sources-report.json", "--runtime-sources-dir": "directory holding the verified runtime-corresponding-sources.tar.gz", }, "shared_file": "tools/package_release.py", "requirement": ( "the packager must read the runtime archive through the shared bounded reader, add the " "staged runtime-sources members to the archived_paths set it checks collected inventory " "entries against, and ship the runtime-sources/ tree inside the source archive plus the " "runtime-notices/ tree beside the binaries, so no collected_paths entry is a prose claim" ), "staged_members": common, }, "notes": [ "status is flipped to collected only when the mapped coverage is satisfied and reviewed", "release_ready stays false because human legal approval and the clean extracted-source rebuild are outstanding", ], } # --------------------------------------------------------------------------- # reusable entry point for the release checker # --------------------------------------------------------------------------- def evaluate_bundle_from_archived_evidence( *, report: dict, archive_path: Path, collected_dir: Optional[Path] = None, coverage: Optional[dict] = None, limits: pr.Limits = pr.DEFAULT_LIMITS, created_at: Optional[str] = None, report_name: Optional[str] = None, report_sha256: Optional[str] = None, allow_test_seam: bool = False, ) -> dict: """Return the bundle record for a collector report plus its collected archive. This is the stable entry point the release checker calls with current metadata. It audits the archive, binds it to the report, evaluates the mapped coverage, computes the license findings and notice mapping, and returns the same record the CLI emits. The coverage mapping is the built-in trusted minimum unless a caller passes one. An injected mapping MUST be a trusted, reviewed mapping -- never one derived from an untrusted collector report -- and it is validated against the trusted minimum by default, so it can only strengthen the floor. ``allow_test_seam`` skips that validation and exists only for tests; it must never be used to admit an untrusted or self-approved mapping, because a mapping that merely sets ``reviewed: true`` is not evidence of review. ``source_complete`` is true only when coverage is satisfied and reviewed; ``release_ready`` is always false. """ if coverage is None: coverage = validate_coverage(REQUIRED_COVERAGE) else: coverage = validate_coverage(coverage) if not allow_test_seam: validate_coverage_against_minimum(coverage) collected_dir = Path(collected_dir) if collected_dir is not None else archive_path.parent audit = audit_archive(archive_path, limits=limits) bind_report(report, audit) coverage_state = evaluate_coverage(coverage, audit, report, collected_dir, limits) findings = license_findings(coverage, report) blocking = [finding for finding in findings if finding["kind"] in LICENSE_FINDING_KINDS] if blocking: coverage_state["complete"] = False coverage_state["problems"].extend( f"license coverage: {finding['component']}: {finding['finding']} " f"(declared {finding['declared_license']!r}, expected {finding['expected_license']!r})" for finding in blocking ) notice_mapping, notice_problems = evaluate_notice_mapping(audit, report) if notice_problems: coverage_state["complete"] = False coverage_state["problems"].extend(f"notice mapping: {problem}" for problem in notice_problems) return build_bundle_record( report=report, report_name=report_name if report_name is not None else archive_path.name, report_sha256=report_sha256 or "", coverage=coverage, coverage_name=None, coverage_sha256=None, coverage_state=coverage_state, findings=findings, notice_mapping=notice_mapping, audit=audit, created_at=created_at or utc_now(), limits=limits, ) # --------------------------------------------------------------------------- # entry point # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--runtime-sources-report", "--report", dest="report", type=Path, default=DEFAULT_REPORT, help="collector report to bind (default: target/release-runtime-sources/runtime-sources-report.json)", ) parser.add_argument( "--runtime-sources-dir", "--collected-dir", dest="collected_dir", type=Path, default=DEFAULT_COLLECTED, help="directory holding the verified runtime source archive (default: target/release-runtime-sources/collected)", ) parser.add_argument( "--coverage", type=Path, default=None, help="coverage mapping that may only strengthen the trusted minimum", ) parser.add_argument( "--source-manifest", type=Path, default=DEFAULT_SOURCE_MANIFEST, help="release source manifest, read only (default: tools/release_source_manifest.json)", ) parser.add_argument("--stage", type=Path, default=None, help="write the staged runtime-sources/ and runtime-notices/ trees here") parser.add_argument("--emit-inventory-patch", action="store_true", help="print the exact release_source_manifest.json patch") parser.add_argument("--force", action="store_true", help=argparse.SUPPRESS) parser.add_argument("--json", action="store_true", help="print a machine-readable summary") parser.add_argument("--max-archive-bytes", type=int, default=None, help="test-only bound on the archive size") parser.add_argument("--max-members", type=int, default=None, help="test-only bound on the archive member count") parser.add_argument("--max-member-bytes", type=int, default=None, help="test-only bound on a single member size") return parser def build_limits(args) -> pr.Limits: limits = pr.DEFAULT_LIMITS if args.max_archive_bytes is not None: if args.max_archive_bytes < 1: raise UsageError("--max-archive-bytes must be a positive integer") limits = replace(limits, max_archive_bytes=args.max_archive_bytes) if args.max_members is not None: if args.max_members < 1: raise UsageError("--max-members must be a positive integer") limits = replace(limits, max_members=args.max_members) if args.max_member_bytes is not None: if args.max_member_bytes < 1: raise UsageError("--max-member-bytes must be a positive integer") limits = replace(limits, max_member_bytes=args.max_member_bytes) return limits def run(argv: list[str]) -> int: parser = build_parser() args = parser.parse_args(argv) if args.force: raise UsageError( "--force has been removed: every staged file is created exclusively and atomically, so an " "existing file or a hardlink to an unrelated file is never overwritten" ) limits = build_limits(args) report_path = args.report.absolute() collected_dir = args.collected_dir.absolute() report, report_sha256 = load_report(report_path) archive_path = collected_dir / archive_filename(report) if args.coverage is not None: coverage, coverage_sha256 = load_coverage(args.coverage.absolute()) validate_coverage_against_minimum(coverage) coverage_name = display_path(args.coverage.absolute()) else: coverage = validate_coverage(REQUIRED_COVERAGE) coverage_sha256 = None coverage_name = None audit = audit_archive(archive_path, limits=limits) bind_report(report, audit) coverage_state = evaluate_coverage(coverage, audit, report, collected_dir, limits) findings = license_findings(coverage, report) blocking = [finding for finding in findings if finding["kind"] in LICENSE_FINDING_KINDS] if blocking: coverage_state["complete"] = False coverage_state["problems"].extend( f"license coverage: {finding['component']}: {finding['finding']} " f"(declared {finding['declared_license']!r}, expected {finding['expected_license']!r})" for finding in blocking ) notice_mapping, notice_problems = evaluate_notice_mapping(audit, report) if notice_problems: coverage_state["complete"] = False coverage_state["problems"].extend(f"notice mapping: {problem}" for problem in notice_problems) record = build_bundle_record( report=report, report_name=display_path(report_path), report_sha256=report_sha256, coverage=coverage, coverage_name=coverage_name, coverage_sha256=coverage_sha256, coverage_state=coverage_state, findings=findings, notice_mapping=notice_mapping, audit=audit, created_at=utc_now(), limits=limits, ) fragment = render_fragment(record) stage_result = None if args.stage is not None: notice_payloads = extract_members(archive_path, audit["notice_members"], limits=limits) stage_result = stage_bundle( args.stage, archive_path=archive_path, audit=audit, record=record, fragment=fragment, notice_payloads=notice_payloads, limits=limits, ) patch = None if args.emit_inventory_patch: manifest = load_source_manifest(args.source_manifest.absolute()) patch = build_inventory_patch(manifest, record, report_name=display_path(report_path)) summary = { "report": display_path(report_path), "report_sha256": report_sha256, "archive": display_path(archive_path), "archive_sha256": audit["sha256"], "archive_size_bytes": audit["size_bytes"], "members": audit["members"], "source_members": audit["source_members"], "tree_digest": audit["tree_digest"], "notices": audit["notice_members"], "notices_bound": len(notice_mapping), "coverage_reviewed": coverage_state["reviewed"], "coverage_complete": coverage_state["complete"], "coverage_problems": coverage_state["problems"], "license_findings": findings, "source_complete": coverage_state["complete"], "release_ready": False, "staged": bool(stage_result), "patch": patch is not None, "patch_ready": bool(patch and patch["ready_to_apply"]), } if args.json: payload = dict(summary) if stage_result is not None: payload["staged_files"] = stage_result["files"] if patch is not None: payload["inventory_patch"] = patch print(json.dumps(payload, indent=2, sort_keys=True)) else: print(f"runtime source archive: {summary['archive']}") print(f"sha256: {summary['archive_sha256']}") print( f"members: {summary['members']} " f"({summary['source_members']} source files + internal manifest)" ) print(f"tree digest: {summary['tree_digest']}") print(f"notices bound: {summary['notices_bound']}/{len(summary['notices'])}") print(f"coverage reviewed: {summary['coverage_reviewed']}") print(f"coverage complete: {summary['coverage_complete']}") if coverage_state["problems"]: print("coverage outstanding:") for problem in coverage_state["problems"]: print(f" - {problem}") print(f"source complete: {'collected' if summary['source_complete'] else 'incomplete_coverage'}") print("release ready: false (human legal approval and clean rebuild outstanding)") if findings: print("license findings:") for finding in findings: print(f" - [{finding['kind']}] {finding['component']}: {finding['finding']}") if stage_result is not None: print(f"staged {len(stage_result['files'])} file(s) under {stage_result['stage_dir']}") if patch is not None: print( f"inventory patch operations: {len(patch['operations'])} " f"(ready to apply: {patch['ready_to_apply']})" ) return EXIT_OK def main(argv: Optional[list[str]] = None) -> int: try: return run(list(sys.argv[1:] if argv is None else argv)) except UsageError as error: print(f"error: {error}", file=sys.stderr) return EXIT_USAGE except BundleError as error: print(f"check failed: {error}", file=sys.stderr) return EXIT_FAILED except pr.CheckFailure as error: print(f"check failed: {error}", file=sys.stderr) return EXIT_FAILED except Exception as error: # pragma: no cover - internal error path import traceback traceback.print_exc() print(f"internal error: {error}", file=sys.stderr) return EXIT_INTERNAL if __name__ == "__main__": raise SystemExit(main())