#!/usr/bin/env python3 """First App installation on local L0. No upgrades or deployment version checks.""" import argparse import base64 import hashlib import json import os from pathlib import Path import secrets import subprocess import sys import uuid from urllib.parse import quote from local_deployment import RUNTIME_RPC_SECONDS, cli, image_file, phase, read_env, read_json, run, worker_image_identity, write, write_env, write_json ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "deployments/offline/runtime")) from apply_runtime_app_shell import build_apply_sql, translate_navigation from render_serviceification_d3_integration_bindings import pack_sources, read_authority_snapshot, render_bindings from render_serviceification_test_stack import ( d3_catalog_consumer_capabilities, d3_partition_runtime_capability_grants, d3_session_scope_policy, d3_workload_capabilities, outbound_mail_routes_from_actions, ) def digest(value): return "sha256:" + hashlib.sha256((json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()).hexdigest() def create_runtime_secrets(state, workloads): write_json(state / "secrets/workload-identities.json", { "schema": "nexuscrm.runtime-workload-identities/v1", "identities": [{"workloadId": name, "token": secrets.token_hex(32)} for name in workloads], }) openssl = os.environ.get("OPENSSL", "openssl") key = subprocess.run((openssl, "genpkey", "-algorithm", "ED25519"), check=True, capture_output=True, timeout=10).stdout private = subprocess.run((openssl, "pkey", "-outform", "DER"), input=key, check=True, capture_output=True, timeout=10).stdout public = subprocess.run((openssl, "pkey", "-pubout", "-outform", "DER"), input=key, check=True, capture_output=True, timeout=10).stdout write_json(state / "secrets/signing-keys.json", { "schema": "nexuscrm.runtime-signing-keys/v1", "overlap": [], "active": {"signing_key_id": "local-runtime", "private_key_base64": base64.b64encode(private[-32:] + public[-32:]).decode()}, }) def catalog_request(selection, package, key, host, approval, command_id): app = selection["app_installation"] consumers = selection["catalog_consumers"] handlers = {row["capability_id"]: row["handler_ref"] for row in package["canonical_manifest"]["provided_capabilities"]} capabilities = [{"appCapabilityId": row["capability_id"], "runtimeCapabilityId": row["runtime_capability_id"], "handlerRef": handlers[row["capability_id"]]} for row in app["capabilities"]] frontend = selection["frontend"]["publication"]["digest"] result = { "schema": "nexuscrm.catalog-reconcile-config/v4", "orgId": "nexuscrm-local-org", "appId": package["canonical_manifest"]["app_id"], "expectedAppReleaseDigest": package["app_release_digest"], "metadataSourceDigest": approval["sourceDigest"], "releaseImport": {"packagePath": "/release/app-package.json", "attestationPath": "/release/publisher-attestation.json", "publisher": {"publisherId": key["publisher_id"], "keys": [{"keyId": key["key_id"], "algorithm": key["algorithm"], "publicKeyBase64": key["public_key"], "validFrom": "2026-01-01T00:00:00Z", "validUntil": "2036-01-01T00:00:00Z"}]}}, "artifactStoreRoot": "/var/lib/nexuscrm/catalog", "hostCompatibility": host, "idempotencyKey": command_id, "operationKind": "PREPARE", "appInstallationId": "", "expectedCatalogRevision": 0, "expectedSecurityEpoch": 0, "schemaGeneration": 1, "engineGeneration": 1, "runtimeAudience": "nexuscrm-runtime-api", "workloadId": consumers["handler_workload_id"], "authorizedWorkloadIds": consumers["authorized_workload_ids"], "capabilities": capabilities, "serviceReleaseDigest": package["app_release_digest"], "frontend": {"releaseDigest": frontend, "compatibilityReceiptDigest": digest({"localFrontend": frontend})}, "consumerGrants": [{"authorizedWorkloadId": row["authorized_workload_id"], "capabilityId": row["capability_id"]} for row in consumers["catalog_view_grants"]], "resolvedDependencies": [], "integrationBindings": [], "phaseOnePublication": {"commandId": command_id}, "approvalPublication": {"bundlePath": "/config/approval.json", "expectedDigest": digest(approval)}, } _, workload = d3_partition_runtime_capability_grants(consumers["runtime_capability_grants"]) result["handlerWorkloadCapabilities"] = sorted({row["capability_id"] for row in workload if row["authorized_workload_id"] == consumers["handler_workload_id"]}) return result def serving_config(selection, output, application_id): consumers = selection["catalog_consumers"] admission_lease = f"{RUNTIME_RPC_SECONDS + 1}s" capabilities = [] for row in output["capabilities"]: value = {"app_capability_id": row["appCapabilityId"], "runtime_capability_id": row["runtimeCapabilityId"], "expected_ownership_epoch": row["ownershipEpoch"], "expected_compatibility_receipt_digest": row["compatibilityReceiptDigest"], "authorized_workload_id": consumers["handler_workload_id"], "admission_lease": admission_lease} if row["runtimeCapabilityId"] in ("runtime.records.mutate.v1", "runtime.permissions.authoring.v1"): value["outcome_retention"] = "168h" if row["runtimeCapabilityId"] == "runtime.permissions.authoring.v1" else "24h" capabilities.append(value) grants = consumers["runtime_capability_grants"] catalog, workload = d3_partition_runtime_capability_grants(grants) return {"schema": "nexuscrm.runtime-api-serving/v4", "replica_id": "local-runtime", "org_id": output["orgId"], "app_installation_id": output["appInstallationId"], "catalog_snapshot_store_root": "/var/lib/nexuscrm/catalog", "expected_service_release_digest": output["serviceReleaseDigest"], "schema_generation": output["schemaGeneration"], "engine_generation": output["engineGeneration"], "identity_attempt_cap": "2s", "permission_state_protocol": "permission_state.v1", "authorized_workload_ids": consumers["authorized_workload_ids"], "capabilities": capabilities, "session_scope_capabilities": d3_session_scope_policy([row["runtime_capability_id"] for row in capabilities], grants), "catalog_consumer_capabilities": d3_catalog_consumer_capabilities(catalog, output["catalogSnapshotDigest"], admission_lease), "workload_capabilities": d3_workload_capabilities(workload, admission_lease, application_owner_id=application_id)} def catalog_job(image, root, request, *arguments, release=None, input_flag="-config"): settings = read_json(root / "l0/settings.json") command = ["docker", "run", "--rm", "--network", settings["project"] + "-backend", "--user", f"{os.getuid()}:{os.getgid()}", "--env-file", str(root / "l0/config/database.env"), "-v", f"{root / 'app'}:/config", "-v", f"{root / 'l0/data/catalog'}:/var/lib/nexuscrm/catalog"] if release: command += ["-v", f"{release}:/release:ro"] run((*command, image, *arguments, input_flag, "/config/" + request, "-output", "/config/" + request.replace(".json", "-output.json")), timeout=300) def prepare_workers(root, selection, output, catalog_image): settings = read_json(root / "l0/settings.json") info = json.loads(run(("docker", "image", "inspect", settings["worker_image"]), capture=True))[0] # These are runtime ownership inputs, read directly without comparing releases. worker_digest, worker_source = worker_image_identity(settings["worker_image"], info) write(root / "l0/config/catalog-controlplane-image", catalog_image + "\n") for policy in selection["runtime_worker_roles"]: family = policy["work_family"] stem = family.removeprefix("runtime.").removesuffix(".v1").replace(".", "-") request = {"schema": "nexuscrm.runtime-work-ownership-activation/v1", "operationId": "local-" + stem, "workFamily": family, "deploymentId": "local-" + stem, "workerReleaseDigest": worker_digest, "expectedOwnershipEpoch": "0"} name = "worker-" + stem + ".json" write_json(root / "app" / name, request) catalog_job(catalog_image, root, name, "work-ownership") active = read_json(root / "app" / name.replace(".json", "-output.json")) config = {key: value for key, value in policy.items() if key != "config_schema"} config.update({"schema": policy["config_schema"], "deployment_id": active["deploymentId"], "worker_release_digest": worker_digest, "worker_source_revision": worker_source, "ownership_epoch": int(active["ownershipEpoch"]), "permission_state_protocol": "permission_state.v1"}) if stem in ("record-event-dispatch", "rollup-delta", "scheduled-flow", "platform-command", "bulk-ingest"): capability = "runtime.automation.execute-flow.v1" if stem == "scheduled-flow" else "runtime.records.mutate.v1" binding = next(row for row in output["capabilities"] if row["runtimeCapabilityId"] == capability) config.update({"org_id": output["orgId"], "app_installation_id": output["appInstallationId"], "catalog_snapshot_store_root": "/var/lib/nexuscrm/catalog", "schema_generation": output["schemaGeneration"], "engine_generation": output["engineGeneration"], "runtime_capability_id": capability}) if stem in ("scheduled-flow", "bulk-ingest"): config.update({"app_capability_id": binding["appCapabilityId"], "expected_compatibility_receipt_digest": binding["compatibilityReceiptDigest"]}) if stem == "platform-command": config["runtime_capability_id"] = "runtime.commands.accept.v1" write_json(root / f"l0/config/workers/{stem}.json", config) write(root / f"l0/secrets/workers/{stem}.token", secrets.token_hex(32) + "\n") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--selection", type=Path, required=True, help="local installation settings; read without version or provenance checks") parser.add_argument("--app-packs", type=Path, required=True) parser.add_argument("--setup-binary", type=Path, required=True) parser.add_argument("--approval-publication", type=Path, required=True) parser.add_argument("--catalog-image", required=True) args = parser.parse_args() root = args.root.resolve() app_root = args.app_packs.resolve() state = root / "l0" for name in ("catalog", "files", "uploads"): (state / "data" / name).mkdir(parents=True, exist_ok=True) (root / "app").mkdir(exist_ok=True) env = {**os.environ, **read_env(root / "database.env"), "NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": str(app_root), "API_BASE_URL": read_env(root / "database.env")["PUBLIC_BASE_URL"]} stack = read_json(ROOT / "stack.lock.json") revision = run(("git", "-C", ROOT, "rev-parse", "HEAD"), capture=True).strip() command = [args.setup_binary.resolve(), "apply-offline-candidate", "--operation-id", "local-install"] for entry in stack["packs"]: source = entry if not entry.startswith("official_repo:"): uri = "urn:nexuscrm:release-pack:pingcap-inc:nexuscrm-pingcap-pack:" + quote(entry, safe="") source = "release_repo:" + uri command += ["--release-source-receipt", json.dumps({"uri": uri, "pack_root": str((ROOT / entry).resolve()), "revision": revision})] command += ["--source", source] with phase(root, "app-pack-install"): run(command, env=env, timeout=900) publish_app(args, root) print("App installed. Activate L0, then deploy L1.") def release_directory(app_root, artifact): path = Path(artifact) if path.is_absolute() or not (app_root / path).resolve().is_relative_to(app_root.resolve()): raise ValueError("AppRelease artifact must be relative to the AppPacks checkout") return (app_root / path).resolve().parent def publish_app(args, root): selection = read_json(args.selection) app_root = args.app_packs.resolve() state = root / "l0" env = read_env(root / "database.env") stack = read_json(ROOT / "stack.lock.json") shell = read_json(ROOT / stack["app_shell_config"]) with phase(root, "app-shell"): shell_sql = root / "app/shell.sql" write(shell_sql, build_apply_sql(shell, navigation_seeded_by_initializer=True)) with shell_sql.open() as source: subprocess.run(("mysql", "--protocol=tcp", "-h", env["TIDB_HOST"], "-P", env["TIDB_PORT"], "-u", env["TIDB_USER"], env["TIDB_DATABASE"]), stdin=source, check=True, timeout=60) write_json(root / "app/navigation.json", {"schema": "nexuscrm.application-navigation-initialization/v1", "owner_id": shell["frontendAppId"], "application_id": shell["id"], "template": translate_navigation(shell["navigationItems"], read_json(app_root / "appservices/sfcrm/navigation_pages.json"))}) catalog_job(args.catalog_image, root, "navigation.json", "application-navigation", input_flag="-input") release = release_directory(app_root, selection["app_installation"]["artifact"]["path"]) package = read_json(release / "app-package.json") key = read_json(release / "publisher-public-key.json") image_file(args.catalog_image, "/etc/nexuscrm/host-compatibility.json", root / "app/catalog-host.json") runtime_host = read_json(state / "config/host-compatibility.json") host = {"protocols": runtime_host["protocols"], "engineCapabilities": runtime_host["engineCapabilities"], "componentKinds": read_json(root / "app/catalog-host.json")["componentKinds"]} approval = read_json(args.approval_publication) write(root / "app/approval.json", args.approval_publication.read_text()) import time bits = (int(time.time() * 1000) << 80) | (7 << 76) | (secrets.randbits(12) << 64) | (2 << 62) | secrets.randbits(62) config = catalog_request(selection, package, key, host, approval, str(uuid.UUID(int=bits))) # Installed connector metadata supplies the Catalog routing facts. This is # metadata assembly, without a release/BOM compatibility preflight. owners = sorted(pack_sources(stack, app_root, ROOT)) authority = read_authority_snapshot(env, owners) config["integrationBindings"] = render_bindings(stack, authority, app_root, ROOT)["integrationBindings"] for entry in stack["packs"]: if entry.startswith("official_repo:"): continue actions = ROOT / entry / "actions/invocable_standard_actions.json" if actions.exists(): config.setdefault("outboundMailRoutes", []).extend(outbound_mail_routes_from_actions(actions, config["integrationBindings"])) config["approvalPublication"]["expectedDigest"] = "sha256:" + hashlib.sha256((root / "app/approval.json").read_bytes()).hexdigest() write_json(root / "app/catalog.json", config) with phase(root, "app-catalog"): catalog_job(args.catalog_image, root, "catalog.json", release=release) output = read_json(root / "app/catalog-output.json") application_id = read_json(ROOT / stack["app_shell_config"])["frontendAppId"] serving = serving_config(selection, output, application_id) with phase(root, "app-workers"): prepare_workers(root, selection, output, args.catalog_image) write_runtime_configuration(root, serving) def write_runtime_configuration(root, serving): state = root / "l0" create_runtime_secrets(state, serving["authorized_workload_ids"]) runtime_env = {**read_env(state / "config/database.env"), "NEXUSCRM_DB_MAX_OPEN_CONNS": "20", "NEXUSCRM_DB_MAX_IDLE_CONNS": "10", "FILE_STORAGE_PROVIDER": "filesystem", "FILE_STORAGE_ROOT": "/var/lib/nexuscrm/files", "FILE_STORAGE_LEGACY_UPLOAD_ROOT": "/var/lib/nexuscrm/uploads", "NEXUSCRM_PUBLIC_OAUTH_PKCE_ENABLED": "true"} write_env(state / "config/runtime.env", runtime_env) write_json(state / "config/serving.json", serving) if __name__ == "__main__": cli(main)