#!/usr/bin/env python3 """Render current serviceification inputs from one resolved BOM and Catalog output.""" from __future__ import annotations import argparse import hashlib import json import re import stat import sys import unicodedata from pathlib import Path from typing import Any from urllib.parse import urlparse SCRIPT_DIRECTORY = Path(__file__).resolve().parent if str(SCRIPT_DIRECTORY) not in sys.path: sys.path.insert(0, str(SCRIPT_DIRECTORY)) from validate_serviceification_bom import ( # noqa: E402 D3_WORKLOAD_ONLY_APP_CAPABILITY_BINDINGS, LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION, LOCAL_TEST3_RELEASE_CLASS, RESOLVED_SCHEMA_VERSION_V10, MUTATION_OUTCOME_RETENTION, SHA256_RE, ValidationFailure, app_package_capabilities, app_package_composite_capability_selections, deployment_bom_view, load_bom, reject_duplicate_keys, validate_bom, validate_deployment_bom_view, verify_app_packs_artifacts, ) RUNTIME_HOST_COMPATIBILITY_SCHEMA = "nexuscrm.runtime-host-compatibility/v1" CATALOG_HOST_COMPATIBILITY_SCHEMA = "nexuscrm.catalog-host-compatibility/v1" CATALOG_RECONCILE_CONFIG_SCHEMA_V4 = "nexuscrm.catalog-reconcile-config/v4" FRONTEND_RECEIPT_SCHEMA = "nexuscrm.frontend-compatibility-receipt/v1" LOCAL_TEST3_FRONTEND_RECEIPT_SCHEMA = "nexuscrm.test3-local-frontend-compatibility-receipt/v1" MUTATION_RUNTIME_CAPABILITY = "runtime.records.mutate.v1" PERMISSION_AUTHORING_RUNTIME_CAPABILITY = "runtime.permissions.authoring.v1" # Keep the Permission owner’s seven-day command replay window explicit at # deployment. Records retains its independently owned 24-hour policy. PERMISSION_AUTHORING_OUTCOME_RETENTION = "168h" D3_INTEGRATION_BINDINGS_SCHEMA = "nexuscrm.serviceification-d3-integration-bindings/v1" D3_FEDERATED_OIDC_SCHEMA = "nexuscrm.serviceification-d3-federated-oidc/v1" SESSION_LIFECYCLE_CAPABILITY = "runtime.identity.session-lifecycle.v1" MAILBOX_AUTHORIZATION_CAPABILITY = "runtime.connected-mailboxes.authorize.v1" APPROVAL_NOTIFICATION_FLOW_READ_CAPABILITY = "runtime.approvals.read-notification-flows.v1" METADATA_ADMIN_READ_CAPABILITY = "runtime.metadata.read-admin.v1" PAGE_LAYOUT_RESOLUTION_CAPABILITY = "runtime.metadata.resolve-page-layout.v1" LAYOUT_AUTHORING_V2_CAPABILITY = "runtime.layout.authoring.v2" APPLICATION_NAVIGATION_CAPABILITY = "runtime.application-navigation.v1" INTEGRATION_BINDING_CAPABILITY = "catalog.consumer.integration-binding-view.v1" D3_DELEGATED_SESSION_SCOPE_CAPABILITIES = ( MAILBOX_AUTHORIZATION_CAPABILITY, APPROVAL_NOTIFICATION_FLOW_READ_CAPABILITY, METADATA_ADMIN_READ_CAPABILITY, LAYOUT_AUTHORING_V2_CAPABILITY, APPLICATION_NAVIGATION_CAPABILITY, PAGE_LAYOUT_RESOLUTION_CAPABILITY, ) D3_WORKLOAD_ADMISSION_CAPABILITIES = frozenset({ SESSION_LIFECYCLE_CAPABILITY, MAILBOX_AUTHORIZATION_CAPABILITY, APPROVAL_NOTIFICATION_FLOW_READ_CAPABILITY, METADATA_ADMIN_READ_CAPABILITY, LAYOUT_AUTHORING_V2_CAPABILITY, APPLICATION_NAVIGATION_CAPABILITY, }) D3_BOUNDED_WORKLOAD_ADMISSION_CAPABILITIES = ( D3_WORKLOAD_ADMISSION_CAPABILITIES - {SESSION_LIFECYCLE_CAPABILITY} ) # These grants are served only through the AppRelease capabilities published by # Catalog, with ownership epochs and compatibility receipts. They are not # Catalog-consumer selections, even when an older AppRelease omits them. D3_SERVING_ADMISSION_CAPABILITIES = frozenset({ PERMISSION_AUTHORING_RUNTIME_CAPABILITY, }) # Engine feature support is AppRelease dependency authority, not an additional # selectable Runtime service. Permission authoring owns this feature's calls. D3_RUNTIME_FEATURE_CAPABILITIES = frozenset({ "runtime.permissions.authoring.record-type-object-groups.v1", }) D3_SNAPSHOT_BOUND_CATALOG_CONSUMER_CAPABILITIES = frozenset({ "runtime.commands.accept.v1", "runtime.outbound-intents.consume.v1", PAGE_LAYOUT_RESOLUTION_CAPABILITY, }) # Hosted Test2 reaches its TiDB over the deployment network. Keep the bounds # explicit and ordered: identity acquisition survives a remote SQL round trip, # a client attempt ends before the Runtime lease, and the outer HTTP request # owns the whole journey. These values do not change Runtime's global admission # cap. D3_IDENTITY_ATTEMPT_CAP = "8s" D3_RUNTIME_ATTEMPT_BUDGET = "30s" D3_CAPABILITY_ADMISSION_LEASE = "35s" def d3_session_scope_capabilities( installed_runtime_capabilities: list[str], runtime_capability_grants: list[dict[str, Any]], ) -> list[str]: """Include delegated workload grants in an API user's session scope.""" capabilities = list(installed_runtime_capabilities) for delegated_capability in D3_DELEGATED_SESSION_SCOPE_CAPABILITIES: if any( grant.get("capability_id") == delegated_capability for grant in runtime_capability_grants ) and delegated_capability not in capabilities: capabilities.append(delegated_capability) return capabilities # The L1 external.records.read contract exposes record query/read and object # describe/list. Describe reads effective permissions and Runtime-owned list # fields through delegated layout resolution. Catalog metadata itself is # obtained through the AppService workload authority. # Keep an explicit allowlist: newly installed capabilities must not silently # expand a public client's authority (including other read/admin surfaces). D3_PUBLIC_RECORDS_READ_CAPABILITIES = ( "runtime.records.query.v1", "runtime.records.read-record-view.v1", "runtime.permissions.read-effective.v1", "runtime.metadata.resolve-page-layout.v1", ) def d3_session_scope_policy( installed_runtime_capabilities: list[str], runtime_capability_grants: list[dict[str, Any]], ) -> dict[str, list[str]]: api = d3_session_scope_capabilities( installed_runtime_capabilities, runtime_capability_grants, ) policy = {"api": api} # Only the SFCRM workload can delegate its read surface to a public CRM # session. Keep the API policy unchanged and intersect with the allowlist. # Delegated capabilities require the SFCRM grant even if an AppRelease # also lists them as installed; installation alone is not delegation. public_available = d3_session_scope_capabilities( [capability for capability in installed_runtime_capabilities if capability not in D3_DELEGATED_SESSION_SCOPE_CAPABILITIES], [grant for grant in runtime_capability_grants if grant.get("authorized_workload_id") == "sfcrm-app-service"], ) public_reads = [ capability for capability in D3_PUBLIC_RECORDS_READ_CAPABILITIES if capability in public_available ] if public_reads: policy["external.records.read"] = public_reads # Confidential api.read sessions need the same record-read closure. # This is not an alias to api: never delegate mutation, authoring, or # unrelated capabilities just because the Connected App can mint it. policy["api.read"] = list(public_reads) return policy def d3_workload_capabilities( grants: list[dict[str, str]], admission_lease: str, *, application_owner_id: str = "", ) -> list[dict[str, str]]: """Render workload-only Runtime admission without inventing Catalog state.""" capabilities: list[dict[str, str]] = [] for grant in grants: capability = { "runtime_capability_id": grant["capability_id"], "authorized_workload_id": grant["authorized_workload_id"], } if grant["capability_id"] in D3_BOUNDED_WORKLOAD_ADMISSION_CAPABILITIES: capability["admission_lease"] = admission_lease if grant["capability_id"] == MAILBOX_AUTHORIZATION_CAPABILITY: # The tenant grants record email to the ApiEnabled audience. # Runtime must support explicit system-permission mailbox admission. capability["required_system_permission"] = "ApiEnabled" if grant["capability_id"] == APPLICATION_NAVIGATION_CAPABILITY: if (not application_owner_id or application_owner_id.strip() != application_owner_id or len(application_owner_id.encode("utf-8")) > 255 or any(unicodedata.category(c) in {"Cc", "Cf"} for c in application_owner_id)): raise ValidationFailure("navigation admission requires a valid sealed application owner") capability["application_owner_id"] = application_owner_id capabilities.append(capability) capabilities.sort(key=lambda capability: capability["runtime_capability_id"]) return capabilities def d3_partition_runtime_capability_grants( grants: list[dict[str, str]], ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: """Render only grants not already owned by AppRelease serving admission.""" catalog_consumer_grants = [ grant for grant in grants if grant["capability_id"] not in ( D3_WORKLOAD_ADMISSION_CAPABILITIES | D3_SERVING_ADMISSION_CAPABILITIES | D3_RUNTIME_FEATURE_CAPABILITIES ) ] workload_grants = [ grant for grant in grants if grant["capability_id"] in D3_WORKLOAD_ADMISSION_CAPABILITIES ] return catalog_consumer_grants, workload_grants def d3_catalog_consumer_capabilities( grants: list[dict[str, str]], catalog_snapshot_digest: str, admission_lease: str, ) -> list[dict[str, str]]: return [{ "runtime_capability_id": grant["capability_id"], "authorized_workload_id": grant["authorized_workload_id"], **({"catalog_snapshot_digest": catalog_snapshot_digest} if grant["capability_id"] in D3_SNAPSHOT_BOUND_CATALOG_CONSUMER_CAPABILITIES else {}), "admission_lease": admission_lease, } for grant in grants] INTEGRATION_BINDING_KINDS = frozenset({ "CONNECTOR_INBOUND_ROUTE", "CONNECTOR_CALLOUT_PROFILE", "CONNECTOR_FIELD_MAPPING", "CONNECTOR_PUSH_CONTRACT", }) INTEGRATION_BINDING_FIELDS = frozenset({ "authorizedWorkloadId", "kind", "ownerPack", "packInstallationId", "packPayloadDigest", "businessKey", "definitionVersion", "credentialRefs", }) RUNTIME_COMMAND_FIELDS = frozenset({ "commandKind", "payloadFormatVersion", "definitionRef", "definitionVersion", }) RUNTIME_COMMAND_BINDING_KINDS = frozenset({"CONNECTOR_INBOUND_ROUTE"}) OUTBOUND_MAIL_DECLARATION_FIELDS = frozenset({ "connectorOwnerKey", "flowActionName", "actionSlotId", "destinationSlotId", "calloutProfileOwnerPack", "calloutProfileBusinessKey", "mailCompletion", }) MAIL_COMPLETION_FIELDS = frozenset({ "recipientAddressField", "recipientOptOutField", "activity", "message", }) MAIL_ACTIVITY_FIELDS = frozenset({ "objectApiName", "sourceRecordField", "subjectField", "activityDateField", "statusField", "priorityField", "bodyField", "ownerField", "typeField", "subtypeField", "fixedSubject", "fixedStatus", "fixedPriority", "fixedType", "fixedSubtype", }) MAIL_MESSAGE_FIELDS = frozenset({ "objectApiName", "activityIdField", "fromAddressField", "toAddressField", "subjectField", "textBodyField", "incomingField", "statusField", "messageDateField", "fixedIncoming", "fixedStatus", }) MAIL_MESSAGE_V2_FIELDS = frozenset({ "ccAddressField", "bccAddressField", "htmlBodyField", }) MAIL_MESSAGE_V3_FIELDS = frozenset({"attachmentIdsField"}) PACK_PAYLOAD_DIGEST_RE = re.compile(r"nexus-pack-sha256-v1:[0-9a-f]{64}") COMMAND_KIND_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_.:/-]*") MAX_INTEGRATION_BINDINGS = 4096 MAX_IDENTIFIER_BYTES = 255 MAX_CREDENTIAL_REFS = 256 MAX_UINT32 = 2**32 - 1 MAX_UINT64 = 2**64 - 1 MAX_FEDERATED_OIDC_PROVIDERS = 32 MAX_FEDERATED_OIDC_ISSUERS = 32 MAX_ACTION_DOCUMENT_BYTES = 4 * 1024 * 1024 MAX_INVOCABLE_ACTIONS = 4096 UUID_V7_RE = 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}" ) def canonical_bytes(value: Any) -> bytes: return (json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n").encode() def require_nonempty(value: Any, label: str) -> str: if not isinstance(value, str) or not value or value.strip() != value: raise ValidationFailure(f"{label} must be a canonical non-empty string") return value def require_sha256(value: Any, label: str) -> str: if not isinstance(value, str) or not SHA256_RE.fullmatch(value): raise ValidationFailure(f"{label} must be a lowercase sha256 digest") return value def require_opaque_identifier(value: Any, label: str) -> str: value = require_nonempty(value, label) if value.strip() != value or len(value.encode("utf-8")) > MAX_IDENTIFIER_BYTES or any( unicodedata.category(character) == "Cc" for character in value ): raise ValidationFailure(f"{label} must be a valid bounded opaque identifier") return value def require_unsigned_integer(value: Any, label: str, maximum: int) -> int: if ( not isinstance(value, int) or isinstance(value, bool) or value <= 0 or value > maximum ): raise ValidationFailure(f"{label} must be a positive bounded integer") return value def validate_d3_federated_oidc_providers(values: Any) -> list[dict[str, Any]]: if ( not isinstance(values, list) or not values or len(values) > MAX_FEDERATED_OIDC_PROVIDERS ): raise ValidationFailure( "D3 federated OIDC provider set must be bounded and non-empty" ) normalized: list[dict[str, Any]] = [] for index, provider in enumerate(values): label = f"D3 federated OIDC provider {index}" if not isinstance(provider, dict) or set(provider) != { "provider_id", "audience", "session_scope", "trusted_issuers", "jwks_url", }: raise ValidationFailure(f"{label} has an unknown shape") provider_id = require_nonempty(provider.get("provider_id"), f"{label} id") audience = require_nonempty(provider.get("audience"), f"{label} audience") session_scope = require_nonempty( provider.get("session_scope"), f"{label} session scope", ) if len(provider_id) > 128 or len(audience) > 512: raise ValidationFailure(f"{label} identity is not bounded") if len(session_scope.encode("utf-8")) > MAX_IDENTIFIER_BYTES or len( session_scope.split() ) != 1: raise ValidationFailure(f"{label} session scope must be one bounded scope") issuers = provider.get("trusted_issuers") if ( not isinstance(issuers, list) or not issuers or len(issuers) > MAX_FEDERATED_OIDC_ISSUERS ): raise ValidationFailure(f"{label} issuers must be bounded and non-empty") normalized_issuers = [ require_nonempty(issuer, f"{label} issuer {issuer_index}") for issuer_index, issuer in enumerate(issuers) ] if ( any(len(issuer) > 2048 for issuer in normalized_issuers) or normalized_issuers != sorted(normalized_issuers) or len(set(normalized_issuers)) != len(normalized_issuers) ): raise ValidationFailure(f"{label} issuers must be sorted, unique, and bounded") jwks_url = require_nonempty(provider.get("jwks_url"), f"{label} JWKS URL") try: parsed = urlparse(jwks_url) hostname = parsed.hostname # Accessing port makes urllib reject malformed or out-of-range # ports instead of deferring the failure to Runtime startup. parsed.port except ValueError as exc: raise ValidationFailure(f"{label} JWKS URL must be canonical HTTPS") from exc if ( parsed.scheme != "https" or not parsed.netloc or not hostname or parsed.username is not None or parsed.password is not None or parsed.fragment or any(character.isspace() or unicodedata.category(character) == "Cc" for character in jwks_url) ): raise ValidationFailure(f"{label} JWKS URL must be canonical HTTPS") normalized.append({ "provider_id": provider_id, "audience": audience, "session_scope": session_scope, "trusted_issuers": normalized_issuers, "jwks_url": jwks_url, }) provider_ids = [provider["provider_id"] for provider in normalized] if provider_ids != sorted(provider_ids) or len(set(provider_ids)) != len(provider_ids): raise ValidationFailure("D3 federated OIDC providers must be sorted and unique") return normalized def federated_oidc_providers_from_document(document: dict[str, Any]) -> list[dict[str, Any]]: if set(document) != {"schema", "providers"}: raise ValidationFailure("D3 federated OIDC document has an unknown shape") if document.get("schema") != D3_FEDERATED_OIDC_SCHEMA: raise ValidationFailure("D3 federated OIDC document has an unknown schema") return validate_d3_federated_oidc_providers(document.get("providers")) def validate_d3_integration_bindings( bom: dict[str, Any], values: Any, *, allow_empty: bool = False, ) -> list[dict[str, Any]]: """Validate deployment-known binding facts without replacing Catalog authority. This mirrors the neutral v0.2.24 value grammar and the workload/grant closure pinned by the D3 BOM. Catalog still owns the transactional check that each reference names the exact active immutable installed-pack generation. """ validate_resolved_bom(bom) if not isinstance(values, list) or len(values) > MAX_INTEGRATION_BINDINGS: raise ValidationFailure( "D3 Catalog reconcile requires a bounded non-empty Integration binding set" ) if not values: if allow_empty: return [] raise ValidationFailure( "D3 Catalog reconcile requires a bounded non-empty Integration binding set" ) consumers = bom["catalog_consumers"] authorized_workloads = set(consumers["authorized_workload_ids"]) catalog_binding_workloads = { grant["authorized_workload_id"] for grant in consumers["catalog_view_grants"] if grant["capability_id"] == INTEGRATION_BINDING_CAPABILITY } normalized: list[dict[str, Any]] = [] identities: set[tuple[str, str, str, str, str]] = set() for index, binding in enumerate(values): label = f"D3 Integration binding {index}" if not isinstance(binding, dict): raise ValidationFailure(f"{label} must be an object") fields = set(binding) allowed = INTEGRATION_BINDING_FIELDS | {"runtimeCommand"} if not INTEGRATION_BINDING_FIELDS.issubset(fields) or not fields.issubset(allowed): raise ValidationFailure(f"{label} has an unknown or incomplete shape") workload = require_opaque_identifier( binding.get("authorizedWorkloadId"), f"{label} authorized workload", ) if workload not in authorized_workloads or workload not in catalog_binding_workloads: raise ValidationFailure( f"{label} workload lacks the D3 IntegrationBindingView authority" ) kind = binding.get("kind") if kind not in INTEGRATION_BINDING_KINDS: raise ValidationFailure(f"{label} kind is outside the neutral contract") owner_pack = require_opaque_identifier(binding.get("ownerPack"), f"{label} owner pack") pack_installation = require_opaque_identifier( binding.get("packInstallationId"), f"{label} pack installation", ) pack_digest = binding.get("packPayloadDigest") if not isinstance(pack_digest, str) or not PACK_PAYLOAD_DIGEST_RE.fullmatch(pack_digest): raise ValidationFailure(f"{label} pack payload digest is invalid") business_key = require_opaque_identifier( binding.get("businessKey"), f"{label} business key", ) definition_version = require_unsigned_integer( binding.get("definitionVersion"), f"{label} definition version", MAX_UINT64, ) credentials = binding.get("credentialRefs") if not isinstance(credentials, list) or len(credentials) > MAX_CREDENTIAL_REFS: raise ValidationFailure(f"{label} credential references are invalid") normalized_credentials = [ require_opaque_identifier(value, f"{label} credential reference {credential_index}") for credential_index, value in enumerate(credentials) ] if normalized_credentials != sorted(normalized_credentials) or len( set(normalized_credentials) ) != len(normalized_credentials): raise ValidationFailure( f"{label} credential references must be sorted and unique" ) identity = (workload, kind, owner_pack, pack_installation, business_key) if identity in identities: raise ValidationFailure("D3 Integration bindings contain a duplicate natural identity") identities.add(identity) normalized_binding: dict[str, Any] = { "authorizedWorkloadId": workload, "kind": kind, "ownerPack": owner_pack, "packInstallationId": pack_installation, "packPayloadDigest": pack_digest, "businessKey": business_key, "definitionVersion": definition_version, "credentialRefs": normalized_credentials, } has_runtime_command = "runtimeCommand" in binding if has_runtime_command != (kind in RUNTIME_COMMAND_BINDING_KINDS): raise ValidationFailure( f"{label} Runtime command presence does not match its binding kind" ) if has_runtime_command: command = binding["runtimeCommand"] if not isinstance(command, dict) or set(command) != RUNTIME_COMMAND_FIELDS: raise ValidationFailure(f"{label} Runtime command has an unknown shape") if kind != "CONNECTOR_INBOUND_ROUTE": raise ValidationFailure( f"{label} Runtime command requires an inbound-route binding" ) command_kind = command.get("commandKind") if ( not isinstance(command_kind, str) or len(command_kind.encode("utf-8")) > MAX_IDENTIFIER_BYTES or not COMMAND_KIND_RE.fullmatch(command_kind) ): raise ValidationFailure(f"{label} Runtime command kind is invalid") payload_format_version = require_unsigned_integer( command.get("payloadFormatVersion"), f"{label} Runtime command payload format version", MAX_UINT32, ) if payload_format_version != 2: raise ValidationFailure( f"{label} Runtime command payload format version must be 2 " "for CONNECTOR_INBOUND_ROUTE" ) normalized_binding["runtimeCommand"] = { "commandKind": command_kind, "payloadFormatVersion": payload_format_version, "definitionRef": require_opaque_identifier( command.get("definitionRef"), f"{label} Runtime command definition", ), "definitionVersion": require_opaque_identifier( command.get("definitionVersion"), f"{label} Runtime command definition version", ), } normalized.append(normalized_binding) return sorted(normalized, key=lambda binding: ( binding["authorizedWorkloadId"], binding["kind"], binding["ownerPack"], binding["packInstallationId"], binding["businessKey"], )) def integration_bindings_from_document( bom: dict[str, Any], document: dict[str, Any], *, allow_empty: bool = False, ) -> list[dict[str, Any]]: if set(document) != {"schema", "integrationBindings"}: raise ValidationFailure("D3 Integration binding document has an unknown shape") if document.get("schema") != D3_INTEGRATION_BINDINGS_SCHEMA: raise ValidationFailure("D3 Integration binding document has an unknown schema") return validate_d3_integration_bindings( bom, document.get("integrationBindings"), allow_empty=allow_empty, ) def _load_invocable_actions(path: Path) -> list[dict[str, Any]]: try: metadata = path.lstat() if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): raise ValidationFailure("outbound-mail actions must be a regular non-symlink file") if metadata.st_size <= 0 or metadata.st_size > MAX_ACTION_DOCUMENT_BYTES: raise ValidationFailure("outbound-mail actions file is empty or exceeds its bound") value = json.loads( path.read_bytes().decode("utf-8"), object_pairs_hook=reject_duplicate_keys, ) except ValidationFailure: raise except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise ValidationFailure(f"cannot read outbound-mail actions: {exc}") from exc if ( not isinstance(value, list) or not value or len(value) > MAX_INVOCABLE_ACTIONS or any(not isinstance(action, dict) for action in value) ): raise ValidationFailure("outbound-mail actions must be a bounded non-empty array") return value def _normalize_mail_completion(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict) or set(value) != MAIL_COMPLETION_FIELDS: raise ValidationFailure(f"{label} has an unknown shape") activity = value.get("activity") message = value.get("message") if not isinstance(activity, dict) or set(activity) != MAIL_ACTIVITY_FIELDS: raise ValidationFailure(f"{label} activity has an unknown shape") message_fields = set(message) if isinstance(message, dict) else set() if message_fields not in ( MAIL_MESSAGE_FIELDS, MAIL_MESSAGE_FIELDS | MAIL_MESSAGE_V2_FIELDS, MAIL_MESSAGE_FIELDS | MAIL_MESSAGE_V2_FIELDS | MAIL_MESSAGE_V3_FIELDS, ): raise ValidationFailure(f"{label} message has an unknown shape") fixed_incoming = message.get("fixedIncoming") if not isinstance(fixed_incoming, bool): raise ValidationFailure(f"{label} message fixedIncoming must be boolean") return { "recipientAddressField": require_opaque_identifier( value.get("recipientAddressField"), f"{label} recipientAddressField", ), "recipientOptOutField": require_opaque_identifier( value.get("recipientOptOutField"), f"{label} recipientOptOutField", ), "activity": { field: require_opaque_identifier(activity.get(field), f"{label} activity {field}") for field in sorted(MAIL_ACTIVITY_FIELDS) }, "message": { **{ field: require_opaque_identifier(message.get(field), f"{label} message {field}") for field in sorted(message_fields - {"fixedIncoming"}) }, "fixedIncoming": fixed_incoming, }, } def outbound_mail_routes_from_actions( path: Path, integration_bindings: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Expand secret-free L1 action declarations into exact Catalog routes.""" routes: list[dict[str, Any]] = [] identities: set[tuple[str, str, str, str]] = set() for index, action in enumerate(_load_invocable_actions(path)): config = action.get("config") if not isinstance(config, dict) or "outboundMail" not in config: continue label = f"outbound-mail action {index}" declaration = config.get("outboundMail") if not isinstance(declaration, dict) or set(declaration) != OUTBOUND_MAIL_DECLARATION_FIELDS: raise ValidationFailure(f"{label} declaration has an unknown shape") if action.get("isAtomic") is not True: raise ValidationFailure(f"{label} must be atomic") # L1 gates record email with a custom permission; PingCAP's L2 # override uses the same ApiEnabled audience as sibling activities. # Keep the supported policies explicit and reject empty/unknown gates. if action.get("requiredPermissions") not in (["SendRecordEmail"], ["ApiEnabled"]): raise ValidationFailure(f"{label} must require only SendRecordEmail or ApiEnabled") flow_name = require_opaque_identifier( action.get("flowDeveloperName"), f"{label} flowDeveloperName", ) targets = action.get("targetObjects") if ( not isinstance(targets, list) or not targets or len(targets) > 64 or any(not isinstance(target, str) for target in targets) ): raise ValidationFailure(f"{label} targetObjects must be a bounded non-empty array") normalized_targets = [ require_opaque_identifier(target, f"{label} targetObject") for target in targets ] if len(set(normalized_targets)) != len(normalized_targets): raise ValidationFailure(f"{label} targetObjects must be unique") scalar = { field: require_opaque_identifier(declaration.get(field), f"{label} {field}") for field in sorted(OUTBOUND_MAIL_DECLARATION_FIELDS - {"mailCompletion"}) } matches = [ binding for binding in integration_bindings if isinstance(binding, dict) and binding.get("kind") == "CONNECTOR_CALLOUT_PROFILE" and binding.get("ownerPack") == scalar["calloutProfileOwnerPack"] and binding.get("businessKey") == scalar["calloutProfileBusinessKey"] ] if len(matches) != 1: raise ValidationFailure(f"{label} requires exactly one selected callout profile") workload = require_opaque_identifier( matches[0].get("authorizedWorkloadId"), f"{label} authorized workload", ) completion = _normalize_mail_completion( declaration.get("mailCompletion"), f"{label} completion", ) for source_object in normalized_targets: identity = ( scalar["connectorOwnerKey"], flow_name, scalar["flowActionName"], source_object, ) if identity in identities: raise ValidationFailure(f"{label} duplicates a Runtime mail route") identities.add(identity) routes.append({ "authorizedWorkloadId": workload, "connectorOwnerKey": scalar["connectorOwnerKey"], "flowDeveloperName": flow_name, "flowActionName": scalar["flowActionName"], "sourceObject": source_object, "actionSlotId": scalar["actionSlotId"], "destinationSlotId": scalar["destinationSlotId"], "calloutProfileOwnerPack": scalar["calloutProfileOwnerPack"], "calloutProfileBusinessKey": scalar["calloutProfileBusinessKey"], "mailCompletion": completion, }) if not routes: raise ValidationFailure("outbound-mail actions contain no route declaration") return sorted(routes, key=lambda route: ( route["connectorOwnerKey"], route["flowDeveloperName"], route["flowActionName"], route["sourceObject"], )) def load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(), object_pairs_hook=reject_duplicate_keys) except (OSError, json.JSONDecodeError) as exc: raise ValidationFailure(f"cannot read {label}: {exc}") from exc if not isinstance(value, dict): raise ValidationFailure(f"{label} must be an object") return value def validate_resolved_bom(bom: dict[str, Any]) -> None: errors = validate_deployment_bom_view(bom) if errors: raise ValidationFailure("; ".join(errors)) def load_execution_bom(path: Path) -> dict[str, Any]: """Validate the sealed release, then derive its non-authoritative run view.""" sealed_bom = load_bom(path) validate_resolved_bom(sealed_bom) return deployment_bom_view(sealed_bom) def release_files(bom: dict[str, Any], app_packs_root: Path) -> tuple[Path, Path, Path]: installation = bom["app_installation"] paths: list[Path] = [] for field in ("artifact", "publisher_attestation", "publisher_verification_key"): locator = installation[field] path = (app_packs_root / locator["path"]).resolve() try: path.relative_to(app_packs_root.resolve()) except ValueError as exc: raise ValidationFailure(f"{field} escapes the exact app-packs checkout") from exc if not path.is_file(): raise ValidationFailure(f"{field} is missing: {path}") paths.append(path) return paths[0], paths[1], paths[2] 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 set(runtime) != {"schema", "protocols", "engineCapabilities"} or runtime.get("schema") != RUNTIME_HOST_COMPATIBILITY_SCHEMA: raise ValidationFailure("Runtime image has an unknown host compatibility manifest") if set(catalog) != {"schema", "componentKinds"} or catalog.get("schema") != CATALOG_HOST_COMPATIBILITY_SCHEMA: raise ValidationFailure("Catalog image has an unknown host compatibility manifest") protocols = runtime.get("protocols") capabilities = runtime.get("engineCapabilities") components = catalog.get("componentKinds") if not isinstance(protocols, list) or not isinstance(capabilities, list) or not isinstance(components, list): raise ValidationFailure("host compatibility manifests are incomplete") for index, protocol in enumerate(protocols): if not isinstance(protocol, dict) or set(protocol) != {"protocolId", "major", "minor"}: raise ValidationFailure(f"Runtime protocol {index} has an unknown shape") require_nonempty(protocol.get("protocolId"), f"Runtime protocol {index} id") for field, minimum in (("major", 1), ("minor", 0)): value = protocol.get(field) if not isinstance(value, int) or isinstance(value, bool) or value < minimum: raise ValidationFailure(f"Runtime protocol {index} {field} is invalid") for index, capability in enumerate(capabilities): require_nonempty(capability, f"Runtime engine capability {index}") for index, component in enumerate(components): if not isinstance(component, dict) or set(component) != {"componentKind", "componentFormatVersion"}: raise ValidationFailure(f"Catalog component kind {index} has an unknown shape") require_nonempty(component.get("componentKind"), f"Catalog component kind {index}") version = component.get("componentFormatVersion") if not isinstance(version, int) or isinstance(version, bool) or version <= 0: raise ValidationFailure(f"Catalog component kind {index} format version is invalid") normalized = { "protocols": sorted(protocols, key=lambda value: (value["protocolId"], value["major"], value["minor"])), "engineCapabilities": sorted(capabilities), "componentKinds": sorted(components, key=lambda value: (value["componentKind"], value["componentFormatVersion"])), } if (len({(row["protocolId"], row["major"], row["minor"]) for row in protocols}) != len(protocols) or len(set(capabilities)) != len(capabilities) or len({(row["componentKind"], row["componentFormatVersion"]) for row in components}) != len(components)): raise ValidationFailure("host compatibility manifests contain duplicate facts") return normalized def validate_app_engine_compatibility( package: dict[str, Any], compatibility: dict[str, Any], ) -> None: """Prove that one independently released L1 App can run on one L0 Engine.""" manifest = package.get("canonical_manifest") if not isinstance(manifest, dict): raise ValidationFailure("immutable AppPackage has no canonical_manifest") required_protocols = manifest.get("required_protocols") required_capabilities = manifest.get("required_engine_capabilities") if not isinstance(required_protocols, list) or not required_protocols: raise ValidationFailure("immutable AppPackage has no required_protocols") if ( not isinstance(required_capabilities, list) or required_capabilities != sorted(set(required_capabilities)) or any( not isinstance(value, str) or not value or value.strip() != value for value in required_capabilities ) ): raise ValidationFailure( "immutable AppPackage has malformed required_engine_capabilities" ) provided_protocols = compatibility.get("protocols") provided_capabilities = compatibility.get("engineCapabilities") if not isinstance(provided_protocols, list) or not isinstance(provided_capabilities, list): raise ValidationFailure("Runtime host compatibility is incomplete") provided_by_id = { (protocol.get("protocolId"), protocol.get("major")): protocol.get("minor") for protocol in provided_protocols if isinstance(protocol, dict) } for index, protocol in enumerate(required_protocols): if not isinstance(protocol, dict) or set(protocol) != { "protocol_id", "major", "minimum_minor", }: raise ValidationFailure( f"AppPackage required protocol {index} has an unknown shape" ) protocol_id = protocol.get("protocol_id") major = protocol.get("major") minimum_minor = protocol.get("minimum_minor") if ( not isinstance(protocol_id, str) or not protocol_id or protocol_id.strip() != protocol_id or not isinstance(major, int) or isinstance(major, bool) or major <= 0 or not isinstance(minimum_minor, int) or isinstance(minimum_minor, bool) or minimum_minor < 0 ): raise ValidationFailure( f"AppPackage required protocol {index} is invalid" ) provided_minor = provided_by_id.get((protocol_id, major)) if not isinstance(provided_minor, int) or provided_minor < minimum_minor: raise ValidationFailure( "L0 Engine does not satisfy L1 protocol " f"{protocol_id} major {major} minor >= {minimum_minor}" ) missing_capabilities = sorted( set(required_capabilities) - set(provided_capabilities) ) if missing_capabilities: raise ValidationFailure( "L0 Engine does not provide L1 required capabilities: " + ", ".join(missing_capabilities) ) def validate_apprelease_compatibility( bom: dict[str, Any], app_packs_root: Path, runtime_host_compatibility_path: Path, catalog_host_compatibility_path: Path, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: """Validate the immutable AppRelease against the already deployed L0.""" validate_resolved_bom(bom) artifact_errors = verify_app_packs_artifacts(bom, app_packs_root) if artifact_errors: raise ValidationFailure("; ".join(artifact_errors)) package_path, _, _ = release_files(bom, app_packs_root) package = load_json(package_path, "AppPackage") manifest = package.get("canonical_manifest") dependencies = manifest.get("dependencies") if isinstance(manifest, dict) else None if dependencies != []: raise ValidationFailure("CRM AppRelease must have no unresolved App dependencies") required_engine_capabilities = ( manifest.get("required_engine_capabilities") if isinstance(manifest, dict) else None ) if not isinstance(required_engine_capabilities, list): raise ValidationFailure("D3 AppRelease has no Runtime capability authority") # Provider-effect consumption is activated by the immutable connector # contract in metadata, not by an sfcrm service-handler declaration. metadata_activated_runtime_capabilities = { "runtime.outbound-intents.consume.v1", } missing_runtime_grants = sorted(({ grant["capability_id"] for grant in bom["catalog_consumers"]["runtime_capability_grants"] } - metadata_activated_runtime_capabilities) - set(required_engine_capabilities)) if missing_runtime_grants: raise ValidationFailure( "D3 Runtime capability grants are absent from the AppRelease authority: " + ", ".join(missing_runtime_grants) ) compatibility = host_compatibility( runtime_host_compatibility_path, catalog_host_compatibility_path, ) validate_app_engine_compatibility(package, compatibility) frontend_receipt = frontend_compatibility_receipt(bom, package) return package, frontend_receipt, compatibility def selected_capabilities( package: dict[str, Any], installation: dict[str, Any], *, composite_dependencies: bool = False, workload_only_app_capability_bindings: dict[str, str] | None = None, ) -> list[dict[str, str]]: capabilities = installation.get("capabilities") if composite_dependencies: package_capabilities = app_package_composite_capability_selections( package, capabilities, workload_only_app_capability_bindings, ) return [{ "appCapabilityId": capability["capability_id"], "handlerRef": capability["handler_ref"], "runtimeCapabilityId": capability["runtime_capability_id"], } for capability in package_capabilities] package_capabilities = app_package_capabilities(package) package_bindings = [{ "capability_id": capability["capability_id"], "runtime_capability_id": capability["runtime_capability_id"], } for capability in package_capabilities] if capabilities != package_bindings: raise ValidationFailure( "AppInstallation capability projection does not exactly match the immutable AppPackage" ) return [{ "appCapabilityId": capability["capability_id"], "handlerRef": capability["handler_ref"], "runtimeCapabilityId": capability["runtime_capability_id"], } for capability in package_capabilities] def frontend_receipt_identity(bom: dict[str, Any]) -> dict[str, Any]: publication = bom["frontend"]["publication"] identity = { "schema": FRONTEND_RECEIPT_SCHEMA, "appReleaseDigest": bom["app_installation"]["app_release_digest"], "frontendReleaseDigest": publication["digest"], "frontendSource": { "repository": publication["source_repository"], "commit": publication["source_commit"], "sourceTupleSHA256": publication["source_tuple_sha256"], }, } if bom.get("schema_version") == LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: identity.update({ "schema": LOCAL_TEST3_FRONTEND_RECEIPT_SCHEMA, "releaseClass": LOCAL_TEST3_RELEASE_CLASS, "imageIdentityKind": "docker-image-id", }) else: identity.update({ "frontendProvenanceDigest": publication["provenance_digest"], "frontendSBOMDigest": publication["sbom_digest"], }) return identity def frontend_compatibility_receipt(bom: dict[str, Any], package: dict[str, Any]) -> dict[str, Any]: manifest = package.get("canonical_manifest") compatibility = manifest.get("frontend_compatibility") if isinstance(manifest, dict) else None if not isinstance(compatibility, dict) or compatibility.get("mode") != "required": raise ValidationFailure("CRM AppRelease must require one frontend compatibility binding") host_sdk = compatibility.get("host_sdk") contribution = compatibility.get("contribution_schema") for label, protocol in (("host SDK", host_sdk), ("contribution schema", contribution)): if not isinstance(protocol, dict): raise ValidationFailure(f"AppPackage frontend {label} compatibility is missing") if set(protocol) != {"protocol_id", "major", "minimum_minor"}: raise ValidationFailure(f"AppPackage frontend {label} compatibility has an unknown shape") require_nonempty(protocol.get("protocol_id"), f"frontend {label} protocol") if (not isinstance(protocol.get("major"), int) or isinstance(protocol.get("major"), bool) or protocol["major"] <= 0 or not isinstance(protocol.get("minimum_minor"), int) or isinstance(protocol.get("minimum_minor"), bool) or protocol["minimum_minor"] < 0): raise ValidationFailure(f"AppPackage frontend {label} version is invalid") installation = bom["app_installation"] if package.get("app_release_digest") != installation["app_release_digest"]: raise ValidationFailure("frontend receipt AppPackage disagrees with the resolved AppRelease") return { **frontend_receipt_identity(bom), "requiredCompatibility": { "hostSdk": host_sdk, "contributionSchema": contribution, }, } def sha256_digest(value: dict[str, Any]) -> str: return "sha256:" + hashlib.sha256(canonical_bytes(value)).hexdigest() def validate_frontend_receipt(bom: dict[str, Any], receipt: dict[str, Any]) -> None: expected = frontend_receipt_identity(bom) for field, value in expected.items(): if receipt.get(field) != value: raise ValidationFailure(f"frontend compatibility receipt {field} disagrees with the resolved BOM") if set(receipt) != set(expected) | {"requiredCompatibility"}: raise ValidationFailure("frontend compatibility receipt has an unknown shape") compatibility = receipt.get("requiredCompatibility") if not isinstance(compatibility, dict) or set(compatibility) != {"hostSdk", "contributionSchema"}: raise ValidationFailure("frontend compatibility receipt requirements are incomplete") def render_catalog_config( bom: dict[str, Any], app_packs_root: Path, runtime_host_compatibility_path: Path, catalog_host_compatibility_path: Path, *, org_id: str, workload_id: str, runtime_audience: str, package_mount_path: str, attestation_mount_path: str, artifact_store_root: str, idempotency_key: str, operation_kind: str, app_installation_id: str, expected_catalog_revision: int, expected_security_epoch: int, publisher_valid_from: str, publisher_valid_until: str, schema_generation: int, engine_generation: int, integration_bindings: list[dict[str, Any]] | None = None, outbound_mail_actions_path: Path | None = None, allow_empty_integration: bool = False, phase_one_publication_command_id: str | None = None, approval_publication_bundle_path: str | None = None, approval_publication_bundle_digest: str | None = None, refresh_active_release: bool = False, ) -> dict[str, Any]: package, frontend_receipt, compatibility = validate_apprelease_compatibility( bom, app_packs_root, runtime_host_compatibility_path, catalog_host_compatibility_path, ) if schema_generation <= 0 or engine_generation <= 0: raise ValidationFailure("schema_generation and engine_generation must be positive") if (not isinstance(expected_catalog_revision, int) or isinstance(expected_catalog_revision, bool) or expected_catalog_revision < 0 or not isinstance(expected_security_epoch, int) or isinstance(expected_security_epoch, bool) or expected_security_epoch < 0): raise ValidationFailure("expected Catalog revision and security epoch must be non-negative integers") _, attestation_path, key_path = release_files(bom, app_packs_root) attestation = load_json(attestation_path, "publisher attestation") key = load_json(key_path, "publisher verification key") installation = bom["app_installation"] if key.get("publisher_id") != attestation.get("publisher_id") or key.get("key_id") != attestation.get("key_id"): raise ValidationFailure("publisher verification key and attestation disagree") if operation_kind not in ("PREPARE", "UPGRADE", "ROLLBACK"): raise ValidationFailure("operation_kind must be PREPARE, UPGRADE, or ROLLBACK") if refresh_active_release and operation_kind != "UPGRADE": raise ValidationFailure("refresh_active_release requires operation_kind UPGRADE") if operation_kind == "PREPARE" and app_installation_id != "": raise ValidationFailure("PREPARE must not provide an AppInstallation identifier") if operation_kind != "PREPARE": require_nonempty(app_installation_id, "app_installation_id") config = { "schema": bom["execution"]["catalog_reconcile_config_schema"], "orgId": require_nonempty(org_id, "org_id"), "appId": installation["app_id"], "expectedAppReleaseDigest": installation["app_release_digest"], "metadataSourceDigest": require_nonempty(bom["l2_metadata"]["tree_sha256"], "L2 metadata tree digest"), "releaseImport": { "packagePath": require_nonempty(package_mount_path, "package_mount_path"), "attestationPath": require_nonempty(attestation_mount_path, "attestation_mount_path"), "publisher": { "publisherId": key["publisher_id"], "keys": [{ "keyId": key["key_id"], "algorithm": key["algorithm"], "publicKeyBase64": key["public_key"], "validFrom": require_nonempty(publisher_valid_from, "publisher_valid_from"), "validUntil": require_nonempty(publisher_valid_until, "publisher_valid_until"), }], }, }, "artifactStoreRoot": require_nonempty(artifact_store_root, "artifact_store_root"), "hostCompatibility": compatibility, "idempotencyKey": require_nonempty(idempotency_key, "idempotency_key"), "operationKind": operation_kind, "appInstallationId": app_installation_id, "expectedCatalogRevision": expected_catalog_revision, "expectedSecurityEpoch": expected_security_epoch, "runtimeAudience": require_nonempty(runtime_audience, "runtime_audience"), "workloadId": require_nonempty(workload_id, "workload_id"), # Catalog binds the immutable AppRelease/capability publication, not # the deployer's OCI implementation choice. The resolved BOM verifies # the sfcrm image independently. "serviceReleaseDigest": installation["app_release_digest"], "frontend": { "releaseDigest": bom["frontend"]["publication"]["digest"], "compatibilityReceiptDigest": sha256_digest(frontend_receipt), }, "schemaGeneration": schema_generation, "engineGeneration": engine_generation, "resolvedDependencies": [], } if refresh_active_release: config["refreshActiveRelease"] = True 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"] } config["capabilities"] = selected_capabilities( package, installation, composite_dependencies=True, workload_only_app_capability_bindings={ 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 }, ) selected_handlers = {item["appCapabilityId"] for item in config["capabilities"]} workload_dependencies = granted_workload_capabilities & D3_WORKLOAD_ADMISSION_CAPABILITIES if any( capability["capability_id"] in selected_handlers and workload_dependencies.intersection(capability["required_runtime_capabilities"]) for capability in package["canonical_manifest"]["provided_capabilities"] ): # A composite selected handler may also call a workload-admitted RPC. # Carry the existing BOM grants explicitly; do not invent a Catalog # projection for those dependencies. Older non-composite inputs retain # their existing v4 representation. config["handlerWorkloadCapabilities"] = sorted(workload_dependencies) if ( not isinstance(phase_one_publication_command_id, str) or not UUID_V7_RE.fullmatch(phase_one_publication_command_id) ): raise ValidationFailure( "D3 phase-one Catalog publication requires a lowercase UUIDv7 command" ) integration_bindings = validate_d3_integration_bindings( bom, integration_bindings, allow_empty=allow_empty_integration, ) outbound_mail_routes = ( outbound_mail_routes_from_actions( outbound_mail_actions_path, integration_bindings, ) if outbound_mail_actions_path is not None else [] ) consumers = bom["catalog_consumers"] config.update({ "authorizedWorkloadIds": list(consumers["authorized_workload_ids"]), "consumerGrants": [{ "authorizedWorkloadId": grant["authorized_workload_id"], "capabilityId": grant["capability_id"], } for grant in consumers["catalog_view_grants"]], "integrationBindings": integration_bindings, **({"outboundMailRoutes": outbound_mail_routes} if outbound_mail_routes else {}), "phaseOnePublication": { "commandId": phase_one_publication_command_id, }, }) if approval_publication_bundle_path is not None: config["approvalPublication"] = { "bundlePath": require_nonempty( approval_publication_bundle_path, "Approval publication bundle path", ), "expectedDigest": require_sha256( approval_publication_bundle_digest, "Approval publication bundle digest", ), } return config def validate_catalog_output_for_deployment( bom: dict[str, Any], catalog_output: dict[str, Any], *, org_id: str, workload_id: str, ) -> None: """Reject a Catalog handoff that is not for this exact deployment input. The Catalog output is the sole authority for the newly allocated AppInstallationID and the selected immutable snapshot. L2 therefore does not manufacture a second selector for either fact. It must, however, bind that output to the reviewed release, exact metadata tree, and operator-declared Org/workload before either serving or worker config can copy the Catalog-owned scope. """ validate_resolved_bom(bom) expected_org_id = require_nonempty(org_id, "org_id") expected_workload_id = require_nonempty(workload_id, "workload_id") execution = bom["execution"] if catalog_output.get("schema") != execution["catalog_reconcile_output_schema"]: raise ValidationFailure("Catalog output has the wrong schema") installation = bom["app_installation"] expected = { "orgId": expected_org_id, "appReleaseDigest": installation["app_release_digest"], "metadataSourceDigest": bom["l2_metadata"]["tree_sha256"], "workloadId": expected_workload_id, } for field, value in expected.items(): if catalog_output.get(field) != value: raise ValidationFailure( f"Catalog output {field} disagrees with the resolved BOM or declared deployment input" ) # ServiceReleaseDigest is the Catalog-owned handler-binding identity. The # resolved BOM independently pins and verifies the actual sfcrm OCI image. # Equating the two here would not attest the calling process (Runtime never # receives an OCI attestation) and would incorrectly require Catalog # publication for every compatible AppService implementation refresh. require_sha256( catalog_output.get("serviceReleaseDigest"), "Catalog output serviceReleaseDigest", ) require_nonempty(catalog_output.get("operationId"), "Catalog output operationId") require_nonempty(catalog_output.get("appInstallationId"), "Catalog output appInstallationId") for field in ("catalogRevision", "schemaGeneration", "engineGeneration", "securityEpoch"): value = catalog_output.get(field) if not isinstance(value, int) or isinstance(value, bool) or value <= 0: raise ValidationFailure(f"Catalog output {field} must be positive") require_sha256(catalog_output.get("catalogSnapshotDigest"), "Catalog output catalogSnapshotDigest") output_capabilities = catalog_output.get("capabilities") expected_capabilities = installation["capabilities"] if not isinstance(output_capabilities, list) or len(output_capabilities) != len(expected_capabilities): raise ValidationFailure("Catalog output must contain the complete version-atomic capability set") for index, (binding, output) in enumerate(zip(expected_capabilities, output_capabilities)): if not isinstance(output, dict): raise ValidationFailure(f"Catalog output capability {index} must be an object") if output.get("appCapabilityId") != binding["capability_id"] or output.get("runtimeCapabilityId") != binding["runtime_capability_id"]: raise ValidationFailure(f"Catalog output capability {index} disagrees with the resolved BOM") require_nonempty(output.get("handlerBindingId"), f"Catalog output capability {index} handlerBindingId") ownership_epoch = output.get("ownershipEpoch") if not isinstance(ownership_epoch, int) or isinstance(ownership_epoch, bool) or ownership_epoch <= 0: raise ValidationFailure(f"Catalog output capability {index} ownershipEpoch must be positive") require_sha256(output.get("runtimeMetadataViewDigest"), f"Catalog output capability {index} runtimeMetadataViewDigest") require_sha256(output.get("compatibilityReceiptDigest"), f"Catalog output capability {index} compatibilityReceiptDigest") def render_serving_inputs( bom: dict[str, Any], catalog_output: dict[str, Any], frontend_receipt: dict[str, Any], *, replica_id: str, org_id: str, workload_id: str, catalog_store_root: str, federated_oidc_providers: list[dict[str, Any]] | None = None, application_owner_id: str = "", ) -> dict[str, Any]: validate_frontend_receipt(bom, frontend_receipt) validate_catalog_output_for_deployment( bom, catalog_output, org_id=org_id, workload_id=workload_id, ) frontend_expected = { "frontendReleaseDigest": bom["frontend"]["publication"]["digest"], "frontendCompatibilityReceiptDigest": sha256_digest(frontend_receipt), } for field, value in frontend_expected.items(): if catalog_output.get(field) != value: raise ValidationFailure( f"Catalog output {field} disagrees with the frontend compatibility receipt" ) consumers = bom["catalog_consumers"] providers = ( validate_d3_federated_oidc_providers(federated_oidc_providers) if federated_oidc_providers is not None else [] ) runtime = { "schema": bom["execution"]["runtime_serving_config_schema"], "replica_id": require_nonempty(replica_id, "replica_id"), "org_id": catalog_output["orgId"], "app_installation_id": catalog_output["appInstallationId"], "catalog_snapshot_store_root": require_nonempty( catalog_store_root, "catalog_store_root", ), "expected_service_release_digest": catalog_output["serviceReleaseDigest"], "schema_generation": catalog_output["schemaGeneration"], "engine_generation": catalog_output["engineGeneration"], "identity_attempt_cap": D3_IDENTITY_ATTEMPT_CAP, "permission_state_protocol": "permission_state.v1", "authorized_workload_ids": sorted(consumers["authorized_workload_ids"] + (["mcp-worker"] if "mcp_worker" in bom["services"] else [])), **({"user_api_callers": ["mcp-worker"]} if "mcp_worker" in bom["services"] else {}), } if providers: runtime["federated_oidc_providers"] = providers output_capabilities = catalog_output.get("capabilities") assert isinstance(output_capabilities, list) runtime_capabilities: list[dict[str, Any]] = [] for index, output in enumerate(output_capabilities): assert isinstance(output, dict) runtime_capability = { "app_capability_id": output["appCapabilityId"], "runtime_capability_id": output["runtimeCapabilityId"], "expected_ownership_epoch": output.get("ownershipEpoch"), "expected_compatibility_receipt_digest": require_sha256( output.get("compatibilityReceiptDigest"), f"Catalog output capability {index} compatibilityReceiptDigest", ), "admission_lease": D3_CAPABILITY_ADMISSION_LEASE, "authorized_workload_id": consumers["handler_workload_id"], } if output["runtimeCapabilityId"] == MUTATION_RUNTIME_CAPABILITY: runtime_capability["outcome_retention"] = MUTATION_OUTCOME_RETENTION elif output["runtimeCapabilityId"] == PERMISSION_AUTHORING_RUNTIME_CAPABILITY: runtime_capability["outcome_retention"] = PERMISSION_AUTHORING_OUTCOME_RETENTION runtime_capabilities.append(runtime_capability) session_scope_policy = d3_session_scope_policy( [capability["runtime_capability_id"] for capability in runtime_capabilities], consumers["runtime_capability_grants"], ) runtime.update({ "session_scope_capabilities": session_scope_policy, "capabilities": runtime_capabilities, }) catalog_consumer_grants, workload_grants = d3_partition_runtime_capability_grants( consumers["runtime_capability_grants"], ) runtime["catalog_consumer_capabilities"] = d3_catalog_consumer_capabilities( catalog_consumer_grants, catalog_output["catalogSnapshotDigest"], D3_CAPABILITY_ADMISSION_LEASE, ) runtime["workload_capabilities"] = d3_workload_capabilities( workload_grants, D3_CAPABILITY_ADMISSION_LEASE, application_owner_id=application_owner_id, ) return runtime def worker_file_stem(work_family: str) -> str: return work_family.removeprefix("runtime.").removesuffix(".v1").replace(".", "-") def runtime_worker_release_identity( bom: dict[str, Any], selected: dict[str, str] | None = None, ) -> dict[str, str]: """Select the Worker binary identity without changing BOM-owned policy.""" worker = bom["services"]["runtime_worker"] if selected is None else selected if selected is not None and set(selected) != {"digest", "source_commit"}: raise ValidationFailure("selected Runtime Worker release has an unknown shape") return { "digest": require_sha256(worker.get("digest"), "Runtime Worker release digest"), "source_commit": require_nonempty( worker.get("source_commit"), "Runtime Worker source commit", ), } def render_worker_activation_inputs( bom: dict[str, Any], *, deployment_id_prefix: str, expected_ownership_epoch: int, worker_release: dict[str, str] | None = None, ) -> dict[str, dict[str, Any]]: """Render Catalog-owned ownership activation requests, one per worker role.""" validate_resolved_bom(bom) prefix = require_nonempty(deployment_id_prefix, "deployment_id_prefix") if not isinstance(expected_ownership_epoch, int) or isinstance(expected_ownership_epoch, bool) or expected_ownership_epoch < 0: raise ValidationFailure("expected_ownership_epoch must be a non-negative integer") worker_digest = runtime_worker_release_identity(bom, worker_release)["digest"] result: dict[str, dict[str, Any]] = {} for role in bom["runtime_worker_roles"]: stem = worker_file_stem(role["work_family"]) result[stem] = { "schema": bom["execution"]["runtime_work_activation_schema"], "operationId": f"{prefix}-activate-{stem}-e{expected_ownership_epoch + 1}", "workFamily": role["work_family"], "deploymentId": f"{prefix}-{stem}", "workerReleaseDigest": worker_digest, "expectedOwnershipEpoch": str(expected_ownership_epoch), } return result def _canonical_epoch(value: object, label: str) -> int: if not isinstance(value, str) or not value.isascii() or not value.isdecimal(): raise ValidationFailure(f"{label} is not canonical") if len(value) > 1 and value.startswith("0"): raise ValidationFailure(f"{label} is not canonical") return int(value) def render_worker_role_inputs( bom: dict[str, Any], catalog_output: dict[str, Any], activation_requests: dict[str, dict[str, Any]], activation_outputs: dict[str, dict[str, Any]], *, org_id: str, workload_id: str, catalog_store_root: str, worker_release: dict[str, str] | None = None, ) -> dict[str, dict[str, Any]]: """Render worker configs only from outputs bound to their exact CAS inputs.""" validate_catalog_output_for_deployment( bom, catalog_output, org_id=org_id, workload_id=workload_id, ) expected_mutation_bindings = [ binding for binding in bom["app_installation"]["capabilities"] if binding["runtime_capability_id"] == MUTATION_RUNTIME_CAPABILITY ] if len(expected_mutation_bindings) != 1: raise ValidationFailure("Resolved BOM must contain exactly one record-mutation capability binding") expected_mutation = expected_mutation_bindings[0] mutation_bindings = [ binding for binding in catalog_output.get("capabilities", []) if isinstance(binding, dict) and binding.get("appCapabilityId") == expected_mutation["capability_id"] and binding.get("runtimeCapabilityId") == expected_mutation["runtime_capability_id"] ] if len(mutation_bindings) != 1: raise ValidationFailure("Catalog output must contain exactly one record-mutation capability binding") flow_bindings = [ binding for binding in catalog_output.get("capabilities", []) if isinstance(binding, dict) and binding.get("appCapabilityId") == "io.nexuscrm.sfcrm.flow-actions.v1" and binding.get("runtimeCapabilityId") == "runtime.automation.execute-flow.v1" ] if len(flow_bindings) != 1: raise ValidationFailure("Catalog output must contain exactly one scheduled-Flow capability binding") store_root = require_nonempty(catalog_store_root, "catalog_store_root") worker = runtime_worker_release_identity(bom, worker_release) worker_digest = worker["digest"] worker_source = worker["source_commit"] expected_stems = {worker_file_stem(role["work_family"]) for role in bom["runtime_worker_roles"]} if set(activation_requests) != expected_stems: raise ValidationFailure("Runtime Worker activation requests are incomplete or contain another work owner") if set(activation_outputs) != expected_stems: raise ValidationFailure("Runtime Worker activation outputs are incomplete or contain another work owner") result: dict[str, dict[str, Any]] = {} for policy in bom["runtime_worker_roles"]: stem = worker_file_stem(policy["work_family"]) request = activation_requests[stem] activation = activation_outputs[stem] expected_request_fields = { "schema", "operationId", "workFamily", "deploymentId", "workerReleaseDigest", "expectedOwnershipEpoch", } if set(request) != expected_request_fields or request.get("schema") != bom["execution"]["runtime_work_activation_schema"]: raise ValidationFailure(f"Runtime Worker activation request {stem} has an unknown shape") if request.get("workFamily") != policy["work_family"] or request.get("workerReleaseDigest") != worker_digest: raise ValidationFailure(f"Runtime Worker activation request {stem} disagrees with the resolved BOM") operation_id = require_nonempty(request.get("operationId"), f"Runtime Worker {stem} operationId") requested_deployment_id = require_nonempty( request.get("deploymentId"), f"Runtime Worker {stem} requested deploymentId", ) expected_epoch = _canonical_epoch( request.get("expectedOwnershipEpoch"), f"Runtime Worker activation request {stem} expectedOwnershipEpoch", ) expected_activation_fields = { "schema", "operationId", "workFamily", "deploymentId", "workerReleaseDigest", "ownershipEpoch", } if set(activation) != expected_activation_fields or activation.get("schema") != "nexuscrm.runtime-work-ownership-activation-result/v1": raise ValidationFailure(f"Runtime Worker activation output {stem} has an unknown shape") if ( activation.get("operationId") != operation_id or activation.get("workFamily") != request["workFamily"] or activation.get("deploymentId") != requested_deployment_id or activation.get("workerReleaseDigest") != request["workerReleaseDigest"] ): raise ValidationFailure(f"Runtime Worker activation output {stem} does not bind its exact activation request") deployment_id = require_nonempty(activation.get("deploymentId"), f"Runtime Worker {stem} deploymentId") epoch = _canonical_epoch(activation.get("ownershipEpoch"), f"Runtime Worker activation output {stem} ownershipEpoch") if epoch != expected_epoch + 1: raise ValidationFailure(f"Runtime Worker activation output {stem} does not advance the requested ownership epoch") config = { "schema": policy["config_schema"], "work_family": policy["work_family"], "deployment_id": deployment_id, "worker_release_digest": worker_digest, "worker_source_revision": worker_source, "ownership_epoch": epoch, "batch_size": policy["batch_size"], "poll_interval": policy["poll_interval"], "lease_duration": policy["lease_duration"], "retry_jitter_max": policy["retry_jitter_max"], "max_attempts": policy["max_attempts"], "db_max_open_conns": policy["db_max_open_conns"], "db_max_idle_conns": policy["db_max_idle_conns"], "shutdown_timeout": policy["shutdown_timeout"], } if "parallelism" in policy: config["parallelism"] = policy["parallelism"] config["permission_state_protocol"] = "permission_state.v1" if "processed_retention" in policy: config["processed_retention"] = policy["processed_retention"] config["processed_cleanup_interval"] = policy["processed_cleanup_interval"] if policy["work_family"] in { "runtime.record-event-dispatch.v1", "runtime.rollup-delta.v1", "runtime.scheduled-flow.v1", "runtime.platform-command.v1", "runtime.bulk-ingest.v1", }: config.update({ "org_id": catalog_output["orgId"], "app_installation_id": catalog_output["appInstallationId"], "catalog_snapshot_store_root": store_root, "schema_generation": catalog_output["schemaGeneration"], "engine_generation": catalog_output["engineGeneration"], }) if policy["work_family"] == "runtime.scheduled-flow.v1": config.update({ "app_capability_id": flow_bindings[0]["appCapabilityId"], "runtime_capability_id": flow_bindings[0]["runtimeCapabilityId"], "expected_compatibility_receipt_digest": require_sha256( flow_bindings[0].get("compatibilityReceiptDigest"), "scheduled Flow compatibility receipt", ), }) elif policy["work_family"] == "runtime.bulk-ingest.v1": config.update({ # The public Bulk lifecycle authorizes job creation and # upload. The independent worker executes record effects, # so it must freeze the exact record-mutation Catalog # binding rather than the lifecycle transport binding. "app_capability_id": mutation_bindings[0]["appCapabilityId"], "runtime_capability_id": mutation_bindings[0]["runtimeCapabilityId"], "expected_compatibility_receipt_digest": require_sha256( mutation_bindings[0].get("compatibilityReceiptDigest"), "Bulk Ingest record-mutation compatibility receipt", ), }) elif policy["work_family"] == "runtime.platform-command.v1": config["runtime_capability_id"] = "runtime.commands.accept.v1" else: config["runtime_capability_id"] = MUTATION_RUNTIME_CAPABILITY result[stem] = config return result def write_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) if path.is_symlink(): raise ValidationFailure(f"refusing to replace non-regular JSON output: {path}") if path.exists(): current = path.lstat() if not path.is_file(): raise ValidationFailure(f"refusing to replace non-regular JSON output: {path}") if stat.S_IMODE(current.st_mode) & stat.S_IWUSR == 0: raise ValidationFailure(f"refusing to overwrite protected JSON output: {path}") path.write_bytes(canonical_bytes(value)) def main() -> int: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) catalog = subparsers.add_parser("catalog") catalog.add_argument("--bom", required=True) catalog.add_argument("--app-packs-root", required=True) catalog.add_argument("--runtime-host-compatibility", required=True) catalog.add_argument("--catalog-host-compatibility", required=True) catalog.add_argument("--output", required=True) catalog.add_argument("--org-id", required=True) catalog.add_argument("--workload-id", required=True) catalog.add_argument("--runtime-audience", required=True) catalog.add_argument("--package-mount-path", required=True) catalog.add_argument("--attestation-mount-path", required=True) catalog.add_argument("--artifact-store-root", required=True) catalog.add_argument("--idempotency-key", required=True) catalog.add_argument("--operation-kind", choices=("PREPARE", "UPGRADE", "ROLLBACK"), required=True) catalog.add_argument("--app-installation-id", default="") catalog.add_argument("--frontend-receipt-output", required=True) catalog.add_argument("--expected-catalog-revision", type=int, required=True) catalog.add_argument("--expected-security-epoch", type=int, required=True) catalog.add_argument("--publisher-valid-from", required=True) catalog.add_argument("--publisher-valid-until", required=True) catalog.add_argument("--schema-generation", type=int, default=1) catalog.add_argument("--engine-generation", type=int, default=1) catalog.add_argument("--integration-bindings") catalog.add_argument("--outbound-mail-actions") catalog.add_argument("--allow-empty-integration", action="store_true") catalog.add_argument("--phase-one-publication-command-id") catalog.add_argument("--approval-publication-bundle-path") catalog.add_argument("--approval-publication-bundle-digest") catalog.add_argument("--refresh-active-release", action="store_true") serving = subparsers.add_parser("serving") serving.add_argument("--bom", required=True) serving.add_argument("--catalog-output", required=True) serving.add_argument("--frontend-receipt", required=True) serving.add_argument("--runtime-output", required=True) serving.add_argument("--replica-id", required=True) serving.add_argument("--org-id", required=True) serving.add_argument("--workload-id", required=True) serving.add_argument("--catalog-store-root", required=True) serving.add_argument("--federated-oidc-providers") serving.add_argument("--application-owner-id", default="") worker_activation = subparsers.add_parser("worker-activation") worker_activation.add_argument("--bom", required=True) worker_activation.add_argument("--output-dir", required=True) worker_activation.add_argument("--deployment-id-prefix", required=True) worker_activation.add_argument("--expected-ownership-epoch", type=int, default=0) worker_role = subparsers.add_parser("worker-role") worker_role.add_argument("--bom", required=True) worker_role.add_argument("--catalog-output", required=True) worker_role.add_argument("--activation-config-dir", required=True) worker_role.add_argument( "--activation-config-override", help="one newly accepted activation request that replaces its work-family entry", ) worker_role.add_argument("--activation-output-dir", required=True) worker_role.add_argument("--output-dir", required=True) worker_role.add_argument("--org-id", required=True) worker_role.add_argument("--workload-id", required=True) worker_role.add_argument("--catalog-store-root", required=True) args = parser.parse_args() try: bom = load_execution_bom(Path(args.bom).resolve()) if args.command == "catalog": integration_bindings = None if args.integration_bindings: binding_document = load_json( Path(args.integration_bindings).resolve(), "D3 Integration bindings", ) integration_bindings = integration_bindings_from_document( bom, binding_document, allow_empty=args.allow_empty_integration, ) rendered = render_catalog_config( bom, Path(args.app_packs_root).resolve(), Path(args.runtime_host_compatibility).resolve(), Path(args.catalog_host_compatibility).resolve(), org_id=args.org_id, workload_id=args.workload_id, runtime_audience=args.runtime_audience, package_mount_path=args.package_mount_path, attestation_mount_path=args.attestation_mount_path, artifact_store_root=args.artifact_store_root, idempotency_key=args.idempotency_key, operation_kind=args.operation_kind, app_installation_id=args.app_installation_id, expected_catalog_revision=args.expected_catalog_revision, expected_security_epoch=args.expected_security_epoch, publisher_valid_from=args.publisher_valid_from, publisher_valid_until=args.publisher_valid_until, schema_generation=args.schema_generation, engine_generation=args.engine_generation, integration_bindings=integration_bindings, outbound_mail_actions_path=( Path(args.outbound_mail_actions) if args.outbound_mail_actions else None ), allow_empty_integration=args.allow_empty_integration, phase_one_publication_command_id=args.phase_one_publication_command_id, approval_publication_bundle_path=args.approval_publication_bundle_path, approval_publication_bundle_digest=args.approval_publication_bundle_digest, refresh_active_release=args.refresh_active_release, ) write_json(Path(args.output).resolve(), rendered) package_path, _, _ = release_files(bom, Path(args.app_packs_root).resolve()) package = load_json(package_path, "AppPackage") write_json( Path(args.frontend_receipt_output).resolve(), frontend_compatibility_receipt(bom, package), ) elif args.command == "serving": output = load_json(Path(args.catalog_output).resolve(), "Catalog output") receipt = load_json(Path(args.frontend_receipt).resolve(), "frontend compatibility receipt") providers = None if args.federated_oidc_providers: provider_document = load_json( Path(args.federated_oidc_providers).resolve(), "D3 federated OIDC providers", ) providers = federated_oidc_providers_from_document(provider_document) runtime = render_serving_inputs( bom, output, receipt, replica_id=args.replica_id, org_id=args.org_id, workload_id=args.workload_id, catalog_store_root=args.catalog_store_root, federated_oidc_providers=providers, application_owner_id=args.application_owner_id, ) write_json(Path(args.runtime_output).resolve(), runtime) elif args.command == "worker-activation": rendered = render_worker_activation_inputs( bom, deployment_id_prefix=args.deployment_id_prefix, expected_ownership_epoch=args.expected_ownership_epoch, ) output = Path(args.output_dir).resolve() for stem, value in rendered.items(): write_json(output / f"{stem}.json", value) else: activation_config_root = Path(args.activation_config_dir).resolve() activation_root = Path(args.activation_output_dir).resolve() activation_requests = { worker_file_stem(role["work_family"]): load_json( activation_config_root / f"{worker_file_stem(role['work_family'])}.json", f"Runtime Worker {role['work_family']} activation request", ) for role in bom["runtime_worker_roles"] } if args.activation_config_override: override = load_json( Path(args.activation_config_override).resolve(), "Runtime Worker activation request override", ) work_family = override.get("workFamily") if isinstance(override, dict) else None if not isinstance(work_family, str) or not work_family: raise ValidationFailure("Runtime Worker activation request override has no workFamily") stem = worker_file_stem(work_family) if stem not in activation_requests: raise ValidationFailure("Runtime Worker activation request override names another work owner") activation_requests[stem] = override activations = { worker_file_stem(role["work_family"]): load_json( activation_root / f"{worker_file_stem(role['work_family'])}.json", f"Runtime Worker {role['work_family']} activation output", ) for role in bom["runtime_worker_roles"] } rendered = render_worker_role_inputs( bom, load_json(Path(args.catalog_output).resolve(), "Catalog output"), activation_requests, activations, org_id=args.org_id, workload_id=args.workload_id, catalog_store_root=args.catalog_store_root, ) output = Path(args.output_dir).resolve() for stem, value in rendered.items(): write_json(output / f"{stem}.json", value) except ValidationFailure as exc: print(exc) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())