#!/usr/bin/env python3 """Materialize the next secret-bearing Prod D3 target from live authority.""" from __future__ import annotations import argparse import json import os import re import shlex import shutil import subprocess import sys import tempfile from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[3] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.connected_mail_oauth import connected_mail_oauth_env # noqa: E402 try: from .prod_d3_ingress import ( IngressError, PROD_SLOT_PORTS, PROD_SERVER_NAME, ProdRoute, next_slot, parse_prod_route, ) from .prod_d3_authority import CURRENT_SCHEMA, IN_PROGRESS_SCHEMA except ImportError: # Direct script execution adds this directory to sys.path. from prod_d3_ingress import ( IngressError, PROD_SLOT_PORTS, PROD_SERVER_NAME, ProdRoute, next_slot, parse_prod_route, ) from prod_d3_authority import CURRENT_SCHEMA, IN_PROGRESS_SCHEMA TAG_RE = re.compile(r"staging-d3-release-[A-Za-z0-9_.-]+") COMMIT_RE = re.compile(r"[0-9a-f]{40}") TREE_DIGEST_RE = re.compile(r"sha256:[0-9a-f]{64}") BOM_RE = re.compile( r"ghcr\.io/pingcap-inc/nexuscrm-serviceification-bom@sha256:[0-9a-f]{64}" ) PROD_NGINX_CONF = Path("/etc/nginx/conf.d/nexuscrm-prod.conf") DEFAULT_MULTI_PROD_ROOT = Path( "/home/ubuntu/nexuscrm-deployment/online/envs/multi_prod" ) DEPLOYMENT_CONFIG = REPO_ROOT / "deployments" / "deployment-config.json" RETIRED_RELEASE_ENV_KEYS = frozenset( {"D3_PATCH_BUNDLE_PATH", "D3_PATCH_TREE_DIGEST"} ) WORKFLOW_AUTHORITY_KEYS = ( "GH_TOKEN", "TIDB_PORT", "TIDB_DATABASE", "TIDB_PASSWORD", "JWT_SECRET", "NEXUS_ADMIN_PASSWORD", "NEXUSCRM_NC_MASTER_KEY", "GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET", "REACT_APP_GOOGLE_CLIENT_ID", "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_ID", "NEXUSCRM_INTEGRATION_CONNECTED_MAIL_OAUTH_CLIENT_SECRET", "EXP_HUBSPOT_APP_NAME", "EXP_HUBSPOT_APP_ID", "EXP_HUBSPOT_CLIENT_ID", "EXP_HUBSPOT_CLIENT_SECRET", "EXP_HUBSPOT_ACCESS_TOKEN", "EXP_HUBSPOT_WEBHOOK_URL", "EXP_HUBSPOT_WORKFLOW_ACTION_SIGNING_SECRET", "EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "EXP_HUBSPOT_WORKFLOW_ACTION_ID", "EXP_HUBSPOT_EXTERNAL_ADAPTER_BASE_URL", ) REQUIRED_RUNTIME_KEYS = ( "TIDB_HOST", "TIDB_PORT", "TIDB_DATABASE", "TIDB_USER", "TIDB_PASSWORD", "JWT_SECRET", "NEXUS_ADMIN_PASSWORD", "NEXUSCRM_NC_MASTER_KEY", "EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "D3_RESOLVED_BOM_REFERENCE", ) class PrepareError(ValueError): pass def _non_prod_tidb_users() -> set[str]: try: config = json.loads(DEPLOYMENT_CONFIG.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PrepareError("cannot load reviewed TiDB environment bindings") from exc users: set[str] = set() for env_name, env_config in config.get("environments", {}).items(): if env_name == "multi_prod" or not isinstance(env_config, dict): continue topology = env_config.get("topology", {}) if not isinstance(topology, dict): continue for user in topology.get("additional_non_prod_tidb_users", []): value = str(user).strip() if value: users.add(value) target_config = topology.get("target_config", {}) if not isinstance(target_config, dict): continue for target in target_config.values(): if not isinstance(target, dict): continue tidb = target.get("tidb", {}) if isinstance(tidb, dict): value = str(tidb.get("user", "")).strip() if value: users.add(value) return users def _reviewed_prod_tidb_binding() -> tuple[str, str]: host = os.environ.get("PROD_TIDB_HOST", "").strip() user = os.environ.get("PROD_RUNTIME_TIDB_USER", "").strip() if not host or not user: raise PrepareError( "Prod D3 requires reviewed PROD_TIDB_HOST and " "PROD_RUNTIME_TIDB_USER bindings" ) return host, user def _require_prod_tidb_binding(authority: dict[str, str]) -> None: expected_host, expected_user = _reviewed_prod_tidb_binding() if ( authority.get("TIDB_HOST") != expected_host or authority.get("TIDB_USER") != expected_user or authority.get("TIDB_PORT") != "4000" or authority.get("TIDB_DATABASE") != "nexuscrm" ): raise PrepareError("Prod D3 TiDB binding is not the reviewed production target") lowered = expected_user.lower() if lowered == "root" or lowered.endswith(".root"): raise PrepareError("Prod D3 refuses a root TiDB runtime identity") if expected_user in _non_prod_tidb_users(): raise PrepareError("Prod D3 TiDB user is assigned to a non-Prod target") def _require_tag_bom_reference( reference: str, tag_target_commit: str ) -> None: pulled = subprocess.run( ["docker", "pull", "--quiet", reference], capture_output=True, text=True, check=False, ) if pulled.returncode != 0: detail = pulled.stderr.strip() or f"docker pull exited {pulled.returncode}" raise PrepareError( "cannot resolve the Tag-selected BOM before publishing authority: " + detail ) inspected = subprocess.run( [ "docker", "image", "inspect", "--format", '{{ index .Config.Labels "io.nexuscrm.serviceification.desired-bom-sha256" }}', reference, ], capture_output=True, text=True, check=False, ) bom_binding = inspected.stdout.strip() if inspected.returncode != 0 or not TREE_DIGEST_RE.fullmatch(bom_binding): detail = ( inspected.stderr.strip() or f"docker image inspect exited {inspected.returncode}" ) raise PrepareError( "Tag-selected OCI artifact has no exact BOM binding: " + detail ) created = subprocess.run( ["docker", "create", reference, "/__nexuscrm_bom_admission_check__"], capture_output=True, text=True, check=False, ) container = created.stdout.strip() if created.returncode != 0 or not re.fullmatch(r"[0-9a-f]{64}", container): detail = created.stderr.strip() or f"docker create exited {created.returncode}" raise PrepareError( "cannot inspect the Tag-selected BOM before publishing authority: " + detail ) try: with tempfile.TemporaryDirectory(prefix="nexuscrm-prod-d3-admission-") as temp: embedded = Path(temp) / "resolved-bom.json" copied = subprocess.run( [ "docker", "cp", f"{container}:/nexuscrm/release/resolved-bom.json", str(embedded), ], capture_output=True, text=True, check=False, ) if copied.returncode != 0 or not embedded.is_file(): detail = ( copied.stderr.strip() or f"docker cp exited {copied.returncode}" ) raise PrepareError( "cannot extract the Tag-selected BOM before publishing " "authority: " + detail ) try: bom = json.loads(embedded.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PrepareError( "Tag-selected OCI artifact has no valid resolved BOM" ) from exc l2_metadata = bom.get("l2_metadata") if isinstance(bom, dict) else None if ( not isinstance(l2_metadata, dict) or l2_metadata.get("commit") != tag_target_commit ): raise PrepareError( "Tag-selected OCI BOM L2 commit differs from the release Tag" ) if bom.get("desired_bom_sha256") != bom_binding: raise PrepareError( "Tag-selected OCI artifact binding differs from its embedded BOM" ) finally: subprocess.run( ["docker", "rm", "-f", container], capture_output=True, text=True, check=False, ) def load_env(path: Path) -> dict[str, str]: values: dict[str, str] = {} try: lines = path.read_text(encoding="utf-8").splitlines() except OSError as exc: raise PrepareError(f"cannot read active Prod runtime env: {path}") from exc for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, value = stripped.split("=", 1) try: tokens = shlex.split(value, comments=False, posix=True) except ValueError as exc: raise PrepareError(f"invalid shell quoting for runtime key {key}") from exc if len(tokens) > 1: raise PrepareError(f"runtime value for {key} is not shell-safe") values[key] = tokens[0] if tokens else "" return values def update_env( path: Path, updates: dict[str, str], *, remove_keys: frozenset[str] = frozenset(), ) -> None: if set(updates) & remove_keys: raise PrepareError("runtime env key cannot be updated and removed together") for key, value in updates.items(): if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key): raise PrepareError(f"invalid runtime key: {key}") if any(char in value for char in ("\x00", "\r", "\n")): raise PrepareError(f"runtime value for {key} contains a line break") lines = path.read_text(encoding="utf-8").splitlines() seen: set[str] = set() output: list[str] = [] for line in lines: if not line or line.lstrip().startswith("#") or "=" not in line: output.append(line) continue key = line.split("=", 1)[0] if key in remove_keys: continue if key in updates: output.append(f"{key}={shlex.quote(updates[key])}") seen.add(key) else: output.append(line) output.extend( f"{key}={shlex.quote(value)}" for key, value in updates.items() if key not in seen ) pending = path.with_name(f".{path.name}.pending") pending.write_text("\n".join(output) + "\n", encoding="utf-8") os.chmod(pending, 0o600) os.replace(pending, path) def _fsync_dir(path: Path) -> None: descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) try: os.fsync(descriptor) finally: os.close(descriptor) def _read_json_authority(path: Path, label: str) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise PrepareError(f"{label} is unavailable: {path}") try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise PrepareError(f"{label} is invalid: {path}") from exc if not isinstance(value, dict): raise PrepareError(f"{label} must be a JSON object") return value def read_in_progress_authority(path: Path) -> dict[str, Any]: return _read_json_authority(path, "Prod D3 in-progress authority") def _write_in_progress(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) pending = path.with_name(f".{path.name}.pending") if pending.exists() or pending.is_symlink(): raise PrepareError("Prod D3 in-progress pending authority already exists") data = json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n" try: with pending.open("x", encoding="utf-8") as stream: os.chmod(pending, 0o600) stream.write(data) stream.flush() os.fsync(stream.fileno()) os.replace(pending, path) _fsync_dir(path.parent) except OSError as exc: if pending.exists() and not pending.is_symlink(): pending.unlink() raise PrepareError("cannot publish durable Prod D3 in-progress authority") from exc def _require_current_route_reconciled( multi_prod_root: Path, route: ProdRoute ) -> None: d3_root = multi_prod_root / "d3" marker_path = d3_root / "current.json" if not route.is_d3: d3_current = d3_root / "current" if ( marker_path.exists() or marker_path.is_symlink() or d3_current.exists() or d3_current.is_symlink() ): raise PrepareError( "classic Prod route disagrees with retained D3 current authority" ) current_env = multi_prod_root / "current" / "env" / "runtime.env" if not current_env.is_file(): raise PrepareError("classic Prod current runtime env is unavailable") current_values = load_env(current_env.resolve()) if current_values.get("BACKEND_PORT") != str(route.backend_port): raise PrepareError( "classic Prod current runtime env disagrees with live nginx route" ) current_root_value = current_values.get("ENV_ROOT", "").strip() if not current_root_value: raise PrepareError("classic Prod current runtime env has no release root") current_root = Path(current_root_value).expanduser().resolve() current_frontend = multi_prod_root / "current" / "frontend-dist" expected_env = current_root / "env" / "runtime.env" expected_frontend = current_root / "frontend-dist" if ( not current_root.is_dir() or not expected_env.is_file() or current_env.resolve() != expected_env.resolve() or not expected_frontend.is_dir() or not current_frontend.is_dir() or current_frontend.resolve() != expected_frontend.resolve() or not current_frontend.resolve().is_relative_to(current_root) ): raise PrepareError("classic Prod current pointer is not a valid live release") return marker = _read_json_authority(marker_path, "Prod D3 current marker") release_root_raw = marker.get("releaseRoot") current_link = d3_root / "current" if ( marker.get("schema") != CURRENT_SCHEMA or marker.get("slot") != route.slot or marker.get("backendPort") != route.backend_port or not isinstance(marker.get("releaseTag"), str) or not TAG_RE.fullmatch(marker["releaseTag"]) or not isinstance(marker.get("resolvedBomReference"), str) or not BOM_RE.fullmatch(marker["resolvedBomReference"]) or not isinstance(release_root_raw, str) or not release_root_raw ): raise PrepareError("Prod D3 current marker disagrees with live nginx route") release_root = Path(release_root_raw).expanduser().resolve() releases_root = (d3_root / "releases").resolve() if ( not current_link.is_symlink() or current_link.resolve() != release_root or release_root.parent != releases_root or not release_root.is_dir() ): raise PrepareError("Prod D3 current pointer disagrees with its current marker") current_env = current_link / "env" / "runtime.env" current_frontend = current_link / "frontend-dist" if ( not current_env.is_file() or not current_frontend.is_dir() or not current_frontend.resolve().is_relative_to(release_root) ): raise PrepareError("Prod D3 current release env or frontend is unavailable") current_values = load_env(current_env.resolve()) current_env_root = Path( current_values.get("ENV_ROOT", "") ).expanduser().resolve() expected = { "ENV_NAME": f"multi-prod-d3-{route.slot}", "BACKEND_PORT": str(route.backend_port), "RELEASE_TAG": marker["releaseTag"], "D3_RESOLVED_BOM_REFERENCE": marker["resolvedBomReference"], } if current_env_root != release_root or any( current_values.get(key) != value for key, value in expected.items() ): raise PrepareError("Prod D3 current runtime env disagrees with its marker") def require_in_progress_authority( payload: dict[str, Any], *, release_tag: str, tag_target_commit: str, resolved_bom_reference: str, release_root: Path, ) -> dict[str, Any]: expected_keys = { "schema", "releaseTag", "tagTargetCommit", "resolvedBomReference", "previousBackendPort", "previousRouteKind", "targetSlot", "targetBackendPort", "releaseRoot", } if set(payload) != expected_keys or payload.get("schema") != IN_PROGRESS_SCHEMA: raise PrepareError("Prod D3 in-progress authority has an invalid schema") if ( payload.get("releaseTag") != release_tag or payload.get("tagTargetCommit") != tag_target_commit or payload.get("resolvedBomReference") != resolved_bom_reference or payload.get("releaseRoot") != str(release_root) ): raise PrepareError("another Prod D3 release owns the durable target") target_slot = payload.get("targetSlot") target_port = payload.get("targetBackendPort") previous_port = payload.get("previousBackendPort") if ( target_slot not in PROD_SLOT_PORTS or target_port != PROD_SLOT_PORTS[target_slot] or previous_port not in PROD_SLOT_PORTS.values() or target_port == previous_port or payload.get("previousRouteKind") not in {"d3", "legacy"} ): raise PrepareError("Prod D3 in-progress authority has an invalid slot binding") return payload def clear_in_progress( *, d3_root: Path, release_tag: str, tag_target_commit: str, resolved_bom_reference: str, target_slot: str, target_backend_port: int, release_root: Path, ) -> None: path = d3_root / "in-progress.json" payload = require_in_progress_authority( _read_json_authority(path, "Prod D3 in-progress authority"), release_tag=release_tag, tag_target_commit=tag_target_commit, resolved_bom_reference=resolved_bom_reference, release_root=release_root, ) if ( payload["targetSlot"] != target_slot or payload["targetBackendPort"] != target_backend_port ): raise PrepareError("Prod D3 in-progress authority disagrees with live target") path.unlink() _fsync_dir(d3_root) def _active_env_source(multi_prod_root: Path, route: ProdRoute) -> Path: candidate = ( multi_prod_root / "d3" / "current" / "env" / "runtime.env" if route.is_d3 else multi_prod_root / "current" / "env" / "runtime.env" ) if candidate.is_file(): return candidate.resolve() raise PrepareError("no active Prod runtime env is available for secret authority") def _active_frontend_source(multi_prod_root: Path, route: ProdRoute) -> Path: candidate = ( multi_prod_root / "d3" / "current" / "frontend-dist" if route.is_d3 else multi_prod_root / "current" / "frontend-dist" ) if candidate.is_dir(): return candidate.resolve() raise PrepareError("no active Prod frontend is available for capture") def prepare( *, release_tag: str, tag_target_commit: str, multi_prod_root: Path, nginx_conf: Path, ) -> dict[str, str]: if not TAG_RE.fullmatch(release_tag): raise PrepareError("release tag is not a staging-d3-release tag") if not COMMIT_RE.fullmatch(tag_target_commit): raise PrepareError("tag target commit is not exact") bom_reference = os.environ.get("D3_RESOLVED_BOM_REFERENCE", "").strip() if not BOM_RE.fullmatch(bom_reference): raise PrepareError("Prod D3 requires this Tag's exact resolved BOM reference") connected_mail_oauth_env(os.environ, error_type=PrepareError) if nginx_conf.is_symlink() or not nginx_conf.is_file(): raise PrepareError(f"Prod nginx authority is unavailable: {nginx_conf}") try: route = parse_prod_route(nginx_conf.read_text(encoding="utf-8")) except (IngressError, OSError) as exc: raise PrepareError(f"cannot resolve current Prod route: {exc}") from exc d3_root = (multi_prod_root / "d3").resolve() release_root = (d3_root / "releases" / release_tag).resolve() if d3_root not in release_root.parents: raise PrepareError("release root escaped the Prod D3 authority") env_file = release_root / "env" / "runtime.env" in_progress_file = d3_root / "in-progress.json" new_in_progress = False if in_progress_file.exists() or in_progress_file.is_symlink(): pending = require_in_progress_authority( _read_json_authority( in_progress_file, "Prod D3 in-progress authority" ), release_tag=release_tag, tag_target_commit=tag_target_commit, resolved_bom_reference=bom_reference, release_root=release_root, ) target_slot = str(pending["targetSlot"]) target_port = int(pending["targetBackendPort"]) previous_port = int(pending["previousBackendPort"]) if route.backend_port not in {previous_port, target_port}: raise PrepareError("live nginx route escaped the durable Prod D3 target") if route.backend_port == previous_port and ( route.is_d3 != (pending["previousRouteKind"] == "d3") ): raise PrepareError("live predecessor route kind changed during D3 release") if route.backend_port == target_port and not route.is_d3: raise PrepareError("durable Prod D3 target is live without D3 ownership") if route.backend_port == previous_port: _require_current_route_reconciled(multi_prod_root, route) source_env = _active_env_source(multi_prod_root, route) source_frontend = _active_frontend_source(multi_prod_root, route) else: source_env = env_file source_frontend = release_root / "frontend-dist" if ( source_env.is_symlink() or not source_env.is_file() or not source_frontend.is_dir() ): raise PrepareError( "live in-progress Prod D3 target has no retained env or frontend" ) else: _require_current_route_reconciled(multi_prod_root, route) source_env = _active_env_source(multi_prod_root, route) source_frontend = _active_frontend_source(multi_prod_root, route) target_slot = next_slot(route) target_port = PROD_SLOT_PORTS[target_slot] pending = { "schema": IN_PROGRESS_SCHEMA, "releaseTag": release_tag, "tagTargetCommit": tag_target_commit, "resolvedBomReference": bom_reference, "previousBackendPort": route.backend_port, "previousRouteKind": "d3" if route.is_d3 else "legacy", "targetSlot": target_slot, "targetBackendPort": target_port, "releaseRoot": str(release_root), } new_in_progress = True _require_tag_bom_reference(bom_reference, tag_target_commit) source_authority = load_env(source_env) preflight_authority = dict(source_authority) for key in WORKFLOW_AUTHORITY_KEYS: value = os.environ.get(key, "").strip() if value: preflight_authority[key] = value connected_mail_oauth_env(preflight_authority, error_type=PrepareError) reviewed_tidb_host, reviewed_tidb_user = _reviewed_prod_tidb_binding() preflight_authority["TIDB_HOST"] = reviewed_tidb_host preflight_authority["TIDB_USER"] = reviewed_tidb_user preflight_authority["D3_RESOLVED_BOM_REFERENCE"] = bom_reference missing_preflight = [ key for key in REQUIRED_RUNTIME_KEYS if not preflight_authority.get(key) ] if missing_preflight: raise PrepareError( "Prod D3 runtime authority is incomplete: " + ", ".join(missing_preflight) ) _require_prod_tidb_binding(preflight_authority) new_release_env = not (env_file.exists() or env_file.is_symlink()) if not new_release_env: if env_file.is_symlink() or not env_file.is_file(): raise PrepareError("retained Prod D3 runtime env is not a regular file") retained = load_env(env_file) retained_slot = retained.get("ENV_NAME", "").removeprefix("multi-prod-d3-") if retained_slot not in PROD_SLOT_PORTS: raise PrepareError("retained Prod D3 runtime env has an invalid slot") expected = { "ENV_ROOT": str(release_root), "RELEASE_TAG": release_tag, "TAG_TARGET_COMMIT": tag_target_commit, "ENV_NAME": f"multi-prod-d3-{target_slot}", "BACKEND_PORT": str(target_port), "D3_RESOLVED_BOM_REFERENCE": bom_reference, "PROD_D3_PREVIOUS_BACKEND_PORT": str(pending["previousBackendPort"]), "PROD_D3_PREVIOUS_ROUTE_KIND": str(pending["previousRouteKind"]), } if any(retained.get(key) != value for key, value in expected.items()): raise PrepareError("retained Prod D3 runtime env belongs to another release") else: if target_port == route.backend_port: raise PrepareError("Prod D3 target port is already live") if new_in_progress: _write_in_progress(in_progress_file, pending) release_root.mkdir(mode=0o700, parents=True, exist_ok=True) env_file.parent.mkdir(mode=0o700, parents=True, exist_ok=True) for child in ("logs", "run", "artifacts"): (release_root / child).mkdir(mode=0o700, exist_ok=True) if new_release_env: shutil.copyfile(source_env, env_file) os.chmod(env_file, 0o600) updates = { "ENV_NAME": f"multi-prod-d3-{target_slot}", "ENV_ROOT": str(release_root), "RELEASE_TAG": release_tag, "TAG_TARGET_COMMIT": tag_target_commit, "BACKEND_PORT": str(target_port), "PORT": str(target_port), "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT": str(target_port + 101), "HUBSPOT_OPTIONS_ADAPTER_SERVICE_NAME": ( f"nexuscrm-multi-prod-d3-{target_slot}-hubspot-options-adapter.service" ), "PUBLIC_BASE_URL": f"https://{PROD_SERVER_NAME}", "API_BASE_URL": f"https://{PROD_SERVER_NAME}", "FRONTEND_URL": f"https://{PROD_SERVER_NAME}", "SERVER_NAME": PROD_SERVER_NAME, "NGINX_LISTEN_PORT": "80", "NGINX_FORWARDED_PROTO": "https", "INSTALL_SYSTEMD": "1", "INSTALL_NGINX": "0", "RESET_DB": "0", "RESET_TIDB_BRANCH": "0", "SKIP_BOOTSTRAP": "false", "ALLOW_PROD_SKIP_BOOTSTRAP": "false", "NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS": "true", "VERIFY_USERNAME": "admin@example.invalid", # The current Tag must always overwrite a copied or retained value. "D3_RESOLVED_BOM_REFERENCE": bom_reference, } if new_release_env: updates.update( { "PROD_D3_PREVIOUS_BACKEND_PORT": str(pending["previousBackendPort"]), "PROD_D3_PREVIOUS_ROUTE_KIND": str(pending["previousRouteKind"]), } ) # Explicit workflow authority wins over retained values. Values are never # printed or returned; the output file remains mode 0600. for key in WORKFLOW_AUTHORITY_KEYS: value = os.environ.get(key, "").strip() if value: updates[key] = value updates["TIDB_HOST"] = reviewed_tidb_host updates["TIDB_USER"] = reviewed_tidb_user update_env(env_file, updates, remove_keys=RETIRED_RELEASE_ENV_KEYS) materialized = load_env(env_file) missing = [key for key in REQUIRED_RUNTIME_KEYS if not materialized.get(key)] if missing: raise PrepareError( "Prod D3 runtime authority is incomplete: " + ", ".join(missing) ) _require_prod_tidb_binding(materialized) live_frontend = release_root / "frontend-dist" if not live_frontend.exists() and not live_frontend.is_symlink(): live_frontend.symlink_to(source_frontend) return { "active_backend_port": str(route.backend_port), "active_slot": route.slot, # The live route may already be the target on an exact-Tag retry. # Predecessor identity remains bound to durable release authority. "previous_backend_port": str(pending["previousBackendPort"]), "target_slot": target_slot, "target_backend_port": str(target_port), "release_root": str(release_root), "runtime_env_file": str(env_file), "in_progress_file": str(in_progress_file), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--release-tag", required=True) parser.add_argument("--tag-target-commit", required=True) parser.add_argument("--multi-prod-root", default=str(DEFAULT_MULTI_PROD_ROOT)) parser.add_argument("--nginx-conf", default=str(PROD_NGINX_CONF)) parser.add_argument("--github-output", default="") return parser.parse_args() def main() -> int: args = parse_args() try: result = prepare( release_tag=args.release_tag.strip(), tag_target_commit=args.tag_target_commit.strip(), multi_prod_root=Path(args.multi_prod_root).expanduser().resolve(), nginx_conf=Path(args.nginx_conf).expanduser().resolve(), ) if args.github_output: with Path(args.github_output).open("a", encoding="utf-8") as stream: for key, value in result.items(): stream.write(f"{key}={value}\n") except PrepareError as exc: print(f"Prod D3 runtime preparation failed: {exc}") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())