#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import os import re import shutil import signal import stat import subprocess import sys import tempfile import time from collections import Counter from pathlib import Path from typing import Any, Iterable, NamedTuple from lib.final_pack_contract import ( FinalPackContractError, FinalPackReceipt, inspect_checkout, metadata_tree_file_hashes, metadata_tree_sha256, ) PLATFORM_PERMISSION_OWNER = "platform-policy" CUSTOM_PERMISSION_FILENAME_SUFFIX = ".custom-permission.json" CUSTOM_PERMISSION_SCHEMA_PREFIX = "nexuscrm.custom-permission/" UPSTREAM_STANDARD_PERMISSIONS_PATH = Path("backend/internal/bootstrap/standard_permissions.json") CONTRACT_MODULE_PATH = "github.com/pingcap-inc/nexuscrm-contract" CONTRACT_STANDARD_PERMISSIONS_PATH = Path("pkg/schemacatalog/standard_permissions.json") GO_MODULE_VERSION_RE = re.compile(r"^v[0-9A-Za-z][0-9A-Za-z.+-]*$") IGNORED_COPY_NAMES = { ".git", ".mypy_cache", ".pytest_cache", ".runtime", ".tmp", ".venv", "__pycache__", "node_modules", } REGEN_WORK_DIR_PREFIX = "final-pack-regen-" REGEN_WORK_DIR_NAME_RE = re.compile(r"final-pack-regen-[a-z0-9_]{8}\Z") REGEN_WORK_DIR_MARKER = ".nexuscrm-final-pack-regen" REGEN_WORK_DIR_MARKER_CONTENT = "nexuscrm-final-pack-regen-v1\n" REGEN_WORK_DIR_RETENTION_SECONDS = 6 * 60 * 60 REGEN_MAX_RUNTIME_SECONDS = 60 * 60 REGEN_SUBPROCESS_TIMEOUT_MARGIN_SECONDS = 30 REGEN_PROCESS_GROUP_TERMINATION_GRACE_SECONDS = 10 LEGACY_REGEN_WORK_DIR_TOP_LEVEL_NAMES = frozenset({"repos", "build-1", "build-2"}) _regen_runtime_deadline: float | None = None class RegenFailure(RuntimeError): pass class PlatformPermissionCatalog(NamedTuple): api_names: frozenset[str] authority: str def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repo-root", default=".") parser.add_argument("--double-build", action="store_true") parser.add_argument("--work-dir", default="") parser.add_argument("--keep-work-dir", action="store_true") parser.add_argument("--artifact-dir", default="") parser.add_argument("--nexuscrm-source-repo", default=os.environ.get("NEXUSCRM_SOURCE_REPO", "")) parser.add_argument( "--official-app-packs-source-repo", default=os.environ.get("OFFICIAL_APP_PACKS_SOURCE_REPO", "") or os.environ.get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT", ""), ) return parser.parse_args() def _warn_regen_cleanup(message: str) -> None: print(f"warning: final pack regen cleanup: {message}", file=sys.stderr) def _has_regen_ownership_marker(path: Path, *, owner_uid: int, device: int) -> bool: marker_path = path / REGEN_WORK_DIR_MARKER flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) marker_fd = os.open(marker_path, flags) try: marker_stat = os.fstat(marker_fd) if not stat.S_ISREG(marker_stat.st_mode): return False if marker_stat.st_uid != owner_uid or marker_stat.st_dev != device: return False expected = REGEN_WORK_DIR_MARKER_CONTENT.encode() return os.read(marker_fd, len(expected) + 1) == expected finally: os.close(marker_fd) def _looks_like_legacy_regen_work_dir(path: Path, *, owner_uid: int, device: int) -> bool: """Recognize the markerless top-level layout emitted by the previous verifier.""" with os.scandir(path) as entries: children = list(entries) if not children: return False for child in children: if child.name not in LEGACY_REGEN_WORK_DIR_TOP_LEVEL_NAMES: return False if not child.is_dir(follow_symlinks=False): return False child_stat = child.stat(follow_symlinks=False) if child_stat.st_uid != owner_uid or child_stat.st_dev != device: return False return True def cleanup_stale_regen_dirs(work_parent: Path, *, now: float | None = None) -> None: if not getattr(shutil.rmtree, "avoids_symlink_attacks", False): _warn_regen_cleanup("disabled because this Python cannot safely remove symlinked trees") return cutoff = (time.time() if now is None else now) - REGEN_WORK_DIR_RETENTION_SECONDS owner_uid = os.geteuid() try: parent_stat = work_parent.stat() with os.scandir(work_parent) as entries: for entry in entries: if REGEN_WORK_DIR_NAME_RE.fullmatch(entry.name) is None: continue try: if not entry.is_dir(follow_symlinks=False): continue original_stat = entry.stat(follow_symlinks=False) if original_stat.st_uid != owner_uid: continue if original_stat.st_dev != parent_stat.st_dev: continue if original_stat.st_mtime >= cutoff: continue path = Path(entry.path) try: owned = _has_regen_ownership_marker( path, owner_uid=owner_uid, device=original_stat.st_dev, ) except FileNotFoundError: owned = _looks_like_legacy_regen_work_dir( path, owner_uid=owner_uid, device=original_stat.st_dev, ) if not owned: continue current_stat = os.lstat(path) if not stat.S_ISDIR(current_stat.st_mode): continue if ( current_stat.st_dev != original_stat.st_dev or current_stat.st_ino != original_stat.st_ino ): continue shutil.rmtree(path) print(f"removed stale final pack regen work dir: {path}") except FileNotFoundError: # Another concurrent regen may already have removed this directory. continue except OSError as exc: _warn_regen_cleanup(f"skipping {entry.path}: {exc}") except OSError as exc: _warn_regen_cleanup(f"unable to scan {work_parent}: {exc}") def create_regen_work_dir(work_parent: Path) -> Path: run_root = Path(tempfile.mkdtemp(prefix=REGEN_WORK_DIR_PREFIX, dir=str(work_parent))) if REGEN_WORK_DIR_NAME_RE.fullmatch(run_root.name) is None: shutil.rmtree(run_root, ignore_errors=True) raise RegenFailure(f"unexpected final pack regen work dir name: {run_root.name}") try: (run_root / REGEN_WORK_DIR_MARKER).write_text(REGEN_WORK_DIR_MARKER_CONTENT) except OSError as exc: shutil.rmtree(run_root, ignore_errors=True) raise RegenFailure(f"cannot mark final pack regen work dir {run_root}: {exc}") from exc return run_root def retain_regen_work_dir(run_root: Path) -> None: try: os.utime(run_root, None, follow_symlinks=False) except FileNotFoundError: return except OSError as exc: _warn_regen_cleanup(f"unable to refresh retained work dir {run_root}: {exc}") def remove_regen_work_dir(run_root: Path) -> None: try: shutil.rmtree(run_root) except FileNotFoundError: return except OSError as exc: _warn_regen_cleanup(f"unable to remove completed work dir {run_root}: {exc}") def redact(text: str, secrets: Iterable[str]) -> str: redacted = text for secret in sorted({secret for secret in secrets if secret}, key=len, reverse=True): redacted = redacted.replace(secret, "") return redacted def _regen_runtime_limit_message() -> str: return f"final pack regen exceeded its {REGEN_MAX_RUNTIME_SECONDS // 60}-minute runtime limit" def _raise_regen_runtime_limit(_signum: int, _frame: object) -> None: raise RegenFailure(_regen_runtime_limit_message()) def start_regen_runtime_limit() -> Any: global _regen_runtime_deadline if _regen_runtime_deadline is not None: raise RegenFailure("final pack regen runtime limit is already active") previous_handler = signal.getsignal(signal.SIGALRM) _regen_runtime_deadline = time.monotonic() + REGEN_MAX_RUNTIME_SECONDS signal.signal(signal.SIGALRM, _raise_regen_runtime_limit) signal.setitimer(signal.ITIMER_REAL, REGEN_MAX_RUNTIME_SECONDS) return previous_handler def stop_regen_runtime_limit(previous_handler: Any) -> None: global _regen_runtime_deadline signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, previous_handler) _regen_runtime_deadline = None def _remaining_subprocess_timeout() -> float | None: if _regen_runtime_deadline is None: return None remaining = _regen_runtime_deadline - time.monotonic() if remaining <= REGEN_SUBPROCESS_TIMEOUT_MARGIN_SECONDS: raise RegenFailure(_regen_runtime_limit_message()) return remaining - REGEN_SUBPROCESS_TIMEOUT_MARGIN_SECONDS def _terminate_process_group(proc: subprocess.Popen[str]) -> None: try: os.killpg(proc.pid, signal.SIGTERM) except ProcessLookupError: pass try: proc.communicate(timeout=REGEN_PROCESS_GROUP_TERMINATION_GRACE_SECONDS) return except subprocess.TimeoutExpired: pass try: os.killpg(proc.pid, signal.SIGKILL) except ProcessLookupError: pass proc.communicate() def run( args: list[str], *, cwd: Path | None = None, env: dict[str, str] | None = None, secrets: Iterable[str] = (), capture: bool = False, ) -> str: timeout = _remaining_subprocess_timeout() proc = subprocess.Popen( args, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True, ) try: stdout, stderr = proc.communicate(timeout=timeout) except subprocess.TimeoutExpired as exc: _terminate_process_group(proc) raise RegenFailure(_regen_runtime_limit_message()) from exc if proc.returncode != 0: sys.stderr.write(redact(stdout, secrets)) sys.stderr.write(redact(stderr, secrets)) command = " ".join(redact(arg, secrets) for arg in args) raise RegenFailure(f"command failed ({proc.returncode}): {command}") if capture: return stdout if stdout: sys.stdout.write(redact(stdout, secrets)) if stderr: sys.stderr.write(redact(stderr, secrets)) return "" def git_has_commit(repo: Path, commit: str) -> bool: try: result = subprocess.run( ["git", "-C", str(repo), "cat-file", "-e", f"{commit}^{{commit}}"], text=True, capture_output=True, check=False, timeout=_remaining_subprocess_timeout(), ) except subprocess.TimeoutExpired as exc: raise RegenFailure(_regen_runtime_limit_message()) from exc return result.returncode == 0 def github_fetch_url(owner_repo: str, token: str) -> str: if token: return f"https://x-access-token:{token}@github.com/{owner_repo}.git" return f"https://github.com/{owner_repo}.git" def ensure_source_repo( *, owner_repo: str, commit: str, source_repo: str, target: Path, token: str, ) -> Path: secrets = [token] if source_repo: repo = Path(source_repo).expanduser().resolve() if not repo.exists(): raise RegenFailure(f"source repo does not exist: {repo}") if git_has_commit(repo, commit): return repo fetch_url = github_fetch_url(owner_repo, token) run(["git", "-C", str(repo), "fetch", "--depth=1", fetch_url, commit], secrets=secrets) if git_has_commit(repo, commit): return repo raise RegenFailure(f"{owner_repo}@{commit} is not available in {repo}") target.parent.mkdir(parents=True, exist_ok=True) run(["git", "init", "-q", str(target)], secrets=secrets) fetch_url = github_fetch_url(owner_repo, token) run(["git", "-C", str(target), "remote", "add", "origin", fetch_url], secrets=secrets) run(["git", "-C", str(target), "fetch", "--depth=1", "origin", commit], secrets=secrets) run(["git", "-C", str(target), "checkout", "--detach", "FETCH_HEAD"], secrets=secrets) run(["git", "-C", str(target), "remote", "set-url", "origin", f"https://github.com/{owner_repo}.git"]) return target def parse_permission_api_names( payload_text: str, *, source: str, require_non_empty: bool = False, ) -> frozenset[str]: try: payload = json.loads(payload_text) except json.JSONDecodeError as exc: raise RegenFailure(f"permission catalog is not valid JSON: {source}: {exc}") from exc if not isinstance(payload, list): raise RegenFailure(f"permission catalog must be a JSON array: {source}") api_names: set[str] = set() for index, row in enumerate(payload): if not isinstance(row, dict): raise RegenFailure(f"permission catalog row {index} must be an object: {source}") api_name = row.get("apiName") if not isinstance(api_name, str) or not api_name.strip(): raise RegenFailure(f"permission catalog row {index} missing apiName: {source}") api_names.add(api_name) if require_non_empty and not api_names: raise RegenFailure(f"permission catalog has no apiName entries: {source}") return frozenset(api_names) def read_permission_api_names(path: Path, *, require_non_empty: bool = False) -> frozenset[str]: try: payload_text = path.read_text() except FileNotFoundError as exc: raise RegenFailure(f"permission catalog not found: {path}") from exc return parse_permission_api_names( payload_text, source=str(path), require_non_empty=require_non_empty, ) def git_show_text(repo: Path, commit: str, path: Path) -> str | None: try: result = subprocess.run( ["git", "-C", str(repo), "show", f"{commit}:{path.as_posix()}"], text=True, capture_output=True, check=False, timeout=_remaining_subprocess_timeout(), ) except subprocess.TimeoutExpired as exc: raise RegenFailure(_regen_runtime_limit_message()) from exc if result.returncode == 0: return result.stdout return None def parse_go_mod_module_version(go_mod_text: str, module_path: str) -> str: in_require_block = False for raw_line in go_mod_text.splitlines(): line = raw_line.split("//", 1)[0].strip() if not line: continue if in_require_block: if line == ")": in_require_block = False continue parts = line.split() if len(parts) >= 2 and parts[0] == module_path: return parts[1] continue if line.startswith("require ("): in_require_block = True continue if line.startswith("require "): parts = line.split() if len(parts) >= 3 and parts[1] == module_path: return parts[2] return "" def escape_go_module_path(module_path: str) -> str: # Mirrors the part of Go's module-cache escaping needed for mixed-case paths. return "".join(f"!{char.lower()}" if "A" <= char <= "Z" else char for char in module_path) def default_go_module_cache_root() -> Path | None: try: cache_root = run(["go", "env", "GOMODCACHE"], capture=True).strip() except (OSError, RegenFailure): return None if not cache_root: return None return Path(cache_root).expanduser().resolve() def contract_standard_permissions_from_cache(cache_root: Path | None, version: str) -> Path | None: if cache_root is None: return None module_dir = cache_root / f"{escape_go_module_path(CONTRACT_MODULE_PATH)}@{version}" candidate = module_dir / CONTRACT_STANDARD_PERMISSIONS_PATH if candidate.is_file(): return candidate return None def fetch_contract_standard_permissions(version: str, work_root: Path | None, token: str) -> Path: if not GO_MODULE_VERSION_RE.fullmatch(version): raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"unsupported {CONTRACT_MODULE_PATH} version {version!r}" ) if work_root is None: raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"{CONTRACT_MODULE_PATH}@{version} is not in the Go module cache" ) target = work_root / f"nexuscrm-contract-{version}" if target.exists(): shutil.rmtree(target) target.parent.mkdir(parents=True, exist_ok=True) secrets = [token] fetch_url = github_fetch_url("pingcap-inc/nexuscrm-contract", token) try: run(["git", "init", "-q", str(target)], secrets=secrets) run(["git", "-C", str(target), "remote", "add", "origin", fetch_url], secrets=secrets) run(["git", "-C", str(target), "fetch", "--depth=1", "origin", f"refs/tags/{version}"], secrets=secrets) run(["git", "-C", str(target), "checkout", "--detach", "FETCH_HEAD"], secrets=secrets) run(["git", "-C", str(target), "remote", "set-url", "origin", "https://github.com/pingcap-inc/nexuscrm-contract.git"]) except RegenFailure as exc: raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"failed to fetch {CONTRACT_MODULE_PATH}@{version}: {exc}" ) from exc candidate = target / CONTRACT_STANDARD_PERMISSIONS_PATH if not candidate.is_file(): raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"{CONTRACT_MODULE_PATH}@{version} missing {CONTRACT_STANDARD_PERMISSIONS_PATH.as_posix()}" ) return candidate def resolve_platform_permission_catalog( *, upstream_repo: Path, upstream_commit: str = "", cache_root: Path | None = None, contract_work_root: Path | None = None, token: str = "", ) -> PlatformPermissionCatalog: if upstream_commit: upstream_standard_text = git_show_text(upstream_repo, upstream_commit, UPSTREAM_STANDARD_PERMISSIONS_PATH) if upstream_standard_text is not None: authority = f"NexusCRM@{upstream_commit}:{UPSTREAM_STANDARD_PERMISSIONS_PATH.as_posix()}" return PlatformPermissionCatalog( api_names=parse_permission_api_names( upstream_standard_text, source=authority, require_non_empty=True, ), authority=authority, ) go_mod_path = Path("backend/go.mod") go_mod_text = git_show_text(upstream_repo, upstream_commit, go_mod_path) if go_mod_text is None: raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"missing NexusCRM@{upstream_commit}:{go_mod_path.as_posix()}" ) else: upstream_standard_permissions = upstream_repo / UPSTREAM_STANDARD_PERMISSIONS_PATH if upstream_standard_permissions.is_file(): return PlatformPermissionCatalog( api_names=read_permission_api_names(upstream_standard_permissions, require_non_empty=True), authority=UPSTREAM_STANDARD_PERMISSIONS_PATH.as_posix(), ) go_mod_path = upstream_repo / "backend" / "go.mod" try: go_mod_text = go_mod_path.read_text() except FileNotFoundError as exc: raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"missing {go_mod_path}" ) from exc contract_version = parse_go_mod_module_version(go_mod_text, CONTRACT_MODULE_PATH) if not contract_version: raise RegenFailure( "cannot determine deterministic platform permission catalog authority: " f"{CONTRACT_MODULE_PATH} is not pinned in {go_mod_path}" ) effective_cache_root = cache_root if cache_root is not None else default_go_module_cache_root() contract_permissions = contract_standard_permissions_from_cache(effective_cache_root, contract_version) if contract_permissions is None: contract_permissions = fetch_contract_standard_permissions(contract_version, contract_work_root, token) authority = f"{CONTRACT_MODULE_PATH}@{contract_version}:{CONTRACT_STANDARD_PERMISSIONS_PATH.as_posix()}" return PlatformPermissionCatalog( api_names=read_permission_api_names(contract_permissions, require_non_empty=True), authority=authority, ) def repo_relative(repo_root: Path, path: Path) -> str: try: return path.resolve().relative_to(repo_root.resolve()).as_posix() except ValueError: return path.as_posix() def load_json_payload(path: Path, *, source: str) -> Any: try: return json.loads(path.read_text()) except FileNotFoundError as exc: raise RegenFailure(f"permission payload not found: {source}") from exc except OSError as exc: raise RegenFailure(f"permission payload not readable: {source}: {exc}") from exc except json.JSONDecodeError as exc: raise RegenFailure(f"permission payload is not valid JSON: {source}: {exc}") from exc def is_custom_permission_document(payload: object) -> bool: if not isinstance(payload, dict): return False schema = payload.get("schema") return isinstance(schema, str) and schema.startswith(CUSTOM_PERMISSION_SCHEMA_PREFIX) def declared_pack_permission_kinds(pack_dir: Path) -> dict[str, str]: pack_json_path = pack_dir / "pack.json" if not pack_json_path.is_file(): return {} pack = load_json_payload(pack_json_path, source=str(pack_json_path)) if not isinstance(pack, dict): raise RegenFailure(f"pack manifest must be a JSON object: {pack_json_path}") kinds: dict[str, str] = {} for component in pack.get("components") or []: if not isinstance(component, dict): continue kind = component.get("kind") relative = component.get("path") if kind in {"permission", "custom_permission"} and isinstance(relative, str) and relative.strip(): kinds[relative] = str(kind) return kinds def classify_permissions_file( path: Path, *, pack_dir: Path, declared_kinds: dict[str, str], ) -> str: source = str(path) payload = load_json_payload(path, source=source) dedicated = is_custom_permission_document(payload) try: relative = path.resolve().relative_to(pack_dir.resolve()).as_posix() except ValueError: relative = path.name declared = declared_kinds.get(relative) if path.name.endswith(CUSTOM_PERMISSION_FILENAME_SUFFIX) and not dedicated: raise RegenFailure( "permissions file " f"{source} uses {CUSTOM_PERMISSION_FILENAME_SUFFIX} but is not a " f"{CUSTOM_PERMISSION_SCHEMA_PREFIX}* document" ) if declared == "custom_permission" and not dedicated: raise RegenFailure( "pack component custom_permission " f"{relative} is not a {CUSTOM_PERMISSION_SCHEMA_PREFIX}* document: {source}" ) if declared == "permission" and dedicated: raise RegenFailure( "pack component permission " f"{relative} is a dedicated CustomPermission document: {source}" ) if dedicated: return "custom_permission" return "catalog" def downstream_permission_catalog_paths( repo_root: Path, pack_paths: Iterable[str], ) -> list[Path]: paths: list[Path] = [] seen: set[Path] = set() for pack_path in pack_paths: if not pack_path.strip(): continue pack_dir = repo_root / pack_path permissions_dir = pack_dir / "permissions" if not permissions_dir.is_dir(): continue declared_kinds = declared_pack_permission_kinds(pack_dir) for path in sorted(permissions_dir.glob("*.json")): resolved = path.resolve() if not path.is_file() or resolved in seen: continue if classify_permissions_file( path, pack_dir=pack_dir, declared_kinds=declared_kinds, ) == "custom_permission": continue seen.add(resolved) paths.append(path) return paths def collect_platform_permission_catalog_errors( *, repo_root: Path, pack_paths: Iterable[str], platform_permission_api_names: Iterable[str], ) -> list[str]: platform_api_names = set(platform_permission_api_names) errors: list[str] = [] for path in downstream_permission_catalog_paths(repo_root, pack_paths): downstream_api_names = read_permission_api_names(path) for api_name in sorted(downstream_api_names & platform_api_names): errors.append( "platform permission catalog collision: " f"{api_name} in {repo_relative(repo_root, path)}; " f"expected owner {PLATFORM_PERMISSION_OWNER}" ) return errors def copy_repo_tree(source: Path, target: Path) -> None: def ignore(_dir: str, names: list[str]) -> set[str]: return {name for name in names if name in IGNORED_COPY_NAMES} shutil.copytree( source, target, symlinks=True, ignore=ignore, ignore_dangling_symlinks=True, ) def component_identity(component: dict[str, Any]) -> tuple[object, object, object, str]: return ( component.get("kind"), component.get("businessKey"), component.get("path"), json.dumps(component.get("dependsOn", []), sort_keys=True, separators=(",", ":")), ) def sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def git_rev_parse(repo_root: Path, ref: str) -> str: return run(["git", "rev-parse", ref], cwd=repo_root, capture=True).strip() def tree_files(root: Path) -> dict[str, str]: return metadata_tree_file_hashes(root) def diff_pack_trees(expected_root: Path, actual_root: Path) -> list[str]: expected = tree_files(expected_root) actual = tree_files(actual_root) changes: list[str] = [] for path in sorted(set(expected) | set(actual)): if path not in expected: changes.append(f"added in rebuilt pack: {path}") elif path not in actual: changes.append(f"missing from rebuilt pack: {path}") elif expected[path] != actual[path]: changes.append(f"changed in rebuilt pack: {path}") return changes def summarize_pack_json_diff(expected_root: Path, actual_root: Path) -> list[str]: expected_path = expected_root / "pack.json" actual_path = actual_root / "pack.json" if not expected_path.is_file() or not actual_path.is_file(): return ["pack.json drift classification: semantic", "pack.json is missing from one side"] expected_payload = json.loads(expected_path.read_text()) actual_payload = json.loads(actual_path.read_text()) expected_components = expected_payload.get("components", []) actual_components = actual_payload.get("components", []) expected_ids = [component_identity(component) for component in expected_components if isinstance(component, dict)] actual_ids = [component_identity(component) for component in actual_components if isinstance(component, dict)] if Counter(expected_ids) == Counter(actual_ids): return [ "pack.json drift classification: order-only", "The rebuilt final pack has the same component multiset as the committed mirror.", ] expected_only = sorted(set(expected_ids) - set(actual_ids)) actual_only = sorted(set(actual_ids) - set(expected_ids)) lines = ["pack.json drift classification: semantic"] if expected_only: lines.append("Committed-only components:") for kind, business_key, path, _depends_on in expected_only[:20]: lines.append(f" - {kind}:{business_key} ({path})") if actual_only: lines.append("Rebuilt-only components:") for kind, business_key, path, _depends_on in actual_only[:20]: lines.append(f" - {kind}:{business_key} ({path})") return lines def collect_lineage_identity_errors(final_pack: Path, label: str) -> list[str]: lineage_path = final_pack / ".build-lineage.json" if not lineage_path.is_file(): return [] try: lineage = json.loads(lineage_path.read_text()) except json.JSONDecodeError: return [] if not isinstance(lineage, dict): return [] downstream = lineage.get("downstream") if isinstance(downstream, dict) and "ref" in downstream: return [ f"{label} final pack lineage must not contain downstream.ref; " "the containing Git commit is the downstream release identity" ] return [] def verify_rebuilt_pack( *, source_receipt: FinalPackReceipt, rebuilt_receipt: FinalPackReceipt, label: str = "rebuilt", ) -> list[str]: final_pack_path = source_receipt.final_pack_root rebuilt_final_pack = rebuilt_receipt.final_pack_root expected_hash = source_receipt.metadata_tree_sha256.removeprefix("sha256:") errors: list[str] = [] errors.extend(collect_lineage_identity_errors(final_pack_path, "committed")) errors.extend(collect_lineage_identity_errors(rebuilt_final_pack, label)) committed_hash = metadata_tree_sha256(final_pack_path).removeprefix("sha256:") if committed_hash != expected_hash: errors.append( "final-pack contract digest does not match committed final pack " f"(contract {expected_hash}, committed {committed_hash})" ) rebuilt_hash = metadata_tree_sha256(rebuilt_final_pack).removeprefix("sha256:") if rebuilt_hash != expected_hash: errors.append( f"{label} final pack hash does not match the committed contract " f"(contract {expected_hash}, {label} {rebuilt_hash})" ) source_identity = ( source_receipt.upstream_repository, source_receipt.upstream_commit, source_receipt.official_app_packs_repository, source_receipt.official_app_packs_commit, source_receipt.official_pack_path, ) rebuilt_identity = ( rebuilt_receipt.upstream_repository, rebuilt_receipt.upstream_commit, rebuilt_receipt.official_app_packs_repository, rebuilt_receipt.official_app_packs_commit, rebuilt_receipt.official_pack_path, ) if rebuilt_identity != source_identity: errors.append(f"{label} final pack source identity differs from committed lineage") changes = diff_pack_trees(final_pack_path, rebuilt_final_pack) if changes: errors.append(f"{label} final pack differs from committed mirror:") errors.extend(f" - {change}" for change in changes[:100]) if len(changes) > 100: errors.append(f" - ... {len(changes) - 100} more changes") if any(change.endswith("./pack.json") for change in changes): errors.extend(summarize_pack_json_diff(final_pack_path, rebuilt_final_pack)) return errors def verify_double_build(first_final_pack: Path, second_final_pack: Path) -> list[str]: errors: list[str] = [] first_hash = metadata_tree_sha256(first_final_pack).removeprefix("sha256:") second_hash = metadata_tree_sha256(second_final_pack).removeprefix("sha256:") if first_hash != second_hash: errors.append( "double-build determinism failed " f"(first {first_hash}, second {second_hash})" ) changes = diff_pack_trees(first_final_pack, second_final_pack) if changes: errors.append("double-build outputs differ:") errors.extend(f" - {change}" for change in changes[:100]) if len(changes) > 100: errors.append(f" - ... {len(changes) - 100} more changes") if any(change.endswith("./pack.json") for change in changes): errors.extend(summarize_pack_json_diff(first_final_pack, second_final_pack)) return errors def build_from_contract( *, repo_root: Path, build_root: Path, upstream_repo: Path, app_packs_repo: Path, ) -> FinalPackReceipt: copy_repo_tree(repo_root, build_root) env = os.environ.copy() env["NEXUSCRM_OFFICIAL_APP_PACKS_ROOT"] = str(app_packs_repo) run( [ "bash", "scripts/build_pingcap_salesforce_sample_pack.sh", "--reset", "--from-final-pack-lineage", "--nexuscrm-source-repo", str(upstream_repo), "--official-app-packs-source-repo", str(app_packs_repo), ], cwd=build_root, env=env, ) try: return inspect_checkout(build_root) except FinalPackContractError as exc: raise RegenFailure(f"rebuilt final-pack contract is invalid: {exc}") from exc def publish_artifacts( *, artifact_dir: Path, receipt: FinalPackReceipt, repo_root: Path, double_build: bool, ) -> None: if artifact_dir.exists(): shutil.rmtree(artifact_dir) artifact_dir.mkdir(parents=True) final_pack_artifact = artifact_dir / "final-pack" shutil.copytree(receipt.final_pack_root, final_pack_artifact, symlinks=True) lineage_path = receipt.final_pack_root / ".build-lineage.json" if lineage_path.is_file(): shutil.copy2(lineage_path, artifact_dir / "build-lineage.json") final_pack_hash = receipt.metadata_tree_sha256.removeprefix("sha256:") pack_build = { "stack_lock_sha256": receipt.stack_lock_sha256, "intermediate_pack_path": receipt.intermediate_pack_path, "final_pack_path": receipt.final_pack_path, "final_pack_hash": final_pack_hash, } manifest = { "schema_version": 1, "artifact_kind": "nexuscrm-final-pack", "final_pack_path": "final-pack", "final_pack_hash": final_pack_hash, "double_build_verified": double_build, "source": { "downstream": { "repo": "pingcap-inc/nexuscrm-pingcap-pack", "commit": git_rev_parse(repo_root, "HEAD"), }, "upstream": { "repo": receipt.upstream_repository, "commit": receipt.upstream_commit, }, "official_app_packs": { "repo": receipt.official_app_packs_repository, "commit": receipt.official_app_packs_commit, "pack_path": receipt.official_pack_path, }, }, "pack_build": pack_build, } (artifact_dir / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") def main() -> int: args = parse_args() repo_root = Path(args.repo_root).expanduser().resolve() work_parent = ( Path(args.work_dir).expanduser().resolve() if args.work_dir else Path(tempfile.gettempdir()).resolve() ) try: source_receipt = inspect_checkout(repo_root) except FinalPackContractError as exc: for error in exc.errors: print(f"final-pack contract validation failed: {error}", file=sys.stderr) return 1 cleanup_stale_regen_dirs(work_parent) upstream_repo_name = source_receipt.upstream_repository app_packs_repo_name = source_receipt.official_app_packs_repository token = os.environ.get("GH_TOKEN", "").strip() or os.environ.get("GITHUB_TOKEN", "").strip() run_root = create_regen_work_dir(work_parent) completed_successfully = False previous_alarm_handler = start_regen_runtime_limit() try: repos_root = run_root / "repos" upstream_repo = ensure_source_repo( owner_repo=upstream_repo_name, commit=source_receipt.upstream_commit, source_repo=args.nexuscrm_source_repo, target=repos_root / "NexusCRM", token=token, ) app_packs_repo = ensure_source_repo( owner_repo=app_packs_repo_name, commit=source_receipt.official_app_packs_commit, source_repo=args.official_app_packs_source_repo, target=repos_root / "nexuscrm-app-packs", token=token, ) platform_catalog = resolve_platform_permission_catalog( upstream_repo=upstream_repo, upstream_commit=source_receipt.upstream_commit, contract_work_root=repos_root, token=token, ) all_errors = collect_platform_permission_catalog_errors( repo_root=repo_root, pack_paths=( source_receipt.intermediate_pack_path, source_receipt.final_pack_path, ), platform_permission_api_names=platform_catalog.api_names, ) if all_errors: print("final pack regen verification FAILED", file=sys.stderr) print(f"platform permission catalog authority: {platform_catalog.authority}", file=sys.stderr) for error in all_errors: print(error, file=sys.stderr) print(f"work dir retained for inspection: {run_root}", file=sys.stderr) args.keep_work_dir = True return 1 first_receipt = build_from_contract( repo_root=repo_root, build_root=run_root / "build-1", upstream_repo=upstream_repo, app_packs_repo=app_packs_repo, ) all_errors = verify_rebuilt_pack( source_receipt=source_receipt, rebuilt_receipt=first_receipt, label="rebuilt", ) if args.double_build: second_receipt = build_from_contract( repo_root=repo_root, build_root=run_root / "build-2", upstream_repo=upstream_repo, app_packs_repo=app_packs_repo, ) all_errors.extend( verify_double_build( first_receipt.final_pack_root, second_receipt.final_pack_root, ) ) if all_errors: print("final pack regen verification FAILED", file=sys.stderr) for error in all_errors: print(error, file=sys.stderr) print(f"work dir retained for inspection: {run_root}", file=sys.stderr) args.keep_work_dir = True return 1 if args.artifact_dir: publish_artifacts( artifact_dir=Path(args.artifact_dir).expanduser().resolve(), receipt=first_receipt, repo_root=repo_root, double_build=args.double_build, ) print("final pack regen verification PASSED") print(f"final-pack metadata tree: {source_receipt.metadata_tree_sha256}") print(f"platform permission catalog authority: {platform_catalog.authority}") if args.artifact_dir: print(f"artifact dir: {Path(args.artifact_dir).expanduser().resolve()}") print(f"work dir: {run_root}") completed_successfully = True return 0 except RegenFailure as exc: print(f"final pack regen verification FAILED: {exc}", file=sys.stderr) print(f"work dir retained for inspection: {run_root}", file=sys.stderr) args.keep_work_dir = True return 1 finally: try: if args.keep_work_dir or not completed_successfully: retain_regen_work_dir(run_root) else: remove_regen_work_dir(run_root) finally: stop_regen_runtime_limit(previous_alarm_handler) if __name__ == "__main__": sys.exit(main())