#!/usr/bin/env python3 from __future__ import annotations import argparse import copy import hashlib import json import re from pathlib import Path from typing import Any try: from .lib.final_pack_contract import metadata_tree_sha256 except ImportError: # pragma: no cover - direct script execution from lib.final_pack_contract import metadata_tree_sha256 SCHEMA_VERSION_V10 = "nexuscrm.serviceification-bom/v10" RESOLVED_SCHEMA_VERSION_V10 = "nexuscrm.serviceification-resolved-bom/v10" RESOLVED_SCHEMA_VERSION = RESOLVED_SCHEMA_VERSION_V10 LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION = "nexuscrm.test3-local-deployment/v1" LOCAL_TEST3_RELEASE_CLASS = "test-only-unsigned-non-promotable" LOCAL_TEST3_DEPLOYMENT_FIELDS = frozenset({ "schema_version", "release_class", "desired_bom", "desired_bom_sha256", "images", "controller_commit", }) LOCAL_TEST3_IMAGE_ROLES = frozenset({ "runtime_api", "runtime_worker", "integration_service", "catalog_reconcile", "sfcrm_app_service", "frontend", }) SHA256_RE = re.compile(r"^sha256:[0-9a-f]{64}$") COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") MODULE_VERSION_RE = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+(?:-[0-9A-Za-z.-]+)?$") L0_ENGINE_RELEASE_VERSION_RE = re.compile( r"^dev-[0-9]{8}-[0-9]+\.[0-9]+\.[0-9]+-[0-9a-f]{8}$" ) FRONTEND_SOURCE_FIELDS = ( "nexuscrm_repository", "nexuscrm_commit", "app_packs_repository", "app_packs_commit", "l2_repository", "l2_commit", ) PUBLICATION_OCI_REPOSITORIES = { "l0_engine": "ghcr.io/pingcap-inc/nexuscrm-l0-engine-release", "services.gateway": "ghcr.io/pingcap-inc/nexuscrm-api-host-gateway", "services.runtime_api": "ghcr.io/pingcap-inc/nexuscrm-runtime-api", "services.runtime_worker": "ghcr.io/pingcap-inc/nexuscrm-runtime-worker", "services.sfcrm_app_service": "ghcr.io/pingcap-inc/nexuscrm-sfcrm-appservice", "services.mcp_worker": "ghcr.io/pingcap-inc/nexuscrm-mcp-worker", "services.integration_service": "ghcr.io/pingcap-inc/nexuscrm-integration-service", "jobs.catalog_reconcile": "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile", "frontend": "ghcr.io/pingcap-inc/nexuscrm-frontend-assets", "frontend.publication": "ghcr.io/pingcap-inc/nexuscrm-frontend-assets", } MUTATION_OUTCOME_RETENTION = "24h" MUTATION_OUTCOME_CLEANUP_INTERVAL = "1h" D3_WORKLOAD_ONLY_APP_CAPABILITY_BINDINGS = { "io.nexuscrm.sfcrm.application-navigation.v1": "runtime.application-navigation.v1", "io.nexuscrm.sfcrm.approval-notification-flows.v1": "runtime.approvals.read-notification-flows.v1", "io.nexuscrm.sfcrm.metadata-admin-read.v1": "runtime.metadata.read-admin.v1", "io.nexuscrm.sfcrm.layout-authoring.v1": "runtime.layout.authoring.v2", } WORKER_ROLE_FIELDS = { "work_family", "config_schema", "batch_size", "poll_interval", "lease_duration", "retry_jitter_max", "max_attempts", "db_max_open_conns", "db_max_idle_conns", "shutdown_timeout", } # D3 deploys one Runtime Worker artifact for each current work family. Every # role uses the current v3 grammar; only record-event owns processed-row retention. D3_WORKER_ROLE_SHAPES = [ ("runtime.audit-history-fanout.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.bulk-ingest.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.file-upload-lifecycle.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.permission-authoring-command-lifecycle.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.platform-command.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.record-command-lifecycle.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.record-event-dispatch.v1", "nexuscrm.runtime-worker-role/v3", True), ("runtime.rollup-delta.v1", "nexuscrm.runtime-worker-role/v3", False), ("runtime.scheduled-flow.v1", "nexuscrm.runtime-worker-role/v3", False), ] class ValidationFailure(RuntimeError): pass def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: value: dict[str, Any] = {} for key, item in pairs: if key in value: raise ValidationFailure(f"duplicate JSON field: {key}") value[key] = item return value def exact_keys(value: Any, expected: set[str], label: str) -> list[str]: if not isinstance(value, dict): return [f"{label} must be an object"] actual = set(value) errors: list[str] = [] if missing := sorted(expected - actual): errors.append(f"{label} missing fields: {', '.join(missing)}") if unknown := sorted(actual - expected): errors.append(f"{label} has unknown fields: {', '.join(unknown)}") return errors def canonical_json_bytes(value: Any) -> bytes: return (json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n").encode() def canonical_json_sha256(value: Any) -> str: return "sha256:" + hashlib.sha256(canonical_json_bytes(value)).hexdigest() def frontend_source_tuple(value: dict[str, Any]) -> dict[str, Any]: return {field: value[field] for field in FRONTEND_SOURCE_FIELDS} def l1_service_roles(services: Any) -> set[str]: """Closed optional independent Worker role; old signed BOMs remain readable.""" return {"sfcrm_app_service"} | ({"mcp_worker"} if isinstance(services, dict) and "mcp_worker" in services else set()) def validate_mcp_service(bom: dict[str, Any]) -> list[str]: service = bom.get("services", {}).get("mcp_worker") if service is None: return [] errors = validate_publication_request( {k: service.get(k) for k in ("oci_repository", "source_repository", "source_commit", "digest")} if isinstance(service, dict) else service, "services.mcp_worker", ) if not isinstance(service, dict) or service.get("source_repository") != "pingcap-inc/nexuscrm-app-packs": errors.append("MCP Worker must be published by nexuscrm-app-packs") consumers = bom.get("catalog_consumers", {}) if "mcp-worker" in consumers.get("authorized_workload_ids", []): errors.append("MCP user API caller must be disjoint from Catalog consumers") capabilities = {c.get("runtime_capability_id") for c in bom.get("app_installation", {}).get("capabilities", []) if isinstance(c, dict)} if not {"runtime.records.query.v1", "runtime.records.mutate.v1"}.issubset(capabilities): errors.append("MCP requires installed query and mutation Runtime APIs") return errors def publication_receipt_roles(bom: dict[str, Any]) -> frozenset[str]: """Return the complete set of owner-published artifacts for one BOM. Callers must validate the BOM before using this helper. The desired BOM owns the topology: a role is required exactly when it appears as a service or job publication request, plus the L2-owned frontend artifact. Keeping this derivation here prevents candidate production, resolution and signing from maintaining divergent role lists as the topology evolves. """ services = bom.get("services") if not isinstance(services, dict): raise ValidationFailure("BOM has no publication receipt role topology") roles = set(services) | {"l0_engine", "frontend"} if not roles or any(not isinstance(role, str) or not role for role in roles): raise ValidationFailure("BOM has an invalid publication receipt role") return frozenset(roles) def validate_locator(value: Any, label: str) -> list[str]: errors = exact_keys(value, {"repository", "commit", "path", "sha256"}, label) if errors: return errors assert isinstance(value, dict) if not isinstance(value["repository"], str) or value["repository"].count("/") != 1: errors.append(f"{label}.repository must be a GitHub owner/repo") if not isinstance(value["commit"], str) or not COMMIT_RE.fullmatch(value["commit"]): errors.append(f"{label}.commit must be a full lowercase commit SHA") path = value["path"] if not isinstance(path, str) or not path or Path(path).is_absolute() or ".." in Path(path).parts: errors.append(f"{label}.path must be a non-empty repository-relative path") if not isinstance(value["sha256"], str) or not SHA256_RE.fullmatch(value["sha256"]): errors.append(f"{label}.sha256 must be a lowercase sha256 digest") return errors def validate_crm_publication_locator_owner(installation: dict[str, Any], label: str) -> list[str]: errors: list[str] = [] for field in ("artifact", "publisher_attestation", "publisher_verification_key"): locator = installation.get(field) if isinstance(locator, dict) and locator.get("repository") != "pingcap-inc/nexuscrm-app-packs": errors.append(f"{label}.{field}.repository must be pingcap-inc/nexuscrm-app-packs") return errors def validate_publication(value: Any, label: str) -> list[str]: expected = { "oci_repository", "digest", "source_repository", "source_commit", "provenance_digest", "sbom_digest", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) repository = value["oci_repository"] if not isinstance(repository, str) or not re.fullmatch(r"ghcr\.io/[a-z0-9._-]+/[a-z0-9._/-]+", repository): errors.append(f"{label}.oci_repository must be a canonical tag-free ghcr.io repository") if isinstance(repository, str) and ("@" in repository or ":" in repository.removeprefix("ghcr.io/")): errors.append(f"{label}.oci_repository must not contain a mutable tag or embedded digest") expected_repository = PUBLICATION_OCI_REPOSITORIES.get(label) if expected_repository is not None and repository != expected_repository: errors.append(f"{label}.oci_repository must be {expected_repository}") for field in ("digest", "provenance_digest", "sbom_digest"): if not isinstance(value[field], str) or not SHA256_RE.fullmatch(value[field]): errors.append(f"{label}.{field} must be a lowercase sha256 digest") source_repository = value["source_repository"] if not isinstance(source_repository, str) or source_repository.count("/") != 1: errors.append(f"{label}.source_repository must be a GitHub owner/repo") if not isinstance(value["source_commit"], str) or not COMMIT_RE.fullmatch(value["source_commit"]): errors.append(f"{label}.source_commit must be a full lowercase commit SHA") return errors def validate_publication_receipt(value: Any, label: str) -> list[str]: expected = { "oci_repository", "digest", "source_repository", "source_commit", "verification_receipt_digest", "provenance_digest", "sbom_digest", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) publication = {key: item for key, item in value.items() if key != "verification_receipt_digest"} errors.extend(validate_publication(publication, label)) receipt_digest = value["verification_receipt_digest"] if not isinstance(receipt_digest, str) or not SHA256_RE.fullmatch(receipt_digest): errors.append(f"{label}.verification_receipt_digest must be a lowercase sha256 digest") return errors def validate_l0_engine_release(value: Any, label: str) -> list[str]: errors = exact_keys( value, {"schema_version", "version", "source", "api", "subjects", "catalog_reconcile"}, label, ) if errors: return errors assert isinstance(value, dict) if value.get("schema_version") != "nexuscrm.l0-engine-release/v1": errors.append(f"{label}.schema_version must be nexuscrm.l0-engine-release/v1") version = value.get("version") if not isinstance(version, str) or not L0_ENGINE_RELEASE_VERSION_RE.fullmatch(version): errors.append(f"{label}.version must be a canonical L0 Engine release version") source = value.get("source") errors.extend(exact_keys(source, {"repository", "commit"}, f"{label}.source")) if isinstance(source, dict): if source.get("repository") != "pingcap-inc/NexusCRM": errors.append(f"{label}.source.repository must be pingcap-inc/NexusCRM") if not isinstance(source.get("commit"), str) or not COMMIT_RE.fullmatch(source["commit"]): errors.append(f"{label}.source.commit must be a full lowercase commit SHA") api = value.get("api") errors.extend(exact_keys(api, {"protocols"}, f"{label}.api")) if isinstance(api, dict): protocols = api.get("protocols") if not isinstance(protocols, list) or not protocols: errors.append(f"{label}.api.protocols must be a non-empty array") else: seen_protocols: set[str] = set() for index, protocol in enumerate(protocols): protocol_label = f"{label}.api.protocols[{index}]" errors.extend( exact_keys(protocol, {"protocolId", "major", "minor"}, protocol_label) ) if not isinstance(protocol, dict): continue protocol_id = protocol.get("protocolId") if not isinstance(protocol_id, str) or not protocol_id: errors.append(f"{protocol_label}.protocolId must be non-empty") elif protocol_id in seen_protocols: errors.append(f"{protocol_label}.protocolId must be unique") else: seen_protocols.add(protocol_id) for field in ("major", "minor"): number = protocol.get(field) if not isinstance(number, int) or isinstance(number, bool) or number < 0: errors.append(f"{protocol_label}.{field} must be a non-negative integer") subjects = value.get("subjects") expected_subjects = { "runtime-api": "ghcr.io/pingcap-inc/nexuscrm-runtime-api", "runtime-worker": "ghcr.io/pingcap-inc/nexuscrm-runtime-worker", "integration-service": "ghcr.io/pingcap-inc/nexuscrm-integration-service", } errors.extend(exact_keys(subjects, set(expected_subjects), f"{label}.subjects")) if isinstance(subjects, dict): for role, repository in expected_subjects.items(): subject = subjects.get(role) errors.extend(exact_keys(subject, {"image_repository", "image_digest"}, f"{label}.subjects.{role}")) if isinstance(subject, dict): if subject.get("image_repository") != repository: errors.append(f"{label}.subjects.{role}.image_repository must be {repository}") if not isinstance(subject.get("image_digest"), str) or not SHA256_RE.fullmatch(subject["image_digest"]): errors.append(f"{label}.subjects.{role}.image_digest must be a lowercase sha256 digest") catalog = value.get("catalog_reconcile") errors.extend(exact_keys(catalog, {"image_repository", "source_sha", "image_digest"}, f"{label}.catalog_reconcile")) if isinstance(catalog, dict): if catalog.get("image_repository") != "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile": errors.append( f"{label}.catalog_reconcile.image_repository must be " "ghcr.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile" ) if not isinstance(catalog.get("source_sha"), str) or not COMMIT_RE.fullmatch(catalog["source_sha"]): errors.append(f"{label}.catalog_reconcile.source_sha must be a full lowercase commit SHA") if not isinstance(catalog.get("image_digest"), str) or not SHA256_RE.fullmatch(catalog["image_digest"]): errors.append(f"{label}.catalog_reconcile.image_digest must be a lowercase sha256 digest") return errors def validate_l0_engine_publication(value: Any, label: str, *, receipt: bool = False) -> list[str]: expected = { "oci_repository", "digest", "source_repository", "source_commit", "provenance_digest", "sbom_digest", "release", } if receipt: expected.add("verification_receipt_digest") errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) publication = {key: value[key] for key in ( "oci_repository", "digest", "source_repository", "source_commit", "provenance_digest", "sbom_digest", )} errors.extend(validate_publication(publication, label)) if receipt: verification = value.get("verification_receipt_digest") if not isinstance(verification, str) or not SHA256_RE.fullmatch(verification): errors.append(f"{label}.verification_receipt_digest must be a lowercase sha256 digest") errors.extend(validate_l0_engine_release(value.get("release"), f"{label}.release")) release = value.get("release") if isinstance(release, dict) and isinstance(release.get("source"), dict): if release["source"].get("commit") != value.get("source_commit"): errors.append(f"{label}.release.source.commit must equal {label}.source_commit") return errors def validate_local_test3_deployment(payload: Any) -> list[str]: """Validate a private-Test3 selection, never a signed release authority. Local Docker image IDs identify the bytes on this host. They are not OCI manifest digests or substitutes for owner provenance/SBOM receipts. Host, controller checkout, image labels and embedded artifacts are verified by the explicit Test3 local deployment entrypoint before mutation. """ errors = _validate_local_test3_deployment_structure(payload) if errors: return errors return [f"desired_bom: {error}" for error in validate_v10_admission_policy(payload["desired_bom"])] def _validate_local_test3_deployment_structure(payload: Any) -> list[str]: """Validate immutable envelope facts without applying today's admission policy.""" errors = exact_keys(payload, LOCAL_TEST3_DEPLOYMENT_FIELDS, "local Test3 deployment") if errors: return errors if payload["schema_version"] != LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: errors.append(f"schema_version must be {LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION}") if payload["release_class"] != LOCAL_TEST3_RELEASE_CLASS: errors.append(f"release_class must be {LOCAL_TEST3_RELEASE_CLASS}") controller = payload["controller_commit"] if not isinstance(controller, str) or not COMMIT_RE.fullmatch(controller): errors.append("controller_commit must be a full lowercase commit SHA") desired = payload["desired_bom"] if not isinstance(desired, dict): return errors + ["desired_bom must be an object"] try: desired_errors = _validate_v10_bom_structure(desired) if not desired_errors: desired_errors.extend(validate_publication_request( desired["services"]["sfcrm_app_service"], "services.sfcrm_app_service", )) except (TypeError, ValueError): # Malformed nested JSON must fail closed, including wrong scalar types # in the shared desired validator's cross-field comparisons. desired_errors = ["invalid nested field types"] errors.extend(f"desired_bom: {error}" for error in desired_errors) if desired_errors: return errors if payload["desired_bom_sha256"] != canonical_json_sha256(desired): errors.append("desired_bom_sha256 does not match the canonical desired D3 projection") images = payload["images"] image_errors = exact_keys(images, LOCAL_TEST3_IMAGE_ROLES, "images") errors.extend(image_errors) if image_errors: return errors if "mcp_worker" in desired["services"]: return errors + ["independent MCP Worker requires the signed BOM deployment path"] frontend = desired["frontend"] runtime_image = images.get("runtime_api") core_commit = runtime_image.get("source_commit", "") if isinstance(runtime_image, dict) else "" if not isinstance(core_commit, str) or not COMMIT_RE.fullmatch(core_commit): errors.append("images.runtime_api.source_commit must be a full lowercase commit SHA") return errors if desired["l0_engine"]["version"].rsplit("-", 1)[-1] != core_commit[:8]: errors.append("desired_bom.l0_engine.version must select the runtime NexusCRM commit") expected_sources = { role: (frontend["nexuscrm_repository"], core_commit) for role in ("runtime_api", "runtime_worker", "integration_service") } expected_sources["sfcrm_app_service"] = ( desired["services"]["sfcrm_app_service"]["source_repository"], desired["services"]["sfcrm_app_service"]["source_commit"], ) expected_sources["frontend"] = (frontend["l2_repository"], frontend["l2_commit"]) for role in sorted(LOCAL_TEST3_IMAGE_ROLES): image = images[role] label = f"images.{role}" image_errors = exact_keys( image, {"image_id", "source_commit", "source_repository"}, label, ) errors.extend(image_errors) if image_errors: continue if not isinstance(image["image_id"], str) or not SHA256_RE.fullmatch(image["image_id"]): errors.append(f"{label}.image_id must be a full lowercase Docker sha256 image ID") if not isinstance(image["source_commit"], str) or not COMMIT_RE.fullmatch(image["source_commit"]): errors.append(f"{label}.source_commit must be a full lowercase commit SHA") if role == "catalog_reconcile": if image["source_repository"] != "pingcap-inc/nexuscrm-controlplane": errors.append(f"{label}.source_repository must be pingcap-inc/nexuscrm-controlplane") else: repository, commit = expected_sources[role] if image["source_repository"] != repository: errors.append(f"{label}.source_repository must equal the selected desired repository") if image["source_commit"] != commit: errors.append(f"{label}.source_commit must equal the selected desired commit") return errors def local_test3_deployment_view(payload: dict[str, Any]) -> dict[str, Any]: """Project only execution fields from an explicit unsigned local selection.""" if errors := validate_local_test3_deployment(payload): raise ValidationFailure("; ".join(errors)) return _project_local_test3_deployment(payload) def _project_local_test3_deployment(payload: dict[str, Any]) -> dict[str, Any]: result = copy.deepcopy(payload) desired = result["desired_bom"] for field in ( "app_installation", "catalog_consumers", "l2_metadata", "execution", "runtime_worker_roles", ): result[field] = copy.deepcopy(desired[field]) def image_identity(role: str) -> dict[str, str]: image = copy.deepcopy(result["images"][role]) # Existing renderers bind opaque content digests. Keep image_id explicit # so this local identity cannot be mistaken for a registry publication. return {**image, "digest": image["image_id"]} result["services"] = { role: image_identity(role) for role in ("runtime_api", "runtime_worker", "integration_service", "sfcrm_app_service") } result["jobs"] = {"catalog_reconcile": image_identity("catalog_reconcile")} source_tuple = frontend_source_tuple(desired["frontend"]) result["frontend"] = { "publication": { **image_identity("frontend"), "source_tuple_sha256": canonical_json_sha256(source_tuple), }, "source_tuple": source_tuple, "routing_mode": desired["frontend"]["routing_mode"], } return result def historical_deployment_bom_view(bom: dict[str, Any]) -> dict[str, Any]: """Read historical facts only after the caller verifies the run-owned seal. This is not candidate admission and must never be used by preparation or rendering. Preserve every historical grant and its desired digest; do not rewrite a predecessor to look like today's selected release. """ if bom.get("schema_version") != LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: return deployment_bom_view(bom) envelope = {field: bom[field] for field in LOCAL_TEST3_DEPLOYMENT_FIELDS if field in bom} if errors := _validate_local_test3_deployment_structure(envelope): raise ValidationFailure("; ".join(errors)) expected = _project_local_test3_deployment(envelope) if bom != envelope and bom != expected: raise ValidationFailure("historical local execution members disagree with the sealed selection") return expected def deployment_bom_view(bom: dict[str, Any]) -> dict[str, Any]: """Derive executable L0 members from the verified v10 Engine authority. The returned value is an operational read model. Callers must retain and persist the original resolved BOM, whose only L0 release owner is ``l0_engine``. Keeping the projection here gives every renderer and deployer one read path without recreating four independently selected release facts. """ if bom.get("schema_version") == LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: if errors := validate_deployment_bom_view(bom): raise ValidationFailure("; ".join(errors)) return local_test3_deployment_view({ field: bom[field] for field in LOCAL_TEST3_DEPLOYMENT_FIELDS }) result = copy.deepcopy(bom) if result.get("schema_version") != RESOLVED_SCHEMA_VERSION_V10: return result engine = result.get("l0_engine") release = engine.get("release") if isinstance(engine, dict) else None source = release.get("source") if isinstance(release, dict) else None subjects = release.get("subjects") if isinstance(release, dict) else None catalog = release.get("catalog_reconcile") if isinstance(release, dict) else None services = result.get("services") if not all(isinstance(value, dict) for value in ( engine, release, source, subjects, catalog, services, )): raise ValidationFailure("resolved v10 Engine release is incomplete") sfcrm = services.get("sfcrm_app_service") if not isinstance(sfcrm, dict): raise ValidationFailure("resolved v10 BOM has no sfcrm AppService publication") role_map = { "runtime_api": "runtime-api", "runtime_worker": "runtime-worker", "integration_service": "integration-service", } operational_services = {role: copy.deepcopy(services[role]) for role in l1_service_roles(services)} for service_role, subject_role in role_map.items(): subject = subjects.get(subject_role) if not isinstance(subject, dict): raise ValidationFailure( f"resolved v10 Engine release has no {subject_role} subject" ) operational_services[service_role] = { "oci_repository": subject.get("image_repository"), "digest": subject.get("image_digest"), "source_repository": source.get("repository"), "source_commit": source.get("commit"), # These receipts prove the containing Engine descriptor and keep # the current operational projection closed. They are not # independent member release claims. "provenance_digest": engine.get("provenance_digest"), "sbom_digest": engine.get("sbom_digest"), } result["services"] = operational_services result["jobs"] = { "catalog_reconcile": { "oci_repository": catalog.get("image_repository"), "digest": catalog.get("image_digest"), "source_repository": "pingcap-inc/nexuscrm-controlplane", "source_commit": catalog.get("source_sha"), "provenance_digest": engine.get("provenance_digest"), "sbom_digest": engine.get("sbom_digest"), } } return result def validate_deployment_bom_view(bom: dict[str, Any]) -> list[str]: """Validate either the sealed v10 authority or its exact execution view. Render helpers deliberately revalidate inputs. A v10 execution view is acceptable only when removing its derived members yields a valid sealed BOM and re-projecting that authority reproduces every execution field. """ if not isinstance(bom, dict): return ["BOM must be an object"] if bom.get("schema_version") == LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: envelope = { field: bom[field] for field in LOCAL_TEST3_DEPLOYMENT_FIELDS if field in bom } if errors := validate_local_test3_deployment(envelope): return errors if canonical_json_bytes(bom) == canonical_json_bytes(envelope): return [] expected = local_test3_deployment_view(envelope) if canonical_json_bytes(bom) != canonical_json_bytes(expected): return ["local Test3 execution members disagree with the explicit local selection"] return [] if bom.get("schema_version") != RESOLVED_SCHEMA_VERSION_V10: return validate_bom(bom, require_deployable=True) sealed = copy.deepcopy(bom) services = sealed.get("services") if not isinstance(services, dict) or not isinstance( services.get("sfcrm_app_service"), dict ): return ["resolved v10 execution view has no sfcrm AppService publication"] sealed["services"] = {role: copy.deepcopy(services[role]) for role in l1_service_roles(services)} sealed.pop("jobs", None) errors = validate_bom(sealed, require_deployable=True) if errors: return errors if bom == sealed: return [] try: expected = deployment_bom_view(sealed) except ValidationFailure as exc: return [str(exc)] if bom != expected: return ["resolved v10 execution members disagree with the sealed Engine release"] return [] def validate_frontend_publication(value: Any, label: str) -> list[str]: expected = { "oci_repository", "digest", "source_repository", "source_commit", "provenance_digest", "sbom_digest", "source_tuple_sha256", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) base = {key: item for key, item in value.items() if key != "source_tuple_sha256"} errors.extend(validate_publication(base, label)) if not isinstance(value["source_tuple_sha256"], str) or not SHA256_RE.fullmatch(value["source_tuple_sha256"]): errors.append(f"{label}.source_tuple_sha256 must be a lowercase sha256 digest") return errors def validate_frontend_publication_receipt(value: Any, label: str) -> list[str]: expected = { "oci_repository", "digest", "source_repository", "source_commit", "verification_receipt_digest", "provenance_digest", "sbom_digest", "source_tuple_sha256", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) publication = {key: item for key, item in value.items() if key != "verification_receipt_digest"} errors.extend(validate_frontend_publication(publication, label)) receipt_digest = value["verification_receipt_digest"] if not isinstance(receipt_digest, str) or not SHA256_RE.fullmatch(receipt_digest): errors.append(f"{label}.verification_receipt_digest must be a lowercase sha256 digest") return errors def validate_publication_request(value: Any, label: str) -> list[str]: expected = {"oci_repository", "source_repository", "source_commit"} if label == "services.mcp_worker": expected.add("digest") errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) repository = value["oci_repository"] if not isinstance(repository, str) or not re.fullmatch(r"ghcr\.io/[a-z0-9._-]+/[a-z0-9._/-]+", repository): errors.append(f"{label}.oci_repository must be a canonical tag-free ghcr.io repository") if isinstance(repository, str) and ("@" in repository or ":" in repository.removeprefix("ghcr.io/")): errors.append(f"{label}.oci_repository must not contain a mutable tag or embedded digest") expected_repository = PUBLICATION_OCI_REPOSITORIES.get(label) if expected_repository is not None and repository != expected_repository: errors.append(f"{label}.oci_repository must be {expected_repository}") source_repository = value["source_repository"] if not isinstance(source_repository, str) or source_repository.count("/") != 1: errors.append(f"{label}.source_repository must be a GitHub owner/repo") if not isinstance(value["source_commit"], str) or not COMMIT_RE.fullmatch(value["source_commit"]): errors.append(f"{label}.source_commit must be a full lowercase commit SHA") if label == "services.mcp_worker" and (not isinstance(value.get("digest"), str) or not SHA256_RE.fullmatch(value["digest"])): errors.append("services.mcp_worker.digest must select an immutable published digest") return errors def validate_l0_engine_request(value: Any, label: str) -> list[str]: """Validate an L2 selection of one owner-native L0 release version.""" expected = {"oci_repository", "version"} errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) repository = value["oci_repository"] if repository != PUBLICATION_OCI_REPOSITORIES["l0_engine"]: errors.append( f"{label}.oci_repository must be " f"{PUBLICATION_OCI_REPOSITORIES['l0_engine']}" ) version = value["version"] if not isinstance(version, str) or not L0_ENGINE_RELEASE_VERSION_RE.fullmatch(version): errors.append(f"{label}.version must be a canonical L0 Engine release version") return errors def validate_frontend_request( value: Any, label: str, routing_mode: str = "same-origin-gateway", ) -> list[str]: expected = { "oci_repository", "nexuscrm_repository", "nexuscrm_commit", "app_packs_repository", "app_packs_commit", "l2_repository", "l2_commit", "routing_mode", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) publication = { "oci_repository": value.get("oci_repository"), "source_repository": value.get("l2_repository"), "source_commit": value.get("l2_commit"), } errors.extend(validate_publication_request(publication, label)) if value.get("oci_repository") != "ghcr.io/pingcap-inc/nexuscrm-frontend-assets": errors.append(f"{label}.oci_repository must be ghcr.io/pingcap-inc/nexuscrm-frontend-assets") expected_repositories = { "nexuscrm_repository": "pingcap-inc/NexusCRM", "app_packs_repository": "pingcap-inc/nexuscrm-app-packs", "l2_repository": "pingcap-inc/nexuscrm-pingcap-pack", } for field, expected_value in expected_repositories.items(): if value.get(field) != expected_value: errors.append(f"{label}.{field} must be {expected_value}") for field in ("nexuscrm_commit", "app_packs_commit", "l2_commit"): if not isinstance(value.get(field), str) or not COMMIT_RE.fullmatch(value[field]): errors.append(f"{label}.{field} must be a full lowercase commit SHA") if value.get("routing_mode") != routing_mode: errors.append(f"{label}.routing_mode must be {routing_mode}") return errors def validate_l2_metadata(value: Any, label: str) -> list[str]: errors = exact_keys(value, {"repository", "commit", "path", "tree_sha256"}, label) if errors: return errors assert isinstance(value, dict) if value["repository"] != "pingcap-inc/nexuscrm-pingcap-pack": errors.append(f"{label}.repository must be pingcap-inc/nexuscrm-pingcap-pack") if not isinstance(value["commit"], str) or not COMMIT_RE.fullmatch(value["commit"]): errors.append(f"{label}.commit must be a full lowercase commit SHA") path = value["path"] if not isinstance(path, str) or not path or Path(path).is_absolute() or ".." in Path(path).parts: errors.append(f"{label}.path must be a non-empty repository-relative path") if not isinstance(value["tree_sha256"], str) or not SHA256_RE.fullmatch(value["tree_sha256"]): errors.append(f"{label}.tree_sha256 must be a lowercase sha256 digest") return errors def validate_capabilities( value: Any, label: str, expected: list[dict[str, str]] | None = None, set_name: str = "App handler", required: list[dict[str, str]] | None = None, ) -> list[str]: if not isinstance(value, list): return [f"{label} must be an array"] errors: list[str] = [] normalized: list[dict[str, str]] = [] for index, capability in enumerate(value): item_label = f"{label}[{index}]" item_errors = exact_keys(capability, {"capability_id", "runtime_capability_id"}, item_label) errors.extend(item_errors) if item_errors or not isinstance(capability, dict): continue app_id = capability.get("capability_id") runtime_id = capability.get("runtime_capability_id") if not isinstance(app_id, str) or not app_id: errors.append(f"{item_label}.capability_id must be non-empty") if not isinstance(runtime_id, str) or not runtime_id: errors.append(f"{item_label}.runtime_capability_id must be non-empty") if isinstance(app_id, str) and isinstance(runtime_id, str): normalized.append({"capability_id": app_id, "runtime_capability_id": runtime_id}) if normalized != sorted(normalized, key=lambda binding: binding["capability_id"]): errors.append(f"{label} must be sorted by capability_id") capability_ids = [binding["capability_id"] for binding in normalized] if len(set(capability_ids)) != len(capability_ids): errors.append(f"{label} must not contain duplicate capability_id values") if expected is not None and normalized != expected: errors.append(f"{label} must be the complete {set_name} capability set") if required is not None: by_capability = { binding["capability_id"]: binding for binding in normalized } missing = [ binding for binding in required if by_capability.get(binding["capability_id"]) != binding ] if missing: errors.append( f"{label} is missing required {set_name} route bindings: {missing!r}" ) return errors def _validate_d3_catalog_consumer_structure(value: Any, label: str) -> list[str]: expected = { "handler_workload_id", "integration_workload_id", "authorized_workload_ids", "catalog_view_grants", "runtime_capability_grants", } errors = exact_keys(value, expected, label) if errors: return errors assert isinstance(value, dict) if value.get("handler_workload_id") != "sfcrm-app-service": errors.append(f"{label}.handler_workload_id must be sfcrm-app-service") if value.get("integration_workload_id") != "integration-service": errors.append(f"{label}.integration_workload_id must be integration-service") if value.get("authorized_workload_ids") != ["integration-service", "sfcrm-app-service"]: errors.append( f"{label}.authorized_workload_ids must contain the exact sorted D3 consumer set" ) for field in ("catalog_view_grants", "runtime_capability_grants"): grants = value.get(field) if not isinstance(grants, list) or not grants: errors.append(f"{label}.{field} must be a non-empty grant array") continue identities = [] for grant in grants: shape = exact_keys(grant, {"authorized_workload_id", "capability_id"}, f"{label}.{field}") errors.extend(shape) if shape: continue workload, capability = grant["authorized_workload_id"], grant["capability_id"] if ( workload not in ("integration-service", "sfcrm-app-service") or not isinstance(capability, str) or not capability or capability.strip() != capability ): errors.append(f"{label}.{field} contains an invalid grant identity") continue identities.append((workload, capability)) if identities != sorted(set(identities)): errors.append(f"{label}.{field} must contain sorted unique grants") return errors def validate_worker_roles( value: Any, label: str, shapes: list[tuple[str, str, bool]] = D3_WORKER_ROLE_SHAPES, allow_record_event_parallelism: bool = True, ) -> list[str]: """Validate the exact current worker-owner set before deployment rendering. Only the record-event consumer owns processed-row retention. The other families must not claim it: retention is a durable-work policy, not an interchangeable worker tuning knob. """ if not isinstance(value, list): return [f"{label} must be an array"] # Existing signed releases remain consumable. Permission cutover explicitly # requires the ninth role at its Test2 application-preparation boundary. if shapes == D3_WORKER_ROLE_SHAPES and len(value) == len(shapes) - 1: shapes = [shape for shape in shapes if shape[0] != "runtime.permission-authoring-command-lifecycle.v1"] errors: list[str] = [] if len(value) != len(shapes): errors.append(f"{label} must contain every required work family exactly once") for index, (work_family, config_schema, owns_processed_retention) in enumerate(shapes): item_label = f"{label}[{index}]" if index >= len(value): break role = value[index] expected_fields = set(WORKER_ROLE_FIELDS) if owns_processed_retention: expected_fields.update({"processed_retention", "processed_cleanup_interval"}) parallel_record_events = ( allow_record_event_parallelism and work_family == "runtime.record-event-dispatch.v1" and config_schema == "nexuscrm.runtime-worker-role/v3" ) # Core owns source-partition exclusion and keyed effect fences. The # explicit lane policy is optional so existing signed serial releases # retain their exact grammar and default-one behavior. parallel_platform_commands = ( work_family == "runtime.platform-command.v1" and config_schema == "nexuscrm.runtime-worker-role/v3" and isinstance(role, dict) and "parallelism" in role ) parallel_work = parallel_record_events or parallel_platform_commands if parallel_work: expected_fields.add("parallelism") errors.extend(exact_keys(role, expected_fields, item_label)) if not isinstance(role, dict): continue if role.get("work_family") != work_family or role.get("config_schema") != config_schema: errors.append(f"{item_label} must bind {work_family} to {config_schema}") for field, lower, upper in ( ("batch_size", 1, 100), ("max_attempts", 1, 20), ("db_max_open_conns", 2, 32), ): candidate = role.get(field) if not isinstance(candidate, int) or isinstance(candidate, bool) or not lower <= candidate <= upper: errors.append(f"{item_label}.{field} is outside the Runtime-enforced range {lower}..{upper}") idle, opened = role.get("db_max_idle_conns"), role.get("db_max_open_conns") if ( not isinstance(idle, int) or isinstance(idle, bool) or idle <= 0 or not isinstance(opened, int) or isinstance(opened, bool) or idle > opened ): errors.append(f"{item_label}.db_max_idle_conns must be positive and no greater than db_max_open_conns") if parallel_work: parallelism = role.get("parallelism") if not isinstance(parallelism, int) or isinstance(parallelism, bool) or not 1 <= parallelism <= 15: errors.append(f"{item_label}.parallelism is outside the Runtime-enforced range 1..15") elif isinstance(opened, int) and opened < 2 * parallelism + 1: errors.append( f"{item_label}.db_max_open_conns must reserve two connections per lane plus the controller" ) for field in ("poll_interval", "lease_duration", "retry_jitter_max", "shutdown_timeout"): if not isinstance(role.get(field), str) or not role[field]: errors.append(f"{item_label}.{field} must be a non-empty Runtime duration") retention_policy = { "processed_retention": MUTATION_OUTCOME_RETENTION, "processed_cleanup_interval": MUTATION_OUTCOME_CLEANUP_INTERVAL, } for field, required_value in retention_policy.items(): if owns_processed_retention: if role.get(field) != required_value: errors.append( f"{item_label}.{field} must equal the reviewed outbound-intent policy {required_value}" ) elif field in role: errors.append(f"{item_label} must not claim processed-row retention") return errors def validate_v10_application(bom: dict[str, Any]) -> list[str]: """Validate the current App, tenant, workload, and execution selections.""" errors = _validate_v10_application_structure(bom) return errors + validate_v10_admission_policy(bom) def _validate_v10_application_structure(bom: dict[str, Any]) -> list[str]: errors: list[str] = [] frontend = bom.get("frontend") errors.extend(validate_frontend_request(frontend, "frontend", "direct-sfcrm")) installation = bom.get("app_installation") installation_fields = { "app_id", "semantic_version", "app_release_digest", "metadata_bundle_digest", "artifact", "publisher_attestation", "publisher_verification_key", "capabilities", } errors.extend(exact_keys(installation, installation_fields, "app_installation")) if isinstance(installation, dict): if installation.get("app_id") != "io.nexuscrm.sfcrm": errors.append("app_installation.app_id must be io.nexuscrm.sfcrm") if not isinstance(installation.get("semantic_version"), str) or not installation["semantic_version"]: errors.append("app_installation.semantic_version must be non-empty") for field in ("app_release_digest", "metadata_bundle_digest"): if not isinstance(installation.get(field), str) or not SHA256_RE.fullmatch(installation[field]): errors.append(f"app_installation.{field} must be a lowercase sha256 digest") for field in ("artifact", "publisher_attestation", "publisher_verification_key"): errors.extend(validate_locator(installation.get(field), f"app_installation.{field}")) errors.extend(validate_crm_publication_locator_owner(installation, "app_installation")) locators = [installation.get(field) for field in ("artifact", "publisher_attestation", "publisher_verification_key")] if all(isinstance(locator, dict) for locator in locators): if len({locator.get("commit") for locator in locators}) != 1: errors.append("app_installation publication files must resolve from one immutable revision") # This validator owns the installation projection's wire shape. The # immutable AppPackage owns the actual capability set, which # verify_app_packs_artifacts compares exactly. Copying today's set here # made an older signed release depend on mutable current-main policy. errors.extend(validate_capabilities( installation.get("capabilities"), "app_installation.capabilities", )) catalog_consumers = bom.get("catalog_consumers") errors.extend(_validate_d3_catalog_consumer_structure(catalog_consumers, "catalog_consumers")) errors.extend(validate_l2_metadata(bom.get("l2_metadata"), "l2_metadata")) errors.extend(validate_worker_roles( bom.get("runtime_worker_roles"), "runtime_worker_roles", D3_WORKER_ROLE_SHAPES, True, )) expected_execution = { "catalog_reconcile_config_schema": "nexuscrm.catalog-reconcile-config/v4", "catalog_reconcile_output_schema": "nexuscrm.catalog-reconcile-output/v3", "runtime_serving_config_schema": "nexuscrm.runtime-api-serving/v4", "runtime_worker_config_schemas": ["nexuscrm.runtime-worker-role/v3"], "runtime_work_activation_schema": "nexuscrm.runtime-work-ownership-activation/v1", "topology": "sfcrm_app_service>runtime_api>tidb;integration_service>runtime_api>tidb;runtime_worker>tidb" + (";mcp_worker>runtime_api" if "mcp_worker" in bom.get("services", {}) else ""), } execution = bom.get("execution") errors.extend(exact_keys(execution, set(expected_execution), "execution")) if isinstance(execution, dict): for field, expected in expected_execution.items(): if execution.get(field) != expected: errors.append(f"execution.{field} must be {expected}") return errors def validate_v10_admission_policy(bom: dict[str, Any]) -> list[str]: """Validate admission structure; the release BOM owns its grant values.""" installation = bom.get("app_installation") catalog_consumers = bom.get("catalog_consumers") errors = _validate_d3_catalog_consumer_structure(catalog_consumers, "catalog_consumers") if isinstance(installation, dict) and isinstance(catalog_consumers, dict): if not isinstance(installation.get("capabilities"), list) or not isinstance( catalog_consumers.get("runtime_capability_grants"), list, ): return errors selected_runtime_capabilities = { binding.get("runtime_capability_id") for binding in installation.get("capabilities", []) if isinstance(binding, dict) and isinstance(binding.get("runtime_capability_id"), str) } granted_runtime_capabilities = { grant.get("capability_id") for grant in catalog_consumers.get("runtime_capability_grants", []) if isinstance(grant, dict) and isinstance(grant.get("capability_id"), str) } # Runtime keys admission by capability, not by (workload, capability). grants = [grant["capability_id"] for grant in catalog_consumers["runtime_capability_grants"] if isinstance(grant, dict) and isinstance(grant.get("capability_id"), str)] if len(grants) != len(set(grants)): errors.append("Runtime capability grants must have one workload per capability") workload_only_selections = sorted( selected_runtime_capabilities & set(D3_WORKLOAD_ONLY_APP_CAPABILITY_BINDINGS.values()) ) if workload_only_selections: errors.append( "Workload-only capabilities cannot be selected through Catalog; " "publish a corrected BOM with explicit Runtime grants: " + ", ".join(workload_only_selections) ) duplicate_runtime_capabilities = sorted( selected_runtime_capabilities & granted_runtime_capabilities ) if duplicate_runtime_capabilities: errors.append( "AppInstallation capabilities and explicit Runtime grants must be disjoint: " + ", ".join(duplicate_runtime_capabilities) ) return errors def validate_v10_bom(bom: dict[str, Any]) -> list[str]: """Validate the D3 tuple that selects one owner-signed L0 Engine release.""" errors = _validate_v10_bom_structure(bom) return errors + validate_v10_admission_policy(bom) def _validate_v10_bom_structure(bom: dict[str, Any]) -> list[str]: expected = { "schema_version", "l0_engine", "services", "frontend", "app_installation", "catalog_consumers", "l2_metadata", "execution", "runtime_worker_roles", } errors = exact_keys(bom, expected, "BOM") if errors: return errors if bom.get("schema_version") != SCHEMA_VERSION_V10: errors.append(f"schema_version must be {SCHEMA_VERSION_V10}") engine = bom.get("l0_engine") errors.extend(validate_l0_engine_request(engine, "l0_engine")) services = bom.get("services") errors.extend(exact_keys(services, l1_service_roles(services), "services")) if not isinstance(engine, dict) or not isinstance(services, dict): return errors errors.extend(_validate_v10_application_structure(bom)) if not errors: errors.extend(validate_mcp_service(bom)) if "mcp_worker" in services: errors.extend(validate_publication_request(services["mcp_worker"], "services.mcp_worker")) return errors def desired_v10_projection_from_resolved(bom: dict[str, Any]) -> dict[str, Any]: if bom.get("schema_version") == LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: if errors := validate_deployment_bom_view(bom): raise ValidationFailure("; ".join(errors)) return copy.deepcopy(bom["desired_bom"]) def request(publication: dict[str, Any], role: str) -> dict[str, Any]: fields = ("oci_repository", "source_repository", "source_commit") + (("digest",) if role == "mcp_worker" else ()) return {field: publication[field] for field in fields} frontend_publication = bom["frontend"]["publication"] return { "schema_version": SCHEMA_VERSION_V10, "l0_engine": { "oci_repository": bom["l0_engine"]["oci_repository"], "version": bom["l0_engine"]["release"]["version"], }, "services": { **{role: request(bom["services"][role], role) for role in l1_service_roles(bom["services"])}, }, "frontend": { "oci_repository": frontend_publication["oci_repository"], **bom["frontend"]["source_tuple"], "routing_mode": bom["frontend"]["routing_mode"], }, "app_installation": copy.deepcopy(bom["app_installation"]), "catalog_consumers": copy.deepcopy(bom["catalog_consumers"]), "l2_metadata": copy.deepcopy(bom["l2_metadata"]), "execution": copy.deepcopy(bom["execution"]), "runtime_worker_roles": copy.deepcopy(bom["runtime_worker_roles"]), } def validate_resolved_bom_v10(bom: dict[str, Any]) -> list[str]: expected = { "schema_version", "desired_bom_sha256", "l0_engine", "services", "frontend", "app_installation", "catalog_consumers", "l2_metadata", "execution", "runtime_worker_roles", } errors = exact_keys(bom, expected, "resolved BOM") if errors: return errors if bom.get("schema_version") != RESOLVED_SCHEMA_VERSION_V10: errors.append(f"schema_version must be {RESOLVED_SCHEMA_VERSION_V10}") desired_digest = bom.get("desired_bom_sha256") if not isinstance(desired_digest, str) or not SHA256_RE.fullmatch(desired_digest): errors.append("desired_bom_sha256 must be a lowercase sha256 digest") errors.extend(validate_l0_engine_publication(bom.get("l0_engine"), "l0_engine")) services = bom.get("services") errors.extend(exact_keys(services, l1_service_roles(services), "services")) if isinstance(services, dict): for role in l1_service_roles(services): errors.extend(validate_publication(services.get(role), "services." + role)) frontend = bom.get("frontend") errors.extend(exact_keys(frontend, {"publication", "source_tuple", "routing_mode"}, "frontend")) if isinstance(frontend, dict): errors.extend(validate_frontend_publication(frontend.get("publication"), "frontend.publication")) if frontend.get("routing_mode") != "direct-sfcrm": errors.append("frontend.routing_mode must be direct-sfcrm") if not errors: desired = desired_v10_projection_from_resolved(bom) errors.extend(validate_v10_bom(desired)) if bom.get("desired_bom_sha256") != canonical_json_sha256(desired): errors.append("desired_bom_sha256 does not match the canonical desired D3 projection") return errors def validate_bom( bom: Any, *, require_desired: bool = False, require_deployable: bool = False, ) -> list[str]: if not isinstance(bom, dict): return ["BOM must be an object"] schema = bom.get("schema_version") if schema == SCHEMA_VERSION_V10: errors = validate_v10_bom(bom) if require_deployable: errors.append( f"deployable test stack requires {RESOLVED_SCHEMA_VERSION}; desired v10 has no resolved artifact digests" ) return errors if schema == RESOLVED_SCHEMA_VERSION_V10: errors = validate_resolved_bom_v10(bom) if require_desired: errors.append(f"desired test stack input requires {SCHEMA_VERSION_V10}, not a resolved output") return errors return ["unsupported serviceification BOM schema"] def resolve_exact_app(bom: dict[str, Any], app_id: str) -> dict[str, Any]: installation = bom.get("app_installation") matches = [installation] if isinstance(installation, dict) and installation.get("app_id") == app_id else [] if len(matches) != 1: raise ValidationFailure(f"BOM resolves {app_id} to {len(matches)} AppRelease entries, want exactly one") return matches[0] def sha256_file(path: Path) -> str: return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() def is_canonical_nonempty_string(value: Any) -> bool: return isinstance(value, str) and bool(value) and value.strip() == value def app_package_capabilities(package: dict[str, Any]) -> list[dict[str, str]]: """Read the complete service-handler facts from their L1 AppPackage owner.""" manifest = package.get("canonical_manifest") provided = manifest.get("provided_capabilities") if isinstance(manifest, dict) else None if not isinstance(provided, list): raise ValidationFailure("immutable AppPackage has no provided_capabilities") bindings: list[dict[str, str]] = [] seen: set[str] = set() for capability in provided: required = capability.get("required_runtime_capabilities") if isinstance(capability, dict) else None capability_id = capability.get("capability_id") if isinstance(capability, dict) else None handler_ref = capability.get("handler_ref") if isinstance(capability, dict) else None if ( not is_canonical_nonempty_string(capability_id) or not is_canonical_nonempty_string(handler_ref) or not isinstance(required, list) or len(required) != 1 or not is_canonical_nonempty_string(required[0]) ): raise ValidationFailure("immutable AppPackage has a malformed capability binding") if capability_id in seen: raise ValidationFailure( f"immutable AppPackage has duplicate capability {capability_id!r}" ) seen.add(capability_id) bindings.append({ "capability_id": capability_id, "handler_ref": handler_ref, "runtime_capability_id": required[0], }) if bindings != sorted(bindings, key=lambda binding: binding["capability_id"]): raise ValidationFailure( "immutable AppPackage capabilities are not sorted by capability_id" ) return bindings def app_package_handler_dependencies(package: dict[str, Any]) -> dict[str, set[str]]: manifest = package.get("canonical_manifest") provided = manifest.get("provided_capabilities") if isinstance(manifest, dict) else None if not isinstance(provided, list): raise ValidationFailure("immutable AppPackage has no provided_capabilities") result: dict[str, set[str]] = {} for capability in provided: capability_id = capability.get("capability_id") if isinstance(capability, dict) else None handler_ref = capability.get("handler_ref") if isinstance(capability, dict) else None required = capability.get("required_runtime_capabilities") if isinstance(capability, dict) else None if ( not is_canonical_nonempty_string(capability_id) or not is_canonical_nonempty_string(handler_ref) or not isinstance(required, list) or not required or required != sorted(set(required)) or any(not is_canonical_nonempty_string(item) for item in required) or capability_id in result ): raise ValidationFailure("immutable AppPackage has a malformed handler dependency closure") result[capability_id] = set(required) return result def app_package_composite_capability_selections( package: dict[str, Any], selections: Any, workload_only_app_capability_bindings: dict[str, str] | None = None, ) -> list[dict[str, str]]: """Bind each App handler to one selected serving capability. The immutable AppPackage owns the handler's complete dependency closure. A D3 Catalog serving selection names one member of that closure; the remaining dependencies are checked by Catalog admission rather than flattened into synthetic one-RPC-per-handler AppPackage facts. """ dependencies = app_package_handler_dependencies(package) provided = package["canonical_manifest"]["provided_capabilities"] handlers = { capability["capability_id"]: capability["handler_ref"] for capability in provided } if not isinstance(selections, list): raise ValidationFailure("D3 capability selection must be a list") normalized: list[dict[str, str]] = [] seen_app: set[str] = set() seen_runtime: set[str] = set() for selection in selections: if not isinstance(selection, dict) or set(selection) != { "capability_id", "runtime_capability_id", }: raise ValidationFailure("D3 capability selection has an unknown shape") capability_id = selection.get("capability_id") runtime_id = selection.get("runtime_capability_id") if ( not is_canonical_nonempty_string(capability_id) or not is_canonical_nonempty_string(runtime_id) or capability_id in seen_app or runtime_id in seen_runtime or runtime_id not in dependencies.get(capability_id, set()) ): raise ValidationFailure( "D3 capability selection does not match the immutable AppPackage" ) seen_app.add(capability_id) seen_runtime.add(runtime_id) normalized.append({ "capability_id": capability_id, "handler_ref": handlers[capability_id], "runtime_capability_id": runtime_id, }) workload_only = workload_only_app_capability_bindings or {} available_runtime = seen_runtime | set(workload_only.values()) uncovered = { capability_id for capability_id, required in dependencies.items() if capability_id not in seen_app and ( # Only an explicitly named App capability may remain outside # Catalog's serving projection. The mapping binds that handler to # one reviewed workload grant; sharing another runtime dependency # (for example session-lifecycle) never creates an exemption. capability_id not in workload_only or workload_only[capability_id] not in required or not required.issubset(available_runtime) ) } if uncovered: raise ValidationFailure( "D3 capability selection does not cover every immutable App handler" ) return normalized def verify_app_packs_artifacts(bom: dict[str, Any], app_packs_root: Path) -> list[str]: errors: list[str] = [] installation = bom.get("app_installation") apps = [installation] if isinstance(installation, dict) else [] for index, app in enumerate(apps): if not isinstance(app, dict): continue for field in ("artifact", "publisher_attestation", "publisher_verification_key"): locator = app.get(field) if not isinstance(locator, dict) or not isinstance(locator.get("path"), str): continue path = app_packs_root / locator["path"] if not path.is_file(): errors.append(f"apps[{index}].{field} not found: {path}") continue actual = sha256_file(path) if actual != locator.get("sha256"): errors.append(f"apps[{index}].{field} digest mismatch: expected {locator.get('sha256')}, got {actual}") artifact = app.get("artifact") if not isinstance(artifact, dict): continue package_path = app_packs_root / str(artifact.get("path", "")) if not package_path.is_file(): continue try: package = json.loads(package_path.read_text()) except (OSError, json.JSONDecodeError) as exc: errors.append(f"apps[{index}].artifact is not valid JSON: {exc}") continue manifest = package.get("canonical_manifest") if isinstance(package, dict) else None if not isinstance(manifest, dict): errors.append(f"apps[{index}].artifact has no canonical_manifest") continue comparisons = { "app_id": manifest.get("app_id"), "semantic_version": manifest.get("semantic_version"), "app_release_digest": package.get("app_release_digest"), "metadata_bundle_digest": manifest.get("metadata_bundle_digest"), } for field, actual in comparisons.items(): if actual != app.get(field): errors.append(f"apps[{index}] {field} disagrees with immutable AppPackage: BOM={app.get(field)!r}, package={actual!r}") try: granted_workload_capabilities = { grant["capability_id"] for grant in bom["catalog_consumers"]["runtime_capability_grants"] if grant["authorized_workload_id"] == bom["catalog_consumers"]["handler_workload_id"] } app_package_composite_capability_selections( package, app.get("capabilities"), { app_capability_id: runtime_capability_id for app_capability_id, runtime_capability_id in D3_WORKLOAD_ONLY_APP_CAPABILITY_BINDINGS.items() if runtime_capability_id in granted_workload_capabilities }, ) except ValidationFailure as exc: errors.append(f"apps[{index}] {exc}") attestation_locator = app.get("publisher_attestation") key_locator = app.get("publisher_verification_key") if not isinstance(attestation_locator, dict) or not isinstance(key_locator, dict): continue attestation_path = app_packs_root / str(attestation_locator.get("path", "")) key_path = app_packs_root / str(key_locator.get("path", "")) if not attestation_path.is_file() or not key_path.is_file(): continue try: attestation = json.loads(attestation_path.read_text()) verification_key = json.loads(key_path.read_text()) except (OSError, json.JSONDecodeError) as exc: errors.append(f"apps[{index}] publisher proof is not valid JSON: {exc}") continue if not isinstance(attestation, dict) or not isinstance(verification_key, dict): errors.append(f"apps[{index}] publisher proof roots must be objects") continue proof_comparisons = { "app_id": (attestation.get("app_id"), app.get("app_id")), "semantic_version": (attestation.get("semantic_version"), app.get("semantic_version")), "app_release_digest": (attestation.get("app_release_digest"), app.get("app_release_digest")), "publisher_id": (verification_key.get("publisher_id"), manifest.get("publisher_id")), "key_id": (verification_key.get("key_id"), attestation.get("key_id")), "algorithm": (verification_key.get("algorithm"), attestation.get("signature_algorithm")), } for field, (actual, expected) in proof_comparisons.items(): if actual != expected: errors.append(f"apps[{index}] publisher proof {field} mismatch: got {actual!r}, want {expected!r}") return errors def parse_bom(serialized: bytes, label: str) -> dict[str, Any]: """Parse exactly the bytes a caller read or verified, rejecting duplicate keys.""" try: value = json.loads(serialized, object_pairs_hook=reject_duplicate_keys) except (UnicodeDecodeError, json.JSONDecodeError, ValidationFailure) as exc: raise ValidationFailure(f"cannot read BOM {label}: {exc}") from exc if not isinstance(value, dict): raise ValidationFailure("BOM root must be an object") return value def load_bom(path: Path) -> dict[str, Any]: try: serialized = path.read_bytes() except OSError as exc: raise ValidationFailure(f"cannot read BOM {path}: {exc}") from exc return parse_bom(serialized, str(path)) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--bom", default="release/serviceification-d3-bom.json") parser.add_argument("--app-packs-root", default="") parser.add_argument("--require-desired", action="store_true") parser.add_argument("--require-deployable", action="store_true") args = parser.parse_args() path = Path(args.bom).resolve() try: bom = load_bom(path) except ValidationFailure as exc: print(exc) return 1 if args.require_desired and args.require_deployable: print("--require-desired and --require-deployable are mutually exclusive") return 1 errors = validate_bom( bom, require_desired=args.require_desired, require_deployable=args.require_deployable, ) if args.app_packs_root and not errors: errors.extend(verify_app_packs_artifacts(bom, Path(args.app_packs_root).resolve())) if errors: for error in errors: print(error) return 1 print(f"serviceification BOM ok: {path}") return 0 if __name__ == "__main__": raise SystemExit(main())