#!/usr/bin/env python3 """Prepare the canonical Test1 L0 serving state without operating its process.""" from __future__ import annotations import argparse import base64 import json import os import re import secrets import subprocess import sys from pathlib import Path from typing import Any, Mapping ROOT = Path(__file__).resolve().parents[1] RUNTIME_TOOLS = ROOT / "deployments" / "offline" / "runtime" sys.path.insert(0, str(RUNTIME_TOOLS)) from common import load_env_file # noqa: E402 DATABASE_KEYS = ( "TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE", "TIDB_TLS_DISABLE", ) STORAGE_KEYS = ( "FILE_STORAGE_PROVIDER", "FILE_STORAGE_BUCKET", "FILE_STORAGE_REGION", "FILE_STORAGE_ENDPOINT", "FILE_STORAGE_ACCESS_KEY_ID", "FILE_STORAGE_SECRET_ACCESS_KEY", "FILE_STORAGE_SESSION_TOKEN", "FILE_STORAGE_PREFIX", "FILE_STORAGE_USE_PATH_STYLE", "FILE_STORAGE_ADDITIONAL_S3_AUTHORITIES_JSON", ) class StatePreparationError(RuntimeError): pass def canonical_bytes(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() def load_object(path: Path, label: str) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise StatePreparationError(f"{label} is not a regular file") try: value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise StatePreparationError(f"{label} is not valid JSON") from exc if not isinstance(value, dict): raise StatePreparationError(f"{label} must be a JSON object") return value def write_private(path: Path, raw: bytes, mode: int = 0o400) -> None: path.parent.mkdir(parents=True, mode=0o700, exist_ok=True) pending = path.with_name(f".{path.name}.pending") if pending.is_symlink() or (pending.exists() and not pending.is_file()): raise StatePreparationError(f"invalid L0 state staging path exists: {pending}") if pending.exists(): pending.unlink() pending.write_bytes(raw) os.chmod(pending, mode) os.replace(pending, path) def require_regular_tree(root: Path, label: str) -> None: if root.is_symlink() or not root.is_dir(): raise StatePreparationError(f"{label} is not a regular directory") for path in root.rglob("*"): if path.is_symlink(): raise StatePreparationError(f"{label} contains a symbolic link") def runtime_environment(source: Mapping[str, str]) -> dict[str, str]: required = (*DATABASE_KEYS[:5], "JWT_SECRET", "PUBLIC_BASE_URL") missing = [key for key in required if not str(source.get(key, ""))] if missing: raise StatePreparationError("Test1 Runtime environment lacks " + ", ".join(missing)) result = {key: str(source[key]) for key in DATABASE_KEYS if key in source} result.update({ "NEXUSCRM_DB_MAX_OPEN_CONNS": "100", "NEXUSCRM_DB_MAX_IDLE_CONNS": "100", "NEXUSCRM_RUNTIME_API_ADMISSION_LIMIT": "16", "NEXUSCRM_RUNTIME_API_PROFILING_ENABLED": "false", "NEXUSCRM_RUNTIME_API_SQL_DIAGNOSTICS_ENABLED": "false", "NEXUSCRM_RUNTIME_FLOW_PERF_TIMING_ENABLED": "true", "NEXUSCRM_PUBLIC_OAUTH_PKCE_ENABLED": "true", }) result.update({ key: str(source[key]) for key in STORAGE_KEYS if source.get(key) not in (None, "") }) result.setdefault("FILE_STORAGE_PROVIDER", "filesystem") # These are host paths in runtime.env. L0 owns their translation to its # two writable container mounts. result["FILE_STORAGE_ROOT"] = "/var/lib/nexuscrm/files" result["FILE_STORAGE_LEGACY_UPLOAD_ROOT"] = "/var/lib/nexuscrm/uploads" for key in ("JWT_SECRET", "ENVIRONMENT", "NEXUS_ENV"): if source.get(key): result[key] = str(source[key]) result["API_BASE_URL"] = str(source["PUBLIC_BASE_URL"]).rstrip("/") if any("\n" in value or "\r" in value or "\x00" in value for value in result.values()): raise StatePreparationError("Test1 Runtime environment contains an invalid value") return result def render_env(values: Mapping[str, str]) -> bytes: for key in values: if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key): raise StatePreparationError(f"invalid Runtime environment key: {key}") return ("\n".join(f"{key}={value}" for key, value in values.items()) + "\n").encode() def workload_identities(path: Path, workload_ids: list[str]) -> dict[str, Any]: normalized = sorted(set(workload_ids)) if not normalized or normalized != workload_ids or any(not value for value in normalized): raise StatePreparationError("serving config has an invalid workload identity set") tokens: dict[str, str] = {} if path.exists(): document = load_object(path, "existing L0 workload identities") rows = document.get("identities") if document.get("schema") != "nexuscrm.runtime-workload-identities/v1" or not isinstance(rows, list): raise StatePreparationError("existing L0 workload identities have an unknown shape") for row in rows: if not isinstance(row, dict) or set(row) != {"workloadId", "token"}: raise StatePreparationError("existing L0 workload identity is malformed") workload_id, token = str(row["workloadId"]), str(row["token"]) if not workload_id or not token or workload_id in tokens: raise StatePreparationError("existing L0 workload identities are ambiguous") tokens[workload_id] = token return { "schema": "nexuscrm.runtime-workload-identities/v1", "identities": [ {"workloadId": value, "token": tokens.get(value) or secrets.token_hex(32)} for value in normalized ], } def signing_keys(path: Path) -> dict[str, Any]: if path.exists(): value = load_object(path, "existing L0 signing keys") if value.get("schema") != "nexuscrm.runtime-signing-keys/v1": raise StatePreparationError("existing L0 signing keys have an unknown shape") return value private_pem = subprocess.run( ("openssl", "genpkey", "-algorithm", "ED25519"), check=True, capture_output=True, ).stdout private_der = subprocess.run( ("openssl", "pkey", "-outform", "DER"), input=private_pem, check=True, capture_output=True, ).stdout public_der = subprocess.run( ("openssl", "pkey", "-pubout", "-outform", "DER"), input=private_pem, check=True, capture_output=True, ).stdout if len(private_der) < 32 or len(public_der) < 32: raise StatePreparationError("OpenSSL emitted an incomplete L0 signing key") return { "schema": "nexuscrm.runtime-signing-keys/v1", "active": { "signing_key_id": "test1-runtime", "private_key_base64": base64.b64encode( private_der[-32:] + public_der[-32:], ).decode("ascii"), }, "overlap": [], } def prepare(serving_path: Path, state_root: Path, env_file: Path) -> None: if state_root.is_symlink(): raise StatePreparationError("L0 state root must not be a symbolic link") state_root.mkdir(parents=True, mode=0o700, exist_ok=True) serving = load_object(serving_path, "Runtime serving config") if serving.get("schema") != "nexuscrm.runtime-api-serving/v4": raise StatePreparationError("Runtime serving config has an unknown schema") workload_ids = serving.get("authorized_workload_ids") if not isinstance(workload_ids, list) or not all(isinstance(value, str) for value in workload_ids): raise StatePreparationError("Runtime serving config lacks authorized workload IDs") catalog_store = state_root / "data/catalog" require_regular_tree(catalog_store, "Catalog store") if not any(path.is_file() for path in catalog_store.rglob("*")): raise StatePreparationError("Catalog store contains no snapshot") secrets_root = state_root / "secrets" identities_path = secrets_root / "workload-identities.json" signing_path = secrets_root / "signing-keys.json" identities = workload_identities(identities_path, workload_ids) signing = signing_keys(signing_path) environment = runtime_environment(load_env_file(env_file)) write_private(state_root / "config/serving.json", serving_path.read_bytes()) write_private(state_root / "config/runtime.env", render_env(environment), mode=0o600) write_private(identities_path, canonical_bytes(identities)) write_private(signing_path, canonical_bytes(signing)) for name in ("files", "uploads"): path = state_root / f"data/{name}" if path.is_symlink() or (path.exists() and not path.is_dir()): raise StatePreparationError(f"existing L0 {name} store is invalid") path.mkdir(parents=True, mode=0o700, exist_ok=True) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--serving-config", type=Path, required=True) parser.add_argument("--state-root", type=Path, required=True) parser.add_argument("--env-file", type=Path, required=True) args = parser.parse_args() try: # Keep the final path component unresolved so prepare() can reject a # symlink instead of silently following it. prepare( args.serving_config.absolute(), args.state_root.absolute(), args.env_file.absolute(), ) return 0 except (StatePreparationError, OSError, subprocess.CalledProcessError) as exc: print(f"prepare-test1-l0-state: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())