#!/usr/bin/env python3 """Render the D3 Catalog IntegrationBinding selection from frozen sources. The deployment owner supplies one read-only snapshot of the active immutable pack generations and exact route/Flow incarnations. This script selects only the integration components named by the frozen stack manifests. Catalog still revalidates every pack generation in its UPGRADE transaction; the generated document is not a metadata serving source. """ from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any, Optional SCRIPT_DIRECTORY = Path(__file__).resolve().parent if str(SCRIPT_DIRECTORY) not in sys.path: sys.path.insert(0, str(SCRIPT_DIRECTORY)) from render_serviceification_test_stack import ( # noqa: E402 D3_INTEGRATION_BINDINGS_SCHEMA, canonical_bytes, ) from validate_serviceification_bom import ( # noqa: E402 ValidationFailure, reject_duplicate_keys, ) OFFLINE_RUNTIME_DIRECTORY = SCRIPT_DIRECTORY.parent / "deployments" / "offline" / "runtime" if str(OFFLINE_RUNTIME_DIRECTORY) not in sys.path: sys.path.insert(0, str(OFFLINE_RUNTIME_DIRECTORY)) from common import DeployError, db_query_scalar, load_env_file # noqa: E402 AUTHORITY_SCHEMA = "nexuscrm.serviceification-d3-integration-authority/v1" WORKLOAD_ID = "integration-service" SOURCE_KINDS = { "connector_inbound_route": "CONNECTOR_INBOUND_ROUTE", "connector_callout_profile": "CONNECTOR_CALLOUT_PROFILE", "connector_field_mapping": "CONNECTOR_FIELD_MAPPING", "connector_push_contract": "CONNECTOR_PUSH_CONTRACT", } LOCAL_DECLARATION_KINDS = { "connector_inbound_route", "connector_callout_profile", } 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, UnicodeError, json.JSONDecodeError) as exc: raise ValidationFailure(f"{label} is not strict JSON: {exc}") from exc if not isinstance(value, dict): raise ValidationFailure(f"{label} must be a JSON object") return value def pack_sources( stack_lock: dict[str, Any], app_packs_root: Path, l2_root: Path, ) -> dict[str, tuple[Path, list[dict[str, Any]]]]: entries = stack_lock.get("packs") if not isinstance(entries, list) or not entries: raise ValidationFailure("D3 stack lock has no pack closure") result: dict[str, tuple[Path, list[dict[str, Any]]]] = {} for index, entry in enumerate(entries): if not isinstance(entry, str) or not entry: raise ValidationFailure(f"D3 stack pack {index} is invalid") if entry.startswith("official_repo:"): name = entry.removeprefix("official_repo:") root = app_packs_root / name else: root = l2_root / entry name = "" manifest = load_json(root / "pack.json", f"D3 pack manifest {entry}") manifest_name = manifest.get("name") components = manifest.get("components") if ( not isinstance(manifest_name, str) or not manifest_name or (name and manifest_name != name) or not isinstance(components, list) ): raise ValidationFailure(f"D3 pack manifest {entry} is incomplete") selected: list[dict[str, Any]] = [] identities: set[tuple[str, str]] = set() for component_index, component in enumerate(components): if not isinstance(component, dict): raise ValidationFailure( f"D3 pack {manifest_name} component {component_index} is invalid" ) kind = component.get("kind") if kind not in SOURCE_KINDS: continue business_key = component.get("businessKey") if not isinstance(business_key, str) or not business_key.startswith( manifest_name + "/" ): raise ValidationFailure( f"D3 pack {manifest_name} integration component has an invalid business key" ) identity = (kind, business_key) if identity in identities: raise ValidationFailure( f"D3 pack {manifest_name} has a duplicate integration component" ) identities.add(identity) selected.append({"kind": kind, "businessKey": business_key}) if selected: if manifest_name in result: raise ValidationFailure(f"D3 integration pack {manifest_name} is duplicated") result[manifest_name] = (root, selected) if not result: raise ValidationFailure("D3 stack closure selects no integration components") return result def declaration_identity(owner: str, row: dict[str, Any]) -> Optional[tuple[str, str]]: if not isinstance(row, dict) or not set(row).issubset( {"kind", "businessKey", "projection"} ) or not {"kind", "businessKey"}.issubset(row): raise ValidationFailure(f"D3 pack {owner} has a malformed Runtime declaration") kind, business_key = row["kind"], row["businessKey"] if not isinstance(kind, str) or not isinstance(business_key, str) or not business_key: raise ValidationFailure(f"D3 pack {owner} has an invalid integration declaration") if kind not in SOURCE_KINDS: return None if kind in LOCAL_DECLARATION_KINDS: business_key = owner + "/" + business_key return kind, business_key def _sql_literal(value: str) -> str: return "'" + value.replace("'", "''") + "'" def _decode_hex(value: str, label: str) -> str: try: return bytes.fromhex(value).decode("utf-8") except (ValueError, UnicodeError) as exc: raise ValidationFailure(f"{label} is not canonical UTF-8 authority") from exc def _query_rows(envv: dict[str, str], sql: str, columns: int, label: str) -> list[list[str]]: raw = db_query_scalar(envv, sql, database=envv["TIDB_DATABASE"]) if not raw: return [] rows = [line.split("\t") for line in raw.splitlines()] if any(len(row) != columns for row in rows): raise ValidationFailure(f"{label} returned a malformed authority row") return rows def read_authority_snapshot( envv: dict[str, str], owners: list[str], ) -> dict[str, Any]: generations: list[dict[str, Any]] = [] for owner in owners: quoted = _sql_literal(owner) rows = _query_rows(envv, f""" SELECT HEX(g.PackName), HEX(g.PackInstallId), HEX(g.AppliedPayloadHash), HEX(g.RuntimeStartupShallowDeclarations) FROM _System_PackActivation AS a JOIN _System_PackGeneration AS g ON g.PackInstallId = a.ActivePackInstallId WHERE a.PackName = {quoted} AND a.ActivationEpoch > 0 AND a.ActivatedDate IS NOT NULL AND COALESCE(a.IsDeleted, 0) = 0 AND g.PackName = {quoted} AND g.Status = 'active' AND g.DesiredPayloadHash = g.AppliedPayloadHash AND COALESCE(g.IsDeleted, 0) = 0 AND g.RuntimeStartupShallowDeclarations IS NOT NULL """, 4, f"D3 active generation {owner}") if len(rows) != 1: raise ValidationFailure( f"D3 pack {owner} does not have exactly one active immutable generation" ) pack_name, installation, digest, declarations_raw = ( _decode_hex(rows[0][index], f"D3 active generation {owner} field {index}") for index in range(4) ) try: declarations = json.loads( declarations_raw, object_pairs_hook=reject_duplicate_keys, ) except (UnicodeError, json.JSONDecodeError) as exc: raise ValidationFailure( f"D3 pack {owner} immutable Runtime declaration is invalid" ) from exc route_rows = _query_rows(envv, f""" SELECT HEX(CONCAT(r.Namespace, '/', r.EventKind)), CAST(r.Version AS CHAR), HEX(r.FlowDeveloperName), HEX(COALESCE(fd.OwnerPack, '')), HEX(COALESCE(fd.ActiveVersionId, '')), CAST(COALESCE(fd.IsCustomized, 0) AS CHAR) FROM _System_ConnectorInboundRoute AS r JOIN FlowDefinition AS fd ON fd.DeveloperName = r.FlowDeveloperName WHERE r.Namespace = {quoted} AND r.InstalledPackId = {_sql_literal(installation)} AND r.IsCurrent = 1 AND COALESCE(r.IsDeleted, 0) = 0 AND COALESCE(fd.IsDeleted, 0) = 0 ORDER BY r.EventKind """, 6, f"D3 route authority {owner}") routes: list[dict[str, Any]] = [] for row in route_rows: try: version = int(row[1]) customized = int(row[5]) except ValueError as exc: raise ValidationFailure(f"D3 route authority {owner} has invalid epochs") from exc if customized not in {0, 1}: raise ValidationFailure(f"D3 route authority {owner} has invalid customization state") routes.append({ "businessKey": _decode_hex(row[0], "D3 route business key"), "version": version, "flowDeveloperName": _decode_hex(row[2], "D3 route Flow"), "flowOwnerPack": _decode_hex(row[3], "D3 route Flow owner"), "flowVersionId": _decode_hex(row[4], "D3 route Flow version"), "flowCustomized": customized == 1, }) generations.append({ "ownerPack": pack_name, "packInstallationId": installation, "packPayloadDigest": digest, "runtimeStartupShallowDeclarations": declarations, "routes": routes, }) return {"schema": AUTHORITY_SCHEMA, "activePackGenerations": generations} def callout_credentials(row: dict[str, Any], owner: str, business_key: str) -> list[str]: raw = row.get("projection") if not isinstance(raw, str) or not raw: raise ValidationFailure(f"D3 callout profile {business_key} has no immutable projection") try: projection = json.loads(raw, object_pairs_hook=reject_duplicate_keys) except (UnicodeError, json.JSONDecodeError) as exc: raise ValidationFailure( f"D3 callout profile {business_key} projection is invalid" ) from exc local_name = business_key.removeprefix(owner + "/") if ( not isinstance(projection, dict) or projection.get("ownerPack") != owner or projection.get("profileName") != local_name or not isinstance(projection.get("credentialRef"), str) or not projection["credentialRef"] ): raise ValidationFailure( f"D3 callout profile {business_key} projection identity is invalid" ) return [projection["credentialRef"]] def render_bindings( stack_lock: dict[str, Any], authority: dict[str, Any], app_packs_root: Path, l2_root: Path, ) -> dict[str, Any]: sources = pack_sources(stack_lock, app_packs_root, l2_root) if set(authority) != {"schema", "activePackGenerations"} or authority.get( "schema" ) != AUTHORITY_SCHEMA: raise ValidationFailure("D3 integration authority snapshot has an unknown shape") generations = authority.get("activePackGenerations") if not isinstance(generations, list): raise ValidationFailure("D3 integration authority snapshot has no generations") by_owner: dict[str, dict[str, Any]] = {} for generation in generations: if not isinstance(generation, dict): raise ValidationFailure("D3 integration authority generation is invalid") owner = generation.get("ownerPack") if not isinstance(owner, str) or owner in by_owner: raise ValidationFailure("D3 integration authority mixes pack generations") by_owner[owner] = generation if set(by_owner) != set(sources): raise ValidationFailure("D3 integration authority does not match the frozen stack packs") bindings: list[dict[str, Any]] = [] for owner, (_, components) in sources.items(): generation = by_owner[owner] required_fields = { "ownerPack", "packInstallationId", "packPayloadDigest", "runtimeStartupShallowDeclarations", "routes", } if set(generation) != required_fields: raise ValidationFailure(f"D3 integration generation {owner} has an unknown shape") installation = generation["packInstallationId"] digest = generation["packPayloadDigest"] declarations = generation["runtimeStartupShallowDeclarations"] routes = generation["routes"] if ( not isinstance(installation, str) or not installation or not isinstance(digest, str) or not digest.startswith("nexus-pack-sha256-v1:") or not isinstance(declarations, list) or not isinstance(routes, list) ): raise ValidationFailure(f"D3 integration generation {owner} is incomplete") declaration_by_identity: dict[tuple[str, str], dict[str, Any]] = {} for row in declarations: identity = declaration_identity(owner, row) if identity is None: continue if identity in declaration_by_identity: raise ValidationFailure(f"D3 pack {owner} has an ambiguous Runtime declaration") declaration_by_identity[identity] = row expected = {(component["kind"], component["businessKey"]) for component in components} if set(declaration_by_identity) != expected: raise ValidationFailure( f"D3 pack {owner} immutable declarations disagree with its frozen manifest" ) route_by_key: dict[str, dict[str, Any]] = {} for route in routes: if not isinstance(route, dict) or set(route) != { "businessKey", "version", "flowDeveloperName", "flowOwnerPack", "flowVersionId", "flowCustomized", }: raise ValidationFailure(f"D3 pack {owner} route authority is malformed") key = route["businessKey"] if not isinstance(key, str) or key in route_by_key: raise ValidationFailure(f"D3 pack {owner} route authority is ambiguous") route_by_key[key] = route expected_routes = { component["businessKey"] for component in components if component["kind"] == "connector_inbound_route" } if set(route_by_key) != expected_routes: raise ValidationFailure(f"D3 pack {owner} route authority is incomplete") for component in components: kind, business_key = component["kind"], component["businessKey"] credentials: list[str] = [] runtime_command: dict[str, Any] | None = None definition_version = 1 if kind == "connector_callout_profile": credentials = callout_credentials( declaration_by_identity[(kind, business_key)], owner, business_key, ) if kind == "connector_inbound_route": route = route_by_key[business_key] version = route["version"] flow_name = route["flowDeveloperName"] flow_owner = route["flowOwnerPack"] flow_version = route["flowVersionId"] if ( not isinstance(version, int) or isinstance(version, bool) or version < 1 or not isinstance(flow_name, str) or not flow_name or not isinstance(flow_owner, str) or flow_owner not in sources or flow_owner == owner or not isinstance(flow_version, str) or not flow_version or route["flowCustomized"] is not False ): raise ValidationFailure( f"D3 route {business_key} is not an exact cross-pack Flow incarnation" ) definition_version = version runtime_command = { "commandKind": "runtime.flow.execute", "payloadFormatVersion": 2, "definitionRef": flow_name, "definitionVersion": flow_version, } binding = { "authorizedWorkloadId": WORKLOAD_ID, "kind": SOURCE_KINDS[kind], "ownerPack": owner, "packInstallationId": installation, "packPayloadDigest": digest, "businessKey": business_key, "definitionVersion": definition_version, "credentialRefs": sorted(credentials), } if runtime_command is not None: binding["runtimeCommand"] = runtime_command bindings.append(binding) bindings.sort(key=lambda value: ( value["authorizedWorkloadId"], value["kind"], value["ownerPack"], value["packInstallationId"], value["businessKey"], )) return {"schema": D3_INTEGRATION_BINDINGS_SCHEMA, "integrationBindings": bindings} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--stack-lock", required=True) authority_source = parser.add_mutually_exclusive_group(required=True) authority_source.add_argument("--authority") authority_source.add_argument("--runtime-env") parser.add_argument("--app-packs-root", required=True) parser.add_argument("--l2-root", required=True) parser.add_argument("--output", required=True) args = parser.parse_args() try: stack_lock = load_json(Path(args.stack_lock).resolve(), "D3 stack lock") app_packs_root = Path(args.app_packs_root).resolve() l2_root = Path(args.l2_root).resolve() if args.runtime_env: sources = pack_sources(stack_lock, app_packs_root, l2_root) authority = read_authority_snapshot( load_env_file(Path(args.runtime_env).resolve()), sorted(sources), ) else: authority = load_json( Path(args.authority).resolve(), "D3 integration authority", ) result = render_bindings( stack_lock, authority, app_packs_root, l2_root, ) output = Path(args.output) if output.exists() or output.is_symlink(): raise ValidationFailure("D3 integration binding output already exists") output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(canonical_bytes(result)) except (ValidationFailure, DeployError, OSError) as exc: print(f"error: {exc}") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())