from __future__ import annotations from dataclasses import dataclass import hashlib import json import re import subprocess from pathlib import Path from typing import Any COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") METADATA_TREE_EXCLUDED_NAMES = frozenset({".build-lineage.json"}) MAX_FRONTEND_OVERLAY_FILE_COUNT = 16 UPSTREAM_REPOSITORY = "pingcap-inc/NexusCRM" DOWNSTREAM_REPOSITORY = "pingcap-inc/nexuscrm-pingcap-pack" class FinalPackContractError(RuntimeError): def __init__(self, errors: str | list[str] | tuple[str, ...]) -> None: normalized = (errors,) if isinstance(errors, str) else tuple(errors) self.errors = normalized super().__init__("; ".join(normalized)) @dataclass(frozen=True) class FinalPackReceipt: final_pack_path: str final_pack_root: Path metadata_tree_sha256: str stack_lock_sha256: str intermediate_pack_path: str upstream_repository: str upstream_commit: str official_app_packs_repository: str official_app_packs_commit: str official_pack_path: str def sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def metadata_tree_file_hashes(root: Path) -> dict[str, str]: """Return the canonical served-metadata manifest for one final-pack tree. Build lineage is provenance, not served metadata, so it is deliberately excluded from both this manifest and the release tree digest. """ if not root.is_dir(): raise FinalPackContractError(f"metadata tree path not found: {root}") paths = [ path for path in root.rglob("*") if path.is_file() and path.name not in METADATA_TREE_EXCLUDED_NAMES ] paths.sort(key=lambda path: ("./" + path.relative_to(root).as_posix()).encode("utf-8")) return { "./" + path.relative_to(root).as_posix(): sha256_file(path) for path in paths } def metadata_tree_sha256(root: Path) -> str: manifest = metadata_tree_file_hashes(root) lines = [ f"{digest} {relative}\n".encode("utf-8") for relative, digest in manifest.items() ] return "sha256:" + hashlib.sha256(b"".join(lines)).hexdigest() def _git_json_object( repo_root: Path, revision: str, relative_path: str, label: str, ) -> dict[str, Any]: try: raw = subprocess.run( [ "git", "-C", str(repo_root), "show", f"{revision}:{relative_path}", ], check=True, capture_output=True, text=True, ).stdout except subprocess.CalledProcessError as exc: raise FinalPackContractError( f"{label} not found in exact Git tree: {revision}:{relative_path}" ) from exc try: value = json.loads(raw) except json.JSONDecodeError as exc: raise FinalPackContractError( f"{label} is not valid JSON in exact Git tree: {revision}:{relative_path}: {exc}" ) from exc if not isinstance(value, dict): raise FinalPackContractError( f"{label} must be a JSON object in exact Git tree: {revision}:{relative_path}" ) return value def final_pack_path_from_git(repo_root: Path, revision: str) -> str: """Read the governed final-pack path from one exact downstream Git tree.""" root = repo_root.expanduser().resolve() stack_lock = _git_json_object(root, revision, "stack.lock.json", "stack.lock.json") pack_build = stack_lock.get("pack_build") errors: list[str] = [] pack_build_object = _object(pack_build, "stack.lock.json pack_build", errors) final_pack_path = _relative_path( pack_build_object.get("final_pack_path"), "stack.lock.json pack_build.final_pack_path", errors, ) if errors: raise FinalPackContractError(errors) return final_pack_path def metadata_tree_sha256_from_git( repo_root: Path, revision: str, relative_root: str, ) -> str: """Hash served metadata bytes directly from one exact Git tree. Git object reads keep release verification independent of mutable checkout bytes. The manifest format is identical to :func:`metadata_tree_sha256`. """ root = repo_root.expanduser().resolve() errors: list[str] = [] normalized_root = _relative_path(relative_root, "metadata tree path", errors) if errors: raise FinalPackContractError(errors) try: listing = subprocess.run( [ "git", "-C", str(root), "ls-tree", "-r", "-z", "--full-tree", revision, "--", normalized_root, ], check=True, capture_output=True, ).stdout except subprocess.CalledProcessError as exc: raise FinalPackContractError( f"cannot read metadata tree from Git revision {revision}: {normalized_root}" ) from exc blobs: list[tuple[bytes, str]] = [] prefix = normalized_root.encode("utf-8") + b"/" for raw_entry in listing.split(b"\0"): if not raw_entry: continue metadata, separator, raw_path = raw_entry.partition(b"\t") if not separator: raise FinalPackContractError("metadata Git tree contains a malformed entry") try: mode, object_type, object_id = metadata.decode("ascii").split() except (UnicodeDecodeError, ValueError) as exc: raise FinalPackContractError( "metadata Git tree contains malformed object metadata" ) from exc if mode == "120000": raise FinalPackContractError("metadata Git tree must not contain symlinks") if object_type != "blob" or not mode.startswith("100"): raise FinalPackContractError( "metadata Git tree contains a non-regular file" ) if not raw_path.startswith(prefix): raise FinalPackContractError("metadata Git tree entry escaped its governed path") relative_path = raw_path[len(prefix) :] if relative_path.rsplit(b"/", 1)[-1] == b".build-lineage.json": continue if b"\n" in relative_path or b"\\" in relative_path: raise FinalPackContractError( "metadata Git tree contains an unsupported file name" ) blobs.append((relative_path, object_id)) if not blobs: raise FinalPackContractError("metadata Git tree has no governed payload files") blobs.sort(key=lambda item: item[0]) try: batch = subprocess.run( ["git", "-C", str(root), "cat-file", "--batch"], input=b"".join(object_id.encode("ascii") + b"\n" for _, object_id in blobs), check=True, capture_output=True, ).stdout except subprocess.CalledProcessError as exc: raise FinalPackContractError("cannot read metadata blobs from exact Git tree") from exc cursor = 0 manifest_lines: list[bytes] = [] for relative_path, expected_object_id in blobs: header_end = batch.find(b"\n", cursor) if header_end < 0: raise FinalPackContractError("metadata Git blob batch response is truncated") try: header = batch[cursor:header_end].decode("ascii").split() except UnicodeDecodeError as exc: raise FinalPackContractError("metadata Git blob batch response is invalid") from exc if len(header) != 3 or header[0] != expected_object_id or header[1] != "blob": raise FinalPackContractError("metadata Git blob batch response is invalid") try: size = int(header[2]) except ValueError as exc: raise FinalPackContractError("metadata Git blob batch response is invalid") from exc content_start = header_end + 1 content_end = content_start + size if content_end >= len(batch) or batch[content_end : content_end + 1] != b"\n": raise FinalPackContractError("metadata Git blob batch payload is truncated") content = batch[content_start:content_end] manifest_lines.append( hashlib.sha256(content).hexdigest().encode("ascii") + b" ./" + relative_path + b"\n" ) cursor = content_end + 1 if cursor != len(batch): raise FinalPackContractError("metadata Git blob batch response has trailing data") return "sha256:" + hashlib.sha256(b"".join(manifest_lines)).hexdigest() def final_pack_metadata_sha256_from_git( repo_root: Path, revision: str, ) -> tuple[str, str]: """Return the governed final-pack path and digest from an exact Git tree.""" final_pack_path = final_pack_path_from_git(repo_root, revision) return ( final_pack_path, metadata_tree_sha256_from_git(repo_root, revision, final_pack_path), ) def _load_json_object(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text()) except FileNotFoundError as exc: raise FinalPackContractError(f"{label} not found: {path}") from exc except json.JSONDecodeError as exc: raise FinalPackContractError(f"{label} is not valid JSON: {path}: {exc}") from exc if not isinstance(value, dict): raise FinalPackContractError(f"{label} must be a JSON object: {path}") return value def _object(value: Any, label: str, errors: list[str]) -> dict[str, Any]: if isinstance(value, dict): return value errors.append(f"{label} must be an object") return {} def _relative_path(value: Any, label: str, errors: list[str]) -> str: if not isinstance(value, str) or not value.strip(): errors.append(f"{label} must be a non-empty relative path") return "" path = Path(value) if path.is_absolute() or ".." in path.parts: errors.append(f"{label} must stay inside the repository") return "" return path.as_posix() def _github_repository(value: Any, label: str, errors: list[str]) -> str: if isinstance(value, str): parts = value.split("/") if len(parts) == 2 and all(part.strip() == part and part for part in parts): return value errors.append(f"{label} must be a GitHub owner/repo value") return "" def _source_identity( source: dict[str, Any], label: str, errors: list[str], ) -> tuple[str, str]: repository = _github_repository(source.get("repo"), f"{label}.repo", errors) commit = source.get("ref") if not isinstance(commit, str) or COMMIT_RE.fullmatch(commit) is None: errors.append(f"{label}.ref must be an exact 40-character commit") commit = "" return repository, commit def inspect_checkout( repo_root: Path, *, max_frontend_overlay_file_count: int = MAX_FRONTEND_OVERLAY_FILE_COUNT, ) -> FinalPackReceipt: """Inspect one checkout through the complete final-pack build seam. This proves internal build provenance and the exact served metadata bytes. It deliberately does not select a D3 release; signed BOM authority lives at a separate seam. """ root = repo_root.resolve() stack_lock_path = root / "stack.lock.json" stack_lock = _load_json_object(stack_lock_path, "stack.lock.json") errors: list[str] = [] pack_build = _object(stack_lock.get("pack_build"), "stack.lock.json pack_build", errors) intermediate_pack_path = _relative_path( pack_build.get("intermediate_pack_path"), "stack.lock.json pack_build.intermediate_pack_path", errors, ) final_pack_path = _relative_path( pack_build.get("final_pack_path"), "stack.lock.json pack_build.final_pack_path", errors, ) official_pack_path = _relative_path( pack_build.get("official_pack_path"), "stack.lock.json pack_build.official_pack_path", errors, ) official_repository = _github_repository( pack_build.get("official_app_packs_repo"), "stack.lock.json pack_build.official_app_packs_repo", errors, ) if errors: raise FinalPackContractError(errors) final_pack_root = root / final_pack_path if not final_pack_root.is_dir(): raise FinalPackContractError(f"final pack path not found: {final_pack_root}") if not (final_pack_root / "pack.json").is_file(): errors.append(f"final pack manifest not found: {final_pack_root / 'pack.json'}") lineage_path = final_pack_root / ".build-lineage.json" lineage = _load_json_object(lineage_path, "final pack build lineage") upstream_lineage = _object( lineage.get("upstreamRuntime"), "final pack lineage upstreamRuntime", errors, ) official_lineage = _object( lineage.get("officialPack"), "final pack lineage officialPack", errors, ) upstream_repository, upstream_commit = _source_identity( upstream_lineage, "final pack lineage upstreamRuntime", errors, ) official_lineage_repository, official_commit = _source_identity( official_lineage, "final pack lineage officialPack", errors, ) downstream = _object(lineage.get("downstream"), "final pack lineage downstream", errors) assembly = _object(lineage.get("assembly"), "final pack lineage assembly", errors) if lineage.get("stage") != Path(final_pack_path).name: errors.append("final pack lineage stage does not match the final pack directory") if upstream_repository and upstream_repository != UPSTREAM_REPOSITORY: errors.append( f"final pack lineage upstreamRuntime.repo must be {UPSTREAM_REPOSITORY}" ) if official_lineage_repository and official_lineage_repository != official_repository: errors.append( "final pack lineage officialPack.repo does not match stack.lock.json" ) if official_lineage.get("packPath") != official_pack_path: errors.append( "final pack lineage officialPack.packPath does not match stack.lock.json" ) if downstream.get("repo") != DOWNSTREAM_REPOSITORY: errors.append( f"final pack lineage downstream.repo must be {DOWNSTREAM_REPOSITORY}" ) stack_lock_sha256 = sha256_file(stack_lock_path) if downstream.get("stackLockSha256") != stack_lock_sha256: errors.append( "final pack lineage downstream.stackLockSha256 does not match stack.lock.json" ) if downstream.get("intermediatePackPath") != intermediate_pack_path: errors.append( "final pack lineage downstream.intermediatePackPath does not match stack.lock.json" ) intermediate_pack_root = root / intermediate_pack_path intermediate_manifest = intermediate_pack_root / "pack.json" if not intermediate_manifest.is_file(): errors.append(f"intermediate pack manifest not found: {intermediate_manifest}") elif downstream.get("intermediatePackManifestSha256") != sha256_file(intermediate_manifest): errors.append( "final pack lineage intermediatePackManifestSha256 does not match the intermediate pack" ) overlay_root = intermediate_pack_root / "frontend" overlay_files = ( [path for path in overlay_root.rglob("*") if path.is_file() and not path.is_symlink()] if overlay_root.is_dir() else [] ) overlay_count = len(overlay_files) if overlay_count: expected_overlay_path = f"{intermediate_pack_path}/frontend" if downstream.get("frontendOverlayPath") != expected_overlay_path: errors.append( "final pack lineage frontendOverlayPath does not match the L2 frontend overlay" ) if downstream.get("frontendOverlayFileCount") != overlay_count: errors.append( "final pack lineage frontendOverlayFileCount does not match the L2 frontend overlay " f"(lineage {downstream.get('frontendOverlayFileCount')!r}, actual {overlay_count})" ) elif downstream.get("frontendOverlayFileCount") not in (None, 0): errors.append("final pack lineage declares a missing L2 frontend overlay") if overlay_count > max_frontend_overlay_file_count: errors.append( "L2 frontend overlay exceeds its reviewed file budget " f"({overlay_count} > {max_frontend_overlay_file_count})" ) intermediate_lineage = intermediate_pack_root / ".build-lineage.json" if intermediate_lineage.is_file(): expected_lineage_path = f"{intermediate_pack_path}/.build-lineage.json" if downstream.get("intermediateBuildLineagePath") != expected_lineage_path: errors.append( "final pack lineage intermediateBuildLineagePath does not match the intermediate pack" ) if downstream.get("intermediateBuildLineageSha256") != sha256_file(intermediate_lineage): errors.append( "final pack lineage intermediateBuildLineageSha256 does not match the intermediate pack" ) if assembly.get("finalPackPath") != final_pack_path: errors.append("final pack lineage assembly.finalPackPath does not match stack.lock.json") expected_flows = pack_build.get("flow_whitelist") lineage_flows = assembly.get("flowWhitelist") if isinstance(expected_flows, list) and isinstance(lineage_flows, list): if sorted(expected_flows) != sorted(lineage_flows): errors.append("final pack lineage flowWhitelist does not match stack.lock.json") else: errors.append("final pack flowWhitelist must be declared by stack.lock.json and lineage") if errors: raise FinalPackContractError(errors) return FinalPackReceipt( final_pack_path=final_pack_path, final_pack_root=final_pack_root, metadata_tree_sha256=metadata_tree_sha256(final_pack_root), stack_lock_sha256=stack_lock_sha256, intermediate_pack_path=intermediate_pack_path, upstream_repository=upstream_repository, upstream_commit=upstream_commit, official_app_packs_repository=official_lineage_repository, official_app_packs_commit=official_commit, official_pack_path=official_pack_path, )