#!/usr/bin/env python3 """Install one L1 AppRelease into an existing neutral L0 authority. The installer consumes only L1 release artifacts and explicit L0/operator facts. A D3/L2 BOM is deliberately not an input. """ from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import re import secrets import shutil import subprocess import sys import tempfile import time from typing import Any, Callable, Sequence DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") UUID7 = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") CATALOG_RECONCILE_IMAGE = re.compile( r"^ghcr\.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@sha256:[0-9a-f]{64}$" ) CATALOG_CONSUMER_CAPABILITIES = { "catalog.consumer.app-metadata-view.v1", "catalog.consumer.app-metadata-view.v2", "catalog.consumer.app-metadata-view.v3", "catalog.consumer.integration-binding-view.v1", } APPROVED_PROFILE_DIGESTS = { "sha256:9ad5917dbceca912c6b7be10ab1c4685f60718dfeb169969820e69c4c4d13cc8": "sha256:9816ebfcb7713791f9d1ab6c03292aaddff8b633d2a130a2db1a79f2818737c5", "sha256:0809e2a0ba2e1096577856fded3c2b2cac498b063e050b9647d3a6c13b159df7": "sha256:2f3e633a59d70dfb67998552744ed8bbb23023979190dad65784c10ecaa81825", "sha256:ab1bf9f72573f855dbf541240d793360de42c484b6558b5e2c485795ae5fc6ef": "sha256:af2c4c911bca3b86df12f474d3317762348efaeafc94443f09eb1a4b6f7dac00", "sha256:c2caad0db81f67563eb9660098f7002b114947416217b8599b0aa45d1b574845": "sha256:601cd65d47c519e817260fbf27a5477d8a023e506b6e90d2b86390e2cdaa3a4a", "sha256:739ee4c8111fccb4af65d6d795391e4b3d0468671aabc9672943f868483a5fa9": "sha256:b9e50c618d4257d73cf334ee43a51c14bc8896d4047da542177f3eecad452395", } class InstallError(RuntimeError): pass def _pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise InstallError(f"duplicate JSON key {key!r}") result[key] = value return result def load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(), object_pairs_hook=_pairs_no_duplicates) except (OSError, json.JSONDecodeError) as exc: raise InstallError(f"read {label}: {exc}") from exc if not isinstance(value, dict): raise InstallError(f"{label} must be a JSON object") return value def require_digest(value: str, label: str) -> str: if not DIGEST.fullmatch(value): raise InstallError(f"{label} must be a lowercase sha256 digest") return value def new_uuid7() -> str: raw = bytearray(int(time.time() * 1000).to_bytes(6, "big") + secrets.token_bytes(10)) raw[6] = (raw[6] & 0x0F) | 0x70 raw[8] = (raw[8] & 0x3F) | 0x80 hexed = raw.hex() return f"{hexed[:8]}-{hexed[8:12]}-{hexed[12:16]}-{hexed[16:20]}-{hexed[20:]}" def verify_release(release_dir: Path, repository_root: Path) -> None: command = ( "go", "run", ".", "verify", "--package", str(release_dir / "app-package.json"), "--attestation", str(release_dir / "publisher-attestation.json"), "--public-key", str(release_dir / "publisher-public-key.json"), ) completed = subprocess.run( command, cwd=repository_root / "tools" / "crm-app-release", env={**os.environ, "GOWORK": "off"}, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if completed.returncode != 0: detail = completed.stderr.strip() or completed.stdout.strip() raise InstallError(f"signed AppRelease verification failed: {detail}") def validate_release_closure( package: dict[str, Any], profile: dict[str, Any], ) -> tuple[dict[str, Any], str]: if package.get("schema_version") != "nexuscrm.app-package-artifact/v1": raise InstallError("unsupported AppPackage artifact schema") manifest = package.get("canonical_manifest") if not isinstance(manifest, dict): raise InstallError("AppPackage canonical manifest is missing") release_digest = require_digest(str(package.get("app_release_digest", "")), "AppRelease digest") if ( manifest.get("app_id") != profile.get("appId") or release_digest != profile.get("appReleaseDigest") ): raise InstallError("L1 install profile does not bind the exact AppRelease") if canonical_digest(profile) != APPROVED_PROFILE_DIGESTS.get(release_digest): raise InstallError("L1 install profile is not the approved policy for this AppRelease") metadata_digest = require_digest( str(manifest.get("metadata_bundle_digest", "")), "MetadataBundle digest", ) if profile.get("schema") != "nexuscrm.l1-install-profile/v1": raise InstallError("unsupported L1 install profile schema") workload = profile.get("handlerWorkloadId") authorized = profile.get("authorizedWorkloadIds") if not isinstance(workload, str) or not workload or not isinstance(authorized, list) or workload not in authorized: raise InstallError("L1 install profile must authorize its handler workload") if authorized != sorted(set(authorized)): raise InstallError("authorized workload IDs must be unique and sorted") provided = manifest.get("provided_capabilities") required = manifest.get("required_engine_capabilities") selections = profile.get("capabilities") grants = profile.get("consumerGrants") workload_capabilities = profile.get("workloadCapabilities") catalog_consumer_capabilities = profile.get("catalogConsumerCapabilities") if not all(isinstance(value, list) for value in ( provided, required, selections, grants, workload_capabilities, catalog_consumer_capabilities, )): raise InstallError("AppRelease or L1 install profile capability closure is incomplete") provided_by_id = { row.get("capability_id"): row for row in provided if isinstance(row, dict) } if len(provided_by_id) != len(provided) or len(selections) != len(provided): raise InstallError("L1 install profile must select every App capability exactly once") selected_runtime: set[str] = set() selected_app: set[str] = set() for row in selections: if not isinstance(row, dict) or set(row) != { "appCapabilityId", "handlerRef", "runtimeCapabilityId", }: raise InstallError("L1 capability selection has an unknown shape") app_capability = row["appCapabilityId"] runtime_capability = row["runtimeCapabilityId"] declared = provided_by_id.get(app_capability) if ( not isinstance(declared, dict) or declared.get("kind") != "service-handler" or declared.get("handler_ref") != row["handlerRef"] or runtime_capability not in declared.get("required_runtime_capabilities", []) or runtime_capability not in required or app_capability in selected_app ): raise InstallError("L1 capability selection disagrees with the signed AppRelease") selected_app.add(app_capability) selected_runtime.add(runtime_capability) grant_pairs: set[tuple[str, str]] = set() handler_grants: set[str] = set() for grant in grants: if not isinstance(grant, dict) or set(grant) != {"authorizedWorkloadId", "capabilityId"}: raise InstallError("L1 consumer grant has an unknown shape") pair = (grant["authorizedWorkloadId"], grant["capabilityId"]) if pair in grant_pairs or pair[0] not in authorized or pair[1] not in required: raise InstallError("L1 consumer grant is duplicate or outside AppRelease authority") grant_pairs.add(pair) if pair[0] == workload: handler_grants.add(pair[1]) for capability in provided: for dependency in capability["required_runtime_capabilities"]: if dependency in CATALOG_CONSUMER_CAPABILITIES: if dependency not in handler_grants: raise InstallError(f"handler workload lacks required Catalog grant {dependency}") elif dependency not in selected_runtime and dependency not in handler_grants: raise InstallError(f"handler dependency {dependency} has no serving selection") workload_pairs: set[tuple[str, str]] = set() for capability in workload_capabilities: if not isinstance(capability, dict) or set(capability) != { "authorizedWorkloadId", "capabilityId", }: raise InstallError("L1 workload capability has an unknown shape") pair = (capability["authorizedWorkloadId"], capability["capabilityId"]) if pair in workload_pairs or pair[0] not in authorized or pair[1] not in required: raise InstallError("L1 workload capability is duplicate or outside AppRelease authority") workload_pairs.add(pair) runtime_consumer_pairs: set[tuple[str, str]] = set() for capability in catalog_consumer_capabilities: if not isinstance(capability, dict) or set(capability) != { "authorizedWorkloadId", "capabilityId", "catalogSnapshot", } or capability["catalogSnapshot"] not in {"none", "installation"}: raise InstallError("L1 Catalog consumer capability has an unknown shape") pair = (capability["authorizedWorkloadId"], capability["capabilityId"]) if pair in runtime_consumer_pairs or pair[0] not in authorized or pair[1] not in required: raise InstallError("L1 Catalog consumer capability is duplicate or outside AppRelease authority") runtime_consumer_pairs.add(pair) return manifest, metadata_digest def host_compatibility(runtime_path: Path, catalog_path: Path) -> dict[str, Any]: runtime = load_json(runtime_path, "Runtime host compatibility") catalog = load_json(catalog_path, "Catalog host compatibility") if runtime.get("schema") != "nexuscrm.runtime-host-compatibility/v1": raise InstallError("unsupported Runtime host compatibility schema") if catalog.get("schema") != "nexuscrm.catalog-host-compatibility/v1": raise InstallError("unsupported Catalog host compatibility schema") protocols = runtime.get("protocols") capabilities = runtime.get("engineCapabilities") kinds = catalog.get("componentKinds") if not all(isinstance(value, list) for value in (protocols, capabilities, kinds)): raise InstallError("host compatibility facts are incomplete") return { "protocols": sorted(protocols, key=lambda row: (row["protocolId"], row["major"], row["minor"])), "engineCapabilities": sorted(capabilities), "componentKinds": sorted(kinds, key=lambda row: (row["componentKind"], row["componentFormatVersion"])), } def render_config( args: argparse.Namespace, verifier: Callable[[Path, Path], None] = verify_release, *, profile: dict[str, Any] | None = None, ) -> dict[str, Any]: release_dir = args.release_dir.resolve() repository_root = Path(__file__).resolve().parents[2] verifier(release_dir, repository_root) package = load_json(release_dir / "app-package.json", "AppPackage") attestation = load_json(release_dir / "publisher-attestation.json", "publisher attestation") key = load_json(release_dir / "publisher-public-key.json", "publisher public key") if profile is None: profile = load_json(args.profile.resolve(), "L1 install profile") manifest, metadata_digest = validate_release_closure(package, profile) if ( attestation.get("app_release_digest") != package["app_release_digest"] or attestation.get("publisher_id") != key.get("publisher_id") or attestation.get("key_id") != key.get("key_id") ): raise InstallError("signed AppRelease carriers disagree") trusted_key_digest = require_digest( args.trusted_publisher_key_digest, "trusted publisher key digest", ) if canonical_digest(key) != trusted_key_digest: raise InstallError("publisher public key is not approved by the deployment trust input") require_digest(args.service_release_digest, "sfcrm service release digest") frontend_mode = manifest.get("frontend_compatibility", {}).get("mode") frontend: dict[str, str] | None = None if frontend_mode == "required": frontend = { "releaseDigest": require_digest(args.frontend_release_digest, "frontend release digest"), "compatibilityReceiptDigest": require_digest( args.frontend_compatibility_receipt_digest, "frontend compatibility receipt digest", ), } elif frontend_mode != "none": raise InstallError(f"unsupported frontend compatibility mode {frontend_mode!r}") elif args.frontend_release_digest or args.frontend_compatibility_receipt_digest: raise InstallError("AppRelease does not accept a frontend binding") if args.operation_kind == "PREPARE" and args.app_installation_id: raise InstallError("PREPARE cannot provide an AppInstallation ID") if args.operation_kind != "PREPARE" and not args.app_installation_id: raise InstallError(f"{args.operation_kind} requires an AppInstallation ID") command_id = args.publication_command_id or new_uuid7() if not UUID7.fullmatch(command_id): raise InstallError("publication command ID must be a lowercase UUIDv7") config: dict[str, Any] = { "schema": "nexuscrm.catalog-reconcile-config/v4", "orgId": args.org_id, "appId": manifest["app_id"], "expectedAppReleaseDigest": package["app_release_digest"], "metadataSourceDigest": metadata_digest, "releaseImport": { "packagePath": "/release/app-package.json", "attestationPath": "/release/publisher-attestation.json", "publisher": { "publisherId": key["publisher_id"], "keys": [{ "keyId": key["key_id"], "algorithm": key["algorithm"], "publicKeyBase64": key["public_key"], "validFrom": args.publisher_valid_from, "validUntil": args.publisher_valid_until, }], }, }, "artifactStoreRoot": "/var/lib/nexuscrm/catalog", "hostCompatibility": host_compatibility( args.runtime_host_compatibility, args.catalog_host_compatibility, ), "idempotencyKey": args.idempotency_key, "operationKind": args.operation_kind, "appInstallationId": args.app_installation_id, "expectedCatalogRevision": args.expected_catalog_revision, "expectedSecurityEpoch": args.expected_security_epoch, "runtimeAudience": args.runtime_audience, "workloadId": profile["handlerWorkloadId"], "authorizedWorkloadIds": profile["authorizedWorkloadIds"], "appCapabilityId": "", "handlerRef": "", "runtimeCapabilityId": "", "capabilities": profile["capabilities"], "serviceReleaseDigest": args.service_release_digest, "frontend": frontend, "schemaGeneration": args.schema_generation, "engineGeneration": args.engine_generation, "resolvedDependencies": [], "consumerGrants": profile["consumerGrants"], "integrationBindings": [], "phaseOnePublication": {"commandId": command_id}, } return config def write_config(config: dict[str, Any], output: Path) -> None: output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(config, sort_keys=True, separators=(",", ":")) + "\n") output.chmod(0o400) def validate_receipt(config: dict[str, Any], path: Path) -> dict[str, Any]: receipt = load_json(path, "Catalog reconcile receipt") expected = { "orgId": config["orgId"], "appReleaseDigest": config["expectedAppReleaseDigest"], "serviceReleaseDigest": config["serviceReleaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "workloadId": config["workloadId"], "catalogRevision": config["expectedCatalogRevision"] + 1, "securityEpoch": config["expectedSecurityEpoch"] + 1, "schemaGeneration": config["schemaGeneration"], "engineGeneration": config["engineGeneration"], } for field, value in expected.items(): if receipt.get(field) != value: raise InstallError(f"Catalog receipt {field} does not match the L1 install request") requested_installation = config.get("appInstallationId") if requested_installation and receipt.get("appInstallationId") != requested_installation: raise InstallError( "Catalog receipt appInstallationId does not match the L1 install request" ) if not receipt.get("appInstallationId") or not receipt.get("catalogSnapshotDigest"): raise InstallError("Catalog receipt lacks immutable serving authority") return receipt def canonical_digest(value: dict[str, Any]) -> str: raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() return "sha256:" + hashlib.sha256(raw).hexdigest() def render_serving_config( profile: dict[str, Any], receipt: dict[str, Any], *, replica_id: str, catalog_store_root: str, ) -> dict[str, Any]: outputs = receipt.get("capabilities") if not isinstance(outputs, list) or len(outputs) != len(profile["capabilities"]): raise InstallError("Catalog receipt lacks the complete L1 capability set") by_app = { row.get("appCapabilityId"): row for row in outputs if isinstance(row, dict) } if len(by_app) != len(outputs): raise InstallError("Catalog receipt capability set is duplicate or malformed") capabilities: list[dict[str, Any]] = [] selected_runtime: set[str] = set() for selection in profile["capabilities"]: output = by_app.get(selection["appCapabilityId"]) if ( not isinstance(output, dict) or output.get("runtimeCapabilityId") != selection["runtimeCapabilityId"] or not isinstance(output.get("ownershipEpoch"), int) or output["ownershipEpoch"] <= 0 ): raise InstallError("Catalog receipt capability disagrees with the L1 install profile") require_digest( str(output.get("compatibilityReceiptDigest", "")), "Catalog capability compatibility receipt", ) runtime_capability = selection["runtimeCapabilityId"] selected_runtime.add(runtime_capability) row: dict[str, Any] = { "app_capability_id": selection["appCapabilityId"], "runtime_capability_id": runtime_capability, "authorized_workload_id": profile["handlerWorkloadId"], "expected_ownership_epoch": output["ownershipEpoch"], "expected_compatibility_receipt_digest": output["compatibilityReceiptDigest"], "admission_lease": "5s", } if runtime_capability == "runtime.records.mutate.v1": row["outcome_retention"] = "24h" capabilities.append(row) catalog_consumers = [] for grant in profile["catalogConsumerCapabilities"]: if grant["capabilityId"] in selected_runtime: raise InstallError("Catalog consumer capability duplicates a serving selection") value = { "runtime_capability_id": grant["capabilityId"], "authorized_workload_id": grant["authorizedWorkloadId"], "admission_lease": "5s", } if grant["catalogSnapshot"] == "installation": value["catalog_snapshot_digest"] = receipt["catalogSnapshotDigest"] catalog_consumers.append(value) workload_capabilities = [{ "runtime_capability_id": value["capabilityId"], "authorized_workload_id": value["authorizedWorkloadId"], } for value in profile["workloadCapabilities"]] return { "schema": "nexuscrm.runtime-api-serving/v4", "permission_state_protocol": "permission_state.v1", "replica_id": replica_id, "authorized_workload_ids": profile["authorizedWorkloadIds"], "org_id": receipt["orgId"], "app_installation_id": receipt["appInstallationId"], "catalog_snapshot_store_root": catalog_store_root, "expected_service_release_digest": receipt["serviceReleaseDigest"], "schema_generation": receipt["schemaGeneration"], "engine_generation": receipt["engineGeneration"], "identity_attempt_cap": "2s", "session_scope_capabilities": { "api": [row["runtime_capability_id"] for row in capabilities], }, "capabilities": capabilities, "catalog_consumer_capabilities": catalog_consumers, "workload_capabilities": workload_capabilities, } def write_installation_receipt( receipt: dict[str, Any], serving: dict[str, Any], path: Path, ) -> dict[str, Any]: result = { "schema": "nexuscrm.l1-installation-receipt/v1", "orgId": receipt["orgId"], "appInstallationId": receipt["appInstallationId"], "appReleaseDigest": receipt["appReleaseDigest"], "serviceReleaseDigest": receipt["serviceReleaseDigest"], "frontendReleaseDigest": receipt.get("frontendReleaseDigest"), "metadataSourceDigest": receipt["metadataSourceDigest"], "catalogRevision": receipt["catalogRevision"], "securityEpoch": receipt["securityEpoch"], "catalogSnapshotDigest": receipt["catalogSnapshotDigest"], "authorizedWorkloadIds": serving["authorized_workload_ids"], "catalogReconcileReceiptDigest": canonical_digest(receipt), "runtimeServingConfigDigest": canonical_digest(serving), } path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n") path.chmod(0o400) return result def run_checked(command: Sequence[str], *, env: dict[str, str] | None = None) -> None: completed = subprocess.run( command, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) if completed.returncode != 0: tail = "\n".join(completed.stdout.splitlines()[-40:]) raise InstallError( f"command failed with exit code {completed.returncode}: {command[0]}\n{tail}" ) def require_catalog_reconcile_image(image: str) -> str: if not CATALOG_RECONCILE_IMAGE.fullmatch(image): raise InstallError( "Catalog reconcile image must be the published Controlplane image at an immutable digest" ) return image def load_database_environment(path: Path) -> dict[str, str]: values: dict[str, str] = {} try: lines = path.read_text().splitlines() except OSError as exc: raise InstallError(f"read database environment: {exc}") from exc for number, line in enumerate(lines, start=1): if not line or line.startswith("#"): continue if "=" not in line: raise InstallError(f"database environment line {number} is malformed") key, value = line.split("=", 1) if not key or key in values: raise InstallError(f"database environment key {key!r} is invalid or duplicated") values[key] = value required = {"TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE"} missing = sorted(required - values.keys()) if missing: raise InstallError("database environment lacks " + ", ".join(missing)) return values def setup_database_environment(path: Path) -> dict[str, str]: values = load_database_environment(path) # Docker Desktop exposes the host to containers with this name. The setup # binary runs on that host, so the exact same endpoint is its loopback. if values["TIDB_HOST"] == "host.docker.internal": values["TIDB_HOST"] = "127.0.0.1" return {**os.environ, **values} def verify_pack_source(release_dir: Path, pack_source: Path, pack_name: str) -> None: manifest = load_json(pack_source / "pack.json", "pack manifest") if manifest.get("name") != pack_name: raise InstallError("pack name does not match the selected pack source") repository_root = Path(__file__).resolve().parents[2] with tempfile.TemporaryDirectory(prefix="nexuscrm-l1-pack-") as raw: rebuilt = Path(raw) / "app-package.json" command = ( "go", "run", ".", "build", "--pack-root", str(pack_source), "--output", str(rebuilt), ) completed = subprocess.run( command, cwd=repository_root / "tools" / "crm-app-release", env={**os.environ, "GOWORK": "off"}, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) if completed.returncode != 0: tail = "\n".join(completed.stdout.splitlines()[-40:]) raise InstallError(f"rebuild selected pack source: {tail}") if rebuilt.read_bytes() != (release_dir / "app-package.json").read_bytes(): raise InstallError("pack source does not match the signed AppRelease") def snapshot_file(source: Path, destination: Path, label: str, *, executable: bool = False) -> Path: if source.is_symlink() or not source.is_file(): raise InstallError(f"{label} must be a regular file") destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(source.read_bytes()) destination.chmod(0o500 if executable else 0o400) return destination def snapshot_directory(source: Path, destination: Path, label: str) -> Path: if source.is_symlink() or not source.is_dir(): raise InstallError(f"{label} must be a regular directory") for path in source.rglob("*"): if path.is_symlink(): raise InstallError(f"{label} contains a symbolic link") shutil.copytree(source, destination) return destination def snapshot_install_inputs(args: argparse.Namespace, root: Path) -> argparse.Namespace: if not args.setup_binary or not args.pack_source or not args.pack_name: raise InstallError("pack install requires --setup-binary, --pack-source, and --pack-name") snapshot = argparse.Namespace(**vars(args)) snapshot.release_dir = snapshot_directory( args.release_dir.resolve(), root / "release", "AppRelease directory", ) snapshot.profile = snapshot_file( args.profile.resolve(), root / "profile.json", "L1 install profile", ) snapshot.runtime_host_compatibility = snapshot_file( args.runtime_host_compatibility.resolve(), root / "runtime-host-compatibility.json", "Runtime host compatibility", ) snapshot.catalog_host_compatibility = snapshot_file( args.catalog_host_compatibility.resolve(), root / "catalog-host-compatibility.json", "Catalog host compatibility", ) snapshot.database_env = snapshot_file( args.database_env.resolve(), root / "database.env", "database environment", ) snapshot.setup_binary = snapshot_file( args.setup_binary.resolve(), root / "setup", "setup binary", executable=True, ) snapshot.pack_source = str(snapshot_directory( Path(args.pack_source).resolve(), root / "pack", "pack source", )) snapshot.output_config = root / "catalog-request.json" snapshot.output_receipt = root / "output" / "catalog-output.json" return snapshot def copy_protected_file(source: Path, destination: Path, label: str) -> None: if source.is_symlink() or not source.is_file(): raise InstallError(f"{label} is not a regular file") destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(source.read_bytes()) destination.chmod(0o400) def apply(args: argparse.Namespace) -> dict[str, Any]: require_catalog_reconcile_image(args.catalog_reconcile_image) with tempfile.TemporaryDirectory(prefix="nexuscrm-l1-install-") as raw: operation = snapshot_install_inputs(args, Path(raw)) # Keep the exact validated policy object for the entire operation. No # authority-bearing caller path is reopened after this point. profile = load_json(operation.profile, "L1 install profile") config = render_config(operation, profile=profile) write_config(config, operation.output_config) write_config(config, args.output_config) pack_source = Path(operation.pack_source) verify_pack_source(operation.release_dir, pack_source, operation.pack_name) setup_env = setup_database_environment(operation.database_env) for command in ( (str(operation.setup_binary), "validate-pack", "--source", str(pack_source)), (str(operation.setup_binary), "install-pack", "--source", str(pack_source)), (str(operation.setup_binary), "verify-pack", operation.pack_name), ): run_checked(command, env=setup_env) operation.catalog_store.mkdir(parents=True, exist_ok=True) operation.output_receipt.parent.mkdir(parents=True, exist_ok=True) command = ( "docker", "run", "--rm", "--network", operation.docker_network, "--user", f"{os.getuid()}:{os.getgid()}", "--env-file", str(operation.database_env), "--mount", f"type=bind,src={operation.output_config},dst=/etc/nexuscrm/catalog-config.json,readonly", "--mount", f"type=bind,src={operation.release_dir},dst=/release,readonly", "--mount", f"type=bind,src={operation.catalog_store.resolve()},dst=/var/lib/nexuscrm/catalog", "--mount", f"type=bind,src={operation.output_receipt.parent},dst=/output", operation.catalog_reconcile_image, "-config", "/etc/nexuscrm/catalog-config.json", "-output", f"/output/{operation.output_receipt.name}", ) run_checked(command) receipt = validate_receipt(config, operation.output_receipt) # Refresh the caller-visible evidence from the validated in-memory # request. It is never an execution input. write_config(config, args.output_config) copy_protected_file( operation.output_receipt, args.output_receipt, "Catalog reconcile receipt", ) serving = render_serving_config( profile, receipt, replica_id=operation.replica_id, catalog_store_root=operation.runtime_catalog_store_root, ) write_config(serving, args.serving_config_output) return write_installation_receipt( receipt, serving, args.installation_receipt_output, ) def add_common(parser: argparse.ArgumentParser) -> None: parser.add_argument("--release-dir", type=Path, required=True) parser.add_argument("--profile", type=Path, required=True) parser.add_argument("--runtime-host-compatibility", type=Path, required=True) parser.add_argument("--catalog-host-compatibility", type=Path, required=True) parser.add_argument("--org-id", required=True) parser.add_argument("--runtime-audience", required=True) parser.add_argument("--service-release-digest", required=True) parser.add_argument("--frontend-release-digest", default="") parser.add_argument("--frontend-compatibility-receipt-digest", default="") parser.add_argument("--idempotency-key", required=True) parser.add_argument("--operation-kind", choices=("PREPARE", "UPGRADE", "ROLLBACK"), default="PREPARE") parser.add_argument("--app-installation-id", default="") parser.add_argument("--expected-catalog-revision", type=int, required=True) parser.add_argument("--expected-security-epoch", type=int, required=True) parser.add_argument("--schema-generation", type=int, required=True) parser.add_argument("--engine-generation", type=int, required=True) parser.add_argument("--publisher-valid-from", default="2026-01-01T00:00:00Z") parser.add_argument("--publisher-valid-until", default="2036-01-01T00:00:00Z") parser.add_argument("--trusted-publisher-key-digest", required=True) parser.add_argument("--publication-command-id", default="") def parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser(description=__doc__) commands = root.add_subparsers(dest="command", required=True) render = commands.add_parser("render", help="verify L1 inputs and render the exact Catalog request") add_common(render) render.add_argument("--output-config", type=Path, required=True) install = commands.add_parser("install", help="install the pack and apply its Catalog authority") add_common(install) install.add_argument("--output-config", type=Path, required=True) install.add_argument("--output-receipt", type=Path, required=True) install.add_argument("--catalog-store", type=Path, required=True) install.add_argument("--database-env", type=Path, required=True) install.add_argument("--catalog-reconcile-image", required=True) install.add_argument("--docker-network", default="host") install.add_argument("--setup-binary", type=Path) install.add_argument("--pack-source") install.add_argument("--pack-name") install.add_argument("--replica-id", required=True) install.add_argument("--runtime-catalog-store-root", default="/var/lib/nexuscrm/catalog") install.add_argument("--serving-config-output", type=Path, required=True) install.add_argument("--installation-receipt-output", type=Path, required=True) return root def main() -> int: try: args = parser().parse_args() if args.command == "render": config = render_config(args) write_config(config, args.output_config) print(json.dumps({ "appReleaseDigest": config["expectedAppReleaseDigest"], "metadataSourceDigest": config["metadataSourceDigest"], "output": str(args.output_config), }, sort_keys=True)) elif args.command == "install": print(json.dumps(apply(args), sort_keys=True)) return 0 except InstallError as exc: print(f"l1-installer: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())