#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import os import re import stat import subprocess import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Protocol OPERATION = "hubspot_authored_identity_recovery" ISSUE_URL = "https://github.com/pingcap-inc/NexusCRM/issues/3765" COMMAND = "recover-connector-remote-identity-batch" BACKUP_STATE = "complete_encrypted_immutable" DEFAULT_TIMEOUT_SECONDS = 300 MAX_AUTHORITY_FILE_BYTES = 1024 * 1024 RECOVERY_RELEASE_PACK_ROOT_ENV = "NEXUSCRM_RECOVERY_RELEASE_PACK_ROOT" OFFICIAL_APP_PACKS_ROOT_ENV = "NEXUSCRM_OFFICIAL_APP_PACKS_ROOT" APPROVED_CONNECTOR_OWNER_KEY = "pingcap-hubspot-companion" APPROVED_CONNECTOR_NAMESPACE = "hubspot-crm-connector" APPROVED_DESTINATION_SLOT_ID = "hubspot-contact" APPROVED_COHORT_REF = "prod-legacy-unresolved-20260719T150500Z" APPROVED_COHORT_OBSERVED = 68 APPROVED_COHORT_ELIGIBLE = 31 APPROVED_COHORT_FINALIZED = 23 APPROVED_COHORT_EXCLUDED_NO_EVIDENCE = 14 APPROVED_COHORT_CANDIDATE_MEMBERSHIP_SHA256 = ( "46c5544d29cf6ec7222f39b10f358f6fb2409c5d278caa777524a202cb62457f" ) RECOVERY_GIT_BINARY = "/usr/bin/git" RECOVERY_VERIFICATION_PATH = ( "/usr/local/go/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin" ) SHA256_RE = re.compile(r"^[0-9a-f]{64}$") COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") OPAQUE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,255}$") OPAQUE_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/#-]{0,255}$") OBJECT_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,79}$") BLOCK_REASON_RE = re.compile(r"^[A-Z][A-Z0-9_]{2,127}$") BLOCK_REASONS = { "MANIFEST_INVALID", "DRY_RUN_RECEIPT_INVALID", "RUNTIME_UNAVAILABLE", "RECOVERY_ROLLED_BACK", "COMMITTED_READBACK_BLOCKED", "RUNTIME_RESULT_INVALID", } EMAIL_RE = re.compile(r"\b[^\s@]+@[^\s@]+\.[^\s@]+\b") FORBIDDEN_KEYS = { "address", "clientsecret", "company", "companyname", "credential", "email", "emailaddress", "evidenceid", "firstname", "fullname", "hubspotidc", "lastname", "mirror", "mirrorvalue", "mobile", "mobilephone", "password", "phone", "phonenumber", "postalcode", "secret", "street", "token", } class RunBlocked(RuntimeError): pass @dataclass(frozen=True) class ValidatedPlan: manifest_path: Path manifest: dict[str, Any] manifest_sha256: str deployed_tuple_sha256: str runtime_binding_sha256: str candidate_set_sha256: str @dataclass(frozen=True) class ValidatedDryRunReceipt: path: Path sha256: str receipt: dict[str, Any] @dataclass(frozen=True) class ValidatedBackupMarker: path: Path sha256: str marker: dict[str, Any] @dataclass class BoundFile: path: Path real_path: Path fd: int sha256: str raw: bytes | None def inherited_path(self) -> str: proc_path = Path(f"/proc/self/fd/{self.fd}") if proc_path.exists(): return str(proc_path) dev_path = Path(f"/dev/fd/{self.fd}") if dev_path.exists(): return str(dev_path) raise RunBlocked("host does not expose an inheritable file-descriptor path") def close(self) -> None: if self.fd >= 0: os.close(self.fd) self.fd = -1 @dataclass class ValidatedRuntimeBinding: release_json: BoundFile runtime_env: BoundFile setup_binary: BoundFile version_root: Path recovery_release_pack_root: Path official_app_packs_root: Path downstream_commit: str official_app_packs_commit: str deployed_tuple_sha256: str runtime_binding_sha256: str def pass_fds(self) -> tuple[int, int, int]: return ( self.release_json.fd, self.runtime_env.fd, self.setup_binary.fd, ) def close(self) -> None: self.release_json.close() self.runtime_env.close() self.setup_binary.close() @dataclass class ReservedArtifact: path: Path fd: int def overwrite_bytes(self, payload: bytes) -> None: try: os.lseek(self.fd, 0, os.SEEK_SET) os.ftruncate(self.fd, 0) view = memoryview(payload) while view: written = os.write(self.fd, view) if written <= 0: raise OSError("short artifact write") view = view[written:] os.fsync(self.fd) except OSError as exc: raise RunBlocked("unable to durably write reserved artifact") from exc def overwrite_json(self, value: dict[str, Any]) -> None: _reject_forbidden_data(value, "receipt") payload = ( json.dumps(value, indent=2, sort_keys=True, ensure_ascii=True) + "\n" ).encode("utf-8") self.overwrite_bytes(payload) def sync(self) -> None: try: os.fsync(self.fd) except OSError as exc: raise RunBlocked("unable to sync reserved artifact") from exc def close(self) -> None: if self.fd >= 0: os.close(self.fd) self.fd = -1 @dataclass(frozen=True) class RunnerResult: receipt: dict[str, Any] returncode: int @dataclass(frozen=True) class ExecutionResult: receipt: dict[str, Any] blocked: bool class RuntimeRunner(Protocol): def run( self, plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, dry_run_receipt: ValidatedDryRunReceipt | None, backup_marker: ValidatedBackupMarker | None, stderr_fd: int, ) -> RunnerResult: ... def canonical_sha256(value: Any) -> str: payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _digest(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def _candidate_membership_sha256(candidates: list[dict[str, Any]]) -> str: ordered = sorted( candidates, key=lambda item: ( item["remote_id"], item["expected_source_object"], item["expected_source_record_id"], ), ) payload = "\n".join( "\0".join( ( item["remote_id"], item["expected_source_object"], item["expected_source_record_id"], ) ) for item in ordered ).encode("utf-8") return _digest(payload) def _read_bytes(path: Path, context: str) -> bytes: try: return path.read_bytes() except OSError as exc: raise RunBlocked(f"unable to read {context}") from exc def file_sha256(path: Path) -> str: return _digest(_read_bytes(path, "input file")) def _fd_sha256(fd: int, *, capture_raw: bool, maximum_bytes: int | None) -> tuple[str, bytes | None]: digest = hashlib.sha256() captured = bytearray() if capture_raw else None total = 0 try: os.lseek(fd, 0, os.SEEK_SET) while True: chunk = os.read(fd, 1024 * 1024) if not chunk: break total += len(chunk) if maximum_bytes is not None and total > maximum_bytes: raise RunBlocked("runtime authority file exceeds the reviewed size limit") digest.update(chunk) if captured is not None: captured.extend(chunk) os.lseek(fd, 0, os.SEEK_SET) except OSError as exc: raise RunBlocked("unable to read an opened runtime authority file") from exc return digest.hexdigest(), bytes(captured) if captured is not None else None def _opened_real_path(fd: int, fallback: Path) -> Path: proc_path = Path(f"/proc/self/fd/{fd}") if proc_path.exists(): try: target = os.readlink(proc_path) except OSError as exc: raise RunBlocked("unable to resolve opened runtime authority file") from exc if target.endswith(" (deleted)"): raise RunBlocked("opened runtime authority file was deleted") return Path(target).resolve(strict=True) return fallback.resolve(strict=True) def _open_bound_file( path: Path, expected_sha256: str, context: str, *, capture_raw: bool, executable: bool = False, allow_final_symlink: bool = False, ) -> BoundFile: if not path.is_absolute(): raise RunBlocked(f"{context} path must be absolute") raw_path = path.expanduser() try: if raw_path.is_symlink() and not allow_final_symlink: raise RunBlocked(f"{context} must not be a symlink") resolved = raw_path.resolve(strict=True) except OSError as exc: raise RunBlocked(f"unable to resolve {context}") from exc flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW try: fd = os.open(resolved, flags) except OSError as exc: raise RunBlocked(f"unable to open {context}") from exc try: opened_stat = os.fstat(fd) if not stat.S_ISREG(opened_stat.st_mode): raise RunBlocked(f"{context} must be a regular file") if executable and opened_stat.st_mode & 0o111 == 0: raise RunBlocked(f"{context} must have an executable mode bit") actual, raw = _fd_sha256( fd, capture_raw=capture_raw, maximum_bytes=MAX_AUTHORITY_FILE_BYTES if capture_raw else None, ) if actual != _sha(expected_sha256, f"expected {context} SHA-256"): raise RunBlocked(f"{context} SHA-256 mismatch") real_path = _opened_real_path(fd, resolved) real_stat = real_path.stat() if (opened_stat.st_dev, opened_stat.st_ino) != (real_stat.st_dev, real_stat.st_ino): raise RunBlocked(f"{context} path changed while it was opened") return BoundFile( path=raw_path, real_path=real_path, fd=fd, sha256=actual, raw=raw, ) except Exception: os.close(fd) raise def _reject_duplicate_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: value: dict[str, Any] = {} for key, item in pairs: if key in value: raise RunBlocked("JSON input contains duplicate fields") value[key] = item return value def _parse_json(raw: bytes, context: str) -> dict[str, Any]: try: value = json.loads(raw.decode("utf-8"), object_pairs_hook=_reject_duplicate_pairs) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RunBlocked(f"unable to parse {context} as JSON") from exc return _object(value, context) def _object(value: Any, context: str) -> dict[str, Any]: if not isinstance(value, dict): raise RunBlocked(f"{context} must be a JSON object") return value def _exact(value: dict[str, Any], keys: set[str], context: str) -> None: if set(value) != keys: raise RunBlocked(f"{context} fields do not match the reviewed schema") def _sha(value: Any, context: str) -> str: if not isinstance(value, str) or not SHA256_RE.fullmatch(value): raise RunBlocked(f"{context} must be a lowercase SHA-256") return value def _opaque_id(value: Any, context: str) -> str: if not isinstance(value, str) or not OPAQUE_ID_RE.fullmatch(value): raise RunBlocked(f"{context} must be a non-empty opaque identifier") return value def _opaque_ref(value: Any, context: str) -> str: if not isinstance(value, str) or not OPAQUE_REF_RE.fullmatch(value): raise RunBlocked(f"{context} must be a non-empty opaque reference") return value def _count(value: Any, context: str, *, maximum: int | None = None) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise RunBlocked(f"{context} must be a non-negative integer") if maximum is not None and value > maximum: raise RunBlocked(f"{context} exceeds the reviewed candidate count") return value def _version_one(value: Any, context: str) -> None: if isinstance(value, bool) or not isinstance(value, int) or value != 1: raise RunBlocked(f"{context} must equal integer 1") def _boolean(value: Any, context: str) -> bool: if not isinstance(value, bool): raise RunBlocked(f"{context} must be a JSON boolean") return value def _optional_count(value: Any, context: str, *, maximum: int) -> int | None: if value is None: return None return _count(value, context, maximum=maximum) def _reject_forbidden_data(value: Any, path: str) -> None: if isinstance(value, dict): for key, item in value.items(): normalized = re.sub(r"[^a-z0-9]", "", str(key).lower()) if normalized in FORBIDDEN_KEYS: raise RunBlocked(f"forbidden sensitive key at {path}.{key}") _reject_forbidden_data(item, f"{path}.{key}") elif isinstance(value, list): for index, item in enumerate(value): _reject_forbidden_data(item, f"{path}[{index}]") elif isinstance(value, str) and EMAIL_RE.search(value): raise RunBlocked(f"forbidden email-like value at {path}") def _validate_tuple(value: Any) -> dict[str, Any]: deployed = _object(value, "deployed tuple") _exact( deployed, { "deployment_id", "release_tag", "downstream_commit", "upstream_commit", "official_app_packs_commit", "final_pack_hash", }, "deployed tuple", ) _opaque_ref(deployed["deployment_id"], "deployed tuple deployment_id") _opaque_ref(deployed["release_tag"], "deployed tuple release_tag") for key in ("downstream_commit", "upstream_commit", "official_app_packs_commit"): if not isinstance(deployed[key], str) or not COMMIT_RE.fullmatch(deployed[key]): raise RunBlocked(f"deployed tuple {key} must be an immutable commit") _sha(deployed["final_pack_hash"], "deployed tuple final_pack_hash") return deployed def validate_manifest(value: Any) -> dict[str, Any]: manifest = _object(value, "manifest") _reject_forbidden_data(manifest, "manifest") _exact( manifest, { "schema_version", "operation", "issue", "environment", "approval_ref", "deployed_tuple", "runtime_binding", "selector", "census", "candidates", }, "manifest", ) _version_one(manifest["schema_version"], "manifest schema_version") if manifest["operation"] != OPERATION: raise RunBlocked("manifest operation mismatch") if manifest["issue"] != ISSUE_URL: raise RunBlocked("manifest issue binding must equal NexusCRM #3765") if manifest["environment"] != "prod": raise RunBlocked("manifest environment must equal prod") _opaque_ref(manifest["approval_ref"], "manifest approval_ref") _validate_tuple(manifest["deployed_tuple"]) runtime_binding = _object(manifest["runtime_binding"], "manifest runtime_binding") _exact( runtime_binding, {"release_json_sha256", "runtime_env_sha256", "setup_binary_sha256"}, "manifest runtime_binding", ) for key in ("release_json_sha256", "runtime_env_sha256", "setup_binary_sha256"): _sha(runtime_binding[key], f"manifest runtime_binding {key}") selector = _object(manifest["selector"], "manifest selector") _exact( selector, {"connector_owner_key", "connector_namespace", "destination_slot_id"}, "manifest selector", ) for key in ("connector_owner_key", "connector_namespace", "destination_slot_id"): _opaque_id(selector[key], f"manifest selector {key}") approved_selector = { "connector_owner_key": APPROVED_CONNECTOR_OWNER_KEY, "connector_namespace": APPROVED_CONNECTOR_NAMESPACE, "destination_slot_id": APPROVED_DESTINATION_SLOT_ID, } if selector != approved_selector: raise RunBlocked( "manifest selector must equal the approved HubSpot recovery selector" ) census = _object(manifest["census"], "manifest census") _exact( census, { "cohort_ref", "observed", "eligible", "already_finalized", "excluded_no_evidence", }, "manifest census", ) if census["cohort_ref"] != APPROVED_COHORT_REF: raise RunBlocked("manifest census cohort_ref does not name the approved cohort") observed = _count( census["observed"], "manifest census observed", ) eligible = _count( census["eligible"], "manifest census eligible", maximum=APPROVED_COHORT_ELIGIBLE, ) already_finalized = _count( census["already_finalized"], "manifest census already_finalized", ) excluded_no_evidence = _count( census["excluded_no_evidence"], "manifest census excluded_no_evidence", ) if observed != APPROVED_COHORT_OBSERVED: raise RunBlocked("manifest census must retain the approved 68-contact cohort") if eligible != APPROVED_COHORT_ELIGIBLE: raise RunBlocked("manifest census must retain exactly 31 reviewed candidates") if already_finalized != APPROVED_COHORT_FINALIZED: raise RunBlocked("manifest census must retain exactly 23 reviewed finalized rows") if excluded_no_evidence != APPROVED_COHORT_EXCLUDED_NO_EVIDENCE: raise RunBlocked("manifest census must retain the approved 14 no-evidence exclusions") if eligible + already_finalized + excluded_no_evidence != observed: raise RunBlocked("manifest census partition must equal the approved 68-contact cohort") candidates = manifest["candidates"] if not isinstance(candidates, list): raise RunBlocked("manifest candidates must be a JSON array") if len(candidates) != APPROVED_COHORT_ELIGIBLE: raise RunBlocked("manifest must contain exactly 31 reviewed candidates") if len(candidates) != eligible: raise RunBlocked("manifest census eligible must equal candidate count") remote_ids: set[str] = set() source_record_ids: set[str] = set() for index, raw in enumerate(candidates): candidate = _object(raw, f"manifest candidate {index}") _exact( candidate, {"remote_id", "expected_source_object", "expected_source_record_id"}, f"manifest candidate {index}", ) remote_id = _opaque_id(candidate["remote_id"], f"manifest candidate {index} remote_id") source_record_id = _opaque_id( candidate["expected_source_record_id"], f"manifest candidate {index} expected_source_record_id", ) source_object = candidate["expected_source_object"] if not isinstance(source_object, str) or not OBJECT_RE.fullmatch(source_object): raise RunBlocked(f"manifest candidate {index} expected_source_object is invalid") if source_object != "Lead": raise RunBlocked("the approved #3765 production candidate object must equal Lead") if remote_id in remote_ids: raise RunBlocked("duplicate remote_id in manifest candidates") if source_record_id in source_record_ids: raise RunBlocked("duplicate expected_source_record_id in manifest candidates") remote_ids.add(remote_id) source_record_ids.add(source_record_id) if ( _candidate_membership_sha256(candidates) != APPROVED_COHORT_CANDIDATE_MEMBERSHIP_SHA256 ): raise RunBlocked( "manifest candidates do not equal the reviewed candidate membership snapshot" ) return manifest def _sorted_candidates(manifest: dict[str, Any]) -> list[dict[str, str]]: candidates = [ { "remote_id": item["remote_id"], "expected_source_object": item["expected_source_object"], "expected_source_record_id": item["expected_source_record_id"], } for item in manifest["candidates"] ] return sorted( candidates, key=lambda item: ( item["remote_id"], item["expected_source_object"], item["expected_source_record_id"], ), ) def load_plan( manifest_path: Path, expected_manifest_sha256: str, ) -> ValidatedPlan: manifest_path = manifest_path.expanduser().resolve() expected = _sha(expected_manifest_sha256, "expected manifest SHA-256") manifest_raw = _read_bytes(manifest_path, "manifest") actual = _digest(manifest_raw) if actual != expected: raise RunBlocked("manifest SHA-256 mismatch") manifest = validate_manifest(_parse_json(manifest_raw, "manifest")) deployed_tuple = manifest["deployed_tuple"] binding_payload = { "deployed_tuple": deployed_tuple, "runtime_binding": manifest["runtime_binding"], } return ValidatedPlan( manifest_path=manifest_path, manifest=manifest, manifest_sha256=actual, deployed_tuple_sha256=canonical_sha256(deployed_tuple), runtime_binding_sha256=canonical_sha256(binding_payload), candidate_set_sha256=canonical_sha256(_sorted_candidates(manifest)), ) def _parse_runtime_env(raw: bytes) -> dict[str, str]: try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: raise RunBlocked("runtime env authority is not UTF-8") from exc values: dict[str, str] = {} for raw_line in text.splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue if line.startswith("export "): line = line[len("export ") :].strip() key, value = line.split("=", 1) if key in values: raise RunBlocked("runtime env authority contains duplicate fields") if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: value = value[1:-1] values[key] = value return values def _require_env_value(values: dict[str, str], key: str) -> str: value = values.get(key, "").strip() if not value: raise RunBlocked(f"runtime env authority is missing {key}") return value def _require_release_value(release: dict[str, Any], key: str) -> str: value = release.get(key) if not isinstance(value, str) or not value.strip(): raise RunBlocked(f"release authority is missing {key}") return value.strip() def _is_relative_to(path: Path, parent: Path) -> bool: try: path.relative_to(parent) return True except ValueError: return False def recovery_git_verification_env() -> dict[str, str]: return { "PATH": RECOVERY_VERIFICATION_PATH, "HOME": "/nonexistent", "XDG_CONFIG_HOME": "/nonexistent", "LANG": "C", "LC_ALL": "C", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_TERMINAL_PROMPT": "0", "GIT_ASKPASS": "/bin/false", "GIT_ALLOW_PROTOCOL": "https", "GIT_OPTIONAL_LOCKS": "0", } def _git_checkout_output(root: Path, args: list[str], context: str) -> str: try: completed = subprocess.run( [RECOVERY_GIT_BINARY, "-C", str(root), *args], check=False, capture_output=True, text=True, env=recovery_git_verification_env(), ) except OSError as exc: raise RunBlocked(f"unable to inspect {context}") from exc if completed.returncode != 0: raise RunBlocked(f"unable to inspect {context}") return completed.stdout.strip() def _git_checkout_paths(root: Path, args: list[str], context: str) -> set[str]: try: completed = subprocess.run( [RECOVERY_GIT_BINARY, "-C", str(root), *args, "-z"], check=False, capture_output=True, env=recovery_git_verification_env(), ) except OSError as exc: raise RunBlocked(f"unable to inspect {context}") from exc if completed.returncode != 0: raise RunBlocked(f"unable to inspect {context}") try: output = completed.stdout.decode("utf-8") except UnicodeDecodeError as exc: raise RunBlocked(f"unable to inspect {context}") from exc return {path for path in output.split("\0") if path} def _require_git_checkout_content( root: Path, expected_commit: str, context: str ) -> None: index_entries = _git_checkout_paths(root, ["ls-files", "-v"], context) if any(not entry.startswith("H ") for entry in index_entries): raise RunBlocked(f"{context} contains unsupported Git index flags") try: completed = subprocess.run( [ RECOVERY_GIT_BINARY, "-C", str(root), "diff", "--no-ext-diff", "--quiet", expected_commit, "--", ], check=False, capture_output=True, env=recovery_git_verification_env(), ) except OSError as exc: raise RunBlocked(f"unable to inspect {context}") from exc if completed.returncode == 1: raise RunBlocked(f"{context} is not a clean immutable checkout") if completed.returncode != 0: raise RunBlocked(f"unable to inspect {context}") def _validate_release_checkout( root: Path, *, version_root: Path, expected_commit: str, context: str, ) -> Path: if not root.is_absolute(): raise RunBlocked(f"{context} must be absolute") try: root = root.resolve(strict=True) except OSError as exc: raise RunBlocked(f"{context} does not exist") from exc if not _is_relative_to(root, version_root) or root == version_root: raise RunBlocked(f"{context} escapes the active version_root") git_root = Path( _git_checkout_output(root, ["rev-parse", "--show-toplevel"], context) ).resolve(strict=True) if git_root != root: raise RunBlocked(f"{context} is not the exact Git worktree top level") head = _git_checkout_output(root, ["rev-parse", "HEAD^{commit}"], context) if head != expected_commit: raise RunBlocked(f"{context} HEAD disagrees with the deployed commit") _require_git_checkout_content(root, expected_commit, context) status = _git_checkout_output( root, ["status", "--porcelain=v1", "--untracked-files=all"], context, ) if status: raise RunBlocked(f"{context} is not a clean immutable checkout") ignored_paths = _git_checkout_paths( root, ["ls-files", "--others", "--ignored", "--exclude-standard", "--directory"], context, ) if ignored_paths: raise RunBlocked(f"{context} contains ignored paths") return root def _validated_bound_checkout( release: dict[str, Any], runtime_env: dict[str, str], *, release_key: str, env_key: str, version_root: Path, expected_commit: str, context: str, ) -> Path: release_path = Path(_require_release_value(release, release_key)).expanduser() env_path = Path(_require_env_value(runtime_env, env_key)).expanduser() if not release_path.is_absolute() or not env_path.is_absolute(): raise RunBlocked(f"{context} authorities must be absolute") try: resolved_release_path = release_path.resolve(strict=True) resolved_env_path = env_path.resolve(strict=True) except OSError as exc: raise RunBlocked(f"{context} authority does not exist") from exc if resolved_release_path != resolved_env_path: raise RunBlocked(f"runtime env {env_key} disagrees with release.json") return _validate_release_checkout( resolved_release_path, version_root=version_root, expected_commit=expected_commit, context=context, ) def load_runtime_binding( plan: ValidatedPlan, *, release_json_path: Path, runtime_env_path: Path, setup_binary_path: Path, ) -> ValidatedRuntimeBinding: expected = plan.manifest["runtime_binding"] release_file: BoundFile | None = None runtime_file: BoundFile | None = None setup_file: BoundFile | None = None try: release_file = _open_bound_file( release_json_path, expected["release_json_sha256"], "release.json authority", capture_raw=True, ) runtime_file = _open_bound_file( runtime_env_path, expected["runtime_env_sha256"], "active runtime.env authority", capture_raw=True, allow_final_symlink=True, ) setup_file = _open_bound_file( setup_binary_path, expected["setup_binary_sha256"], "setup mutation binary", capture_raw=False, executable=True, ) assert release_file.raw is not None assert runtime_file.raw is not None release = _parse_json(release_file.raw, "release.json authority") runtime_env = _parse_runtime_env(runtime_file.raw) version_root = Path(_require_release_value(release, "version_root")).expanduser() if not version_root.is_absolute(): raise RunBlocked("release authority version_root must be absolute") version_root = version_root.resolve(strict=True) canonical_release_path = version_root / "release.json" try: if canonical_release_path.is_symlink(): raise RunBlocked( "canonical version_root/release.json must not be a symlink" ) canonical_release_path = canonical_release_path.resolve(strict=True) canonical_release_stat = canonical_release_path.stat() except RunBlocked: raise except OSError as exc: raise RunBlocked( "canonical release.json authority does not exist in the exact version_root" ) from exc release_file_stat = os.fstat(release_file.fd) if release_file.real_path != canonical_release_path or ( release_file_stat.st_dev, release_file_stat.st_ino, ) != ( canonical_release_stat.st_dev, canonical_release_stat.st_ino, ): raise RunBlocked( "release.json authority is not the canonical version_root/release.json" ) expected_runtime_env = version_root / "env" / "runtime.env" if runtime_file.real_path != expected_runtime_env: raise RunBlocked("runtime.env authority is not inside the exact version_root") current_runtime_env = version_root.parent.parent / "current" / "env" / "runtime.env" if not current_runtime_env.is_symlink(): raise RunBlocked("active current/runtime.env authority is not a symlink") if current_runtime_env.resolve(strict=True) != runtime_file.real_path: raise RunBlocked("runtime.env authority is not the active current release target") if not _is_relative_to(setup_file.real_path, version_root): raise RunBlocked("setup mutation binary escapes the active version_root") env_root = Path(_require_env_value(runtime_env, "ENV_ROOT")).expanduser() if not env_root.is_absolute(): raise RunBlocked("runtime env ENV_ROOT must be absolute") if env_root.resolve(strict=True) != version_root: raise RunBlocked("runtime env ENV_ROOT disagrees with release version_root") configured_setup = Path( _require_env_value(runtime_env, "NEXUSCRM_SETUP_BIN") ).expanduser() if not configured_setup.is_absolute(): raise RunBlocked("runtime env NEXUSCRM_SETUP_BIN must be absolute") if configured_setup.resolve(strict=True) != setup_file.real_path: raise RunBlocked( "setup mutation binary disagrees with runtime env NEXUSCRM_SETUP_BIN" ) downstream = _require_release_value(release, "resolved_downstream_ref") upstream = _require_release_value(release, "resolved_upstream_ref") official = _require_release_value(release, "resolved_official_ref") tag_target = _require_release_value(release, "tag_target_commit") release_tag = _require_release_value(release, "release_tag") final_pack_hash = _require_release_value(release, "final_pack_hash") for label, value in ( ("resolved_downstream_ref", downstream), ("resolved_upstream_ref", upstream), ("resolved_official_ref", official), ("tag_target_commit", tag_target), ): if not COMMIT_RE.fullmatch(value): raise RunBlocked(f"release authority {label} is not an immutable commit") _opaque_ref(release_tag, "release authority release_tag") _sha(final_pack_hash, "release authority final_pack_hash") if tag_target != downstream: raise RunBlocked("release tag target does not match the resolved downstream commit") observed_tuple = { "deployment_id": _require_env_value(runtime_env, "DEPLOY_ID"), "release_tag": release_tag, "downstream_commit": downstream, "upstream_commit": upstream, "official_app_packs_commit": official, "final_pack_hash": final_pack_hash, } _validate_tuple(observed_tuple) if observed_tuple != plan.manifest["deployed_tuple"]: raise RunBlocked("live release/runtime authorities do not match the manifest deployed tuple") env_ref_pairs = { "DOWNSTREAM_REF": downstream, "UPSTREAM_REF": upstream, "OFFICIAL_APP_PACKS_REF": official, } for key, value in env_ref_pairs.items(): if _require_env_value(runtime_env, key) != value: raise RunBlocked(f"runtime env {key} disagrees with release.json") recovery_release_pack_root = _validated_bound_checkout( release, runtime_env, release_key="recovery_release_pack_root", env_key=RECOVERY_RELEASE_PACK_ROOT_ENV, version_root=version_root, expected_commit=downstream, context="recovery release pack root", ) official_app_packs_root = _validated_bound_checkout( release, runtime_env, release_key="recovery_official_app_packs_root", env_key=OFFICIAL_APP_PACKS_ROOT_ENV, version_root=version_root, expected_commit=official, context="official app-packs root", ) for optional_key, value in ( ("RELEASE_TAG", release_tag), ("FINAL_PACK_HASH", final_pack_hash), ): if optional_key in runtime_env and runtime_env[optional_key].strip() != value: raise RunBlocked(f"runtime env {optional_key} disagrees with release.json") binding_payload = { "deployed_tuple": observed_tuple, "runtime_binding": { "release_json_sha256": release_file.sha256, "runtime_env_sha256": runtime_file.sha256, "setup_binary_sha256": setup_file.sha256, }, } runtime_binding_sha256 = canonical_sha256(binding_payload) if runtime_binding_sha256 != plan.runtime_binding_sha256: raise RunBlocked("live runtime binding hash does not match the reviewed manifest") return ValidatedRuntimeBinding( release_json=release_file, runtime_env=runtime_file, setup_binary=setup_file, version_root=version_root, recovery_release_pack_root=recovery_release_pack_root, official_app_packs_root=official_app_packs_root, downstream_commit=downstream, official_app_packs_commit=official, deployed_tuple_sha256=canonical_sha256(observed_tuple), runtime_binding_sha256=runtime_binding_sha256, ) except Exception: for opened in (release_file, runtime_file, setup_file): if opened is not None: opened.close() raise def revalidate_runtime_binding(runtime_binding: ValidatedRuntimeBinding) -> None: for opened, context, capture_raw in ( (runtime_binding.release_json, "release.json authority", True), (runtime_binding.runtime_env, "active runtime.env authority", True), (runtime_binding.setup_binary, "setup mutation binary", False), ): actual, _ = _fd_sha256( opened.fd, capture_raw=capture_raw, maximum_bytes=MAX_AUTHORITY_FILE_BYTES if capture_raw else None, ) if actual != opened.sha256: raise RunBlocked(f"{context} changed after runtime binding validation") if _opened_real_path(opened.fd, opened.real_path) != opened.real_path: raise RunBlocked(f"{context} target changed after runtime binding validation") opened_stat = os.fstat(opened.fd) try: path_stat = opened.real_path.stat() except OSError as exc: raise RunBlocked(f"{context} path disappeared before mutation") from exc if (opened_stat.st_dev, opened_stat.st_ino) != (path_stat.st_dev, path_stat.st_ino): raise RunBlocked(f"{context} path was replaced before mutation") current_runtime_env = ( runtime_binding.version_root.parent.parent / "current" / "env" / "runtime.env" ) if not current_runtime_env.is_symlink(): raise RunBlocked("active current/runtime.env authority disappeared") if current_runtime_env.resolve(strict=True) != runtime_binding.runtime_env.real_path: raise RunBlocked("active current/runtime.env authority changed before mutation") _validate_release_checkout( runtime_binding.recovery_release_pack_root, version_root=runtime_binding.version_root, expected_commit=runtime_binding.downstream_commit, context="recovery release pack root", ) _validate_release_checkout( runtime_binding.official_app_packs_root, version_root=runtime_binding.version_root, expected_commit=runtime_binding.official_app_packs_commit, context="official app-packs root", ) def validate_backup_marker(value: Any, plan: ValidatedPlan) -> dict[str, Any]: marker = _object(value, "backup marker") _reject_forbidden_data(marker, "backup_marker") _exact( marker, { "schema_version", "operation", "environment", "manifest_sha256", "deployed_tuple", "runtime_binding_sha256", "candidate_set_sha256", "backup_reference", "backup_sha256", "state", }, "backup marker", ) _version_one(marker["schema_version"], "backup marker schema_version") if marker["operation"] != OPERATION: raise RunBlocked("backup marker operation mismatch") if marker["environment"] != "prod" or marker["state"] != BACKUP_STATE: raise RunBlocked("backup marker is not a completed immutable production backup") if marker["manifest_sha256"] != plan.manifest_sha256: raise RunBlocked("backup marker does not bind the exact manifest") if _validate_tuple(marker["deployed_tuple"]) != plan.manifest["deployed_tuple"]: raise RunBlocked("backup marker does not bind the exact deployed tuple") if marker["runtime_binding_sha256"] != plan.runtime_binding_sha256: raise RunBlocked("backup marker does not bind the exact mutation runtime") if marker["candidate_set_sha256"] != plan.candidate_set_sha256: raise RunBlocked("backup marker does not bind the exact candidate set") _opaque_ref(marker["backup_reference"], "backup marker backup_reference") _sha(marker["backup_sha256"], "backup marker backup_sha256") return marker def load_backup_marker( path: Path, expected_sha256: str, plan: ValidatedPlan, ) -> ValidatedBackupMarker: path = path.expanduser().resolve() raw = _read_bytes(path, "backup marker") actual = _digest(raw) if actual != _sha(expected_sha256, "expected backup marker SHA-256"): raise RunBlocked("backup marker SHA-256 mismatch") marker = validate_backup_marker(_parse_json(raw, "backup marker"), plan) return ValidatedBackupMarker(path=path, sha256=actual, marker=marker) def _expected_runtime_items(plan: ValidatedPlan) -> list[dict[str, str]]: return sorted( [ { "remote_id": item["remote_id"], "source_object": item["expected_source_object"], "source_record_id": item["expected_source_record_id"], } for item in plan.manifest["candidates"] ], key=lambda item: (item["remote_id"], item["source_object"], item["source_record_id"]), ) def _validate_runtime_items(value: Any, plan: ValidatedPlan) -> list[dict[str, str]]: if not isinstance(value, list): raise RunBlocked("runtime receipt items must be a JSON array") items: list[dict[str, str]] = [] for index, raw in enumerate(value): item = _object(raw, f"runtime receipt item {index}") _exact(item, {"remote_id", "source_object", "source_record_id"}, f"runtime receipt item {index}") remote_id = _opaque_id(item["remote_id"], f"runtime receipt item {index} remote_id") source_object = item["source_object"] if not isinstance(source_object, str) or not OBJECT_RE.fullmatch(source_object): raise RunBlocked(f"runtime receipt item {index} source_object is invalid") source_record_id = _opaque_id( item["source_record_id"], f"runtime receipt item {index} source_record_id" ) items.append( { "remote_id": remote_id, "source_object": source_object, "source_record_id": source_record_id, } ) if len({item["remote_id"] for item in items}) != len(items): raise RunBlocked("runtime receipt contains duplicate remote_id items") return sorted(items, key=lambda item: (item["remote_id"], item["source_object"], item["source_record_id"])) def validate_runtime_receipt( result: RunnerResult, plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, dry_run_receipt_sha256: str | None = None, backup_marker_sha256: str | None = None, ) -> ExecutionResult: receipt = _object(result.receipt, "runtime receipt") _reject_forbidden_data(receipt, "runtime_receipt") if mode not in {"dry_run", "apply"}: raise RunBlocked("unsupported runtime mode") outcome = receipt.get("outcome") blocked = outcome in {"rolled_back_blocked", "committed_readback_blocked"} keys = { "schema_version", "command", "mode", "manifest_sha256", "outcome", "committed", "readback_verified", "enrollment_allowed", "requested_count", "recovered_count", "finalized_link_count", "consumed_evidence_count", "resolver_match_count", "items", "runtime_binding_sha256", "release_json_sha256", "runtime_env_sha256", "setup_binary_sha256", "deployed_tuple_sha256", "result_contract_valid", } if mode == "apply": keys |= {"backup_marker", "dry_run_receipt_sha256"} if blocked: keys.add("block_reason") _exact(receipt, keys, "runtime receipt") _version_one(receipt["schema_version"], "runtime receipt schema_version") if receipt["command"] != COMMAND: raise RunBlocked("runtime receipt command contract mismatch") if receipt["mode"] != mode or receipt["manifest_sha256"] != plan.manifest_sha256: raise RunBlocked("runtime receipt does not bind the requested mode and manifest") expected_binding_fields = { "runtime_binding_sha256": runtime_binding.runtime_binding_sha256, "release_json_sha256": runtime_binding.release_json.sha256, "runtime_env_sha256": runtime_binding.runtime_env.sha256, "setup_binary_sha256": runtime_binding.setup_binary.sha256, "deployed_tuple_sha256": runtime_binding.deployed_tuple_sha256, } for key, expected_value in expected_binding_fields.items(): if receipt[key] != expected_value: raise RunBlocked(f"runtime receipt {key} does not bind the mutation runtime") if mode == "apply": if receipt["dry_run_receipt_sha256"] != dry_run_receipt_sha256: raise RunBlocked("runtime apply receipt does not bind the dry-run receipt") if receipt["backup_marker"] != backup_marker_sha256: raise RunBlocked("runtime apply receipt does not bind the backup marker") if blocked: reason = receipt["block_reason"] if ( not isinstance(reason, str) or not BLOCK_REASON_RE.fullmatch(reason) or reason not in BLOCK_REASONS ): raise RunBlocked("runtime block_reason must be a stable non-sensitive code") result_contract_valid = _boolean( receipt["result_contract_valid"], "runtime result_contract_valid" ) count = len(plan.manifest["candidates"]) if not result_contract_valid: if outcome != "committed_readback_blocked": raise RunBlocked("invalid runtime result contract must block as committed readback") if receipt.get("block_reason") != "RUNTIME_RESULT_INVALID": raise RunBlocked("invalid runtime result contract has the wrong block reason") flags = ( _boolean(receipt["committed"], "runtime committed"), _boolean(receipt["readback_verified"], "runtime readback_verified"), _boolean(receipt["enrollment_allowed"], "runtime enrollment_allowed"), ) if flags != (True, False, False): raise RunBlocked("invalid committed runtime result has unsafe flags") for key in ( "requested_count", "recovered_count", "finalized_link_count", "consumed_evidence_count", "resolver_match_count", ): _optional_count(receipt[key], f"runtime {key}", maximum=count) items = _validate_runtime_items(receipt["items"], plan) if any(item not in _expected_runtime_items(plan) for item in items): raise RunBlocked("invalid committed runtime result contains a foreign source item") if result.returncode == 0: raise RunBlocked("invalid committed runtime result must return a blocked exit status") return ExecutionResult(receipt=receipt, blocked=True) requested_count = _count( receipt["requested_count"], "runtime requested_count", maximum=count ) counts = { key: _count(receipt[key], f"runtime {key}", maximum=count) for key in ( "recovered_count", "finalized_link_count", "consumed_evidence_count", "resolver_match_count", ) } items = _validate_runtime_items(receipt["items"], plan) expected_items = _expected_runtime_items(plan) expected_outcomes = ( {"dry_run_verified", "rolled_back_blocked"} if mode == "dry_run" else {"committed_verified", "committed_readback_blocked", "rolled_back_blocked"} ) if outcome not in expected_outcomes: raise RunBlocked("runtime receipt outcome is not valid for the requested mode") if outcome == "dry_run_verified": expected_flags = (False, False, False) expected_counts = (count, count, count, 0) if items != expected_items: raise RunBlocked("runtime dry-run receipt does not match all expected source owners") elif outcome == "committed_verified": expected_flags = (True, True, True) expected_counts = (count, count, count, count) if items != expected_items: raise RunBlocked("runtime apply receipt does not prove every exact source readback") elif outcome == "committed_readback_blocked": expected_flags = (True, False, False) expected_counts = (count, count, count, 0) if items != expected_items: raise RunBlocked("committed blocked receipt does not bind the full committed candidate set") else: expected_flags = (False, False, False) expected_counts = (0, 0, 0, 0) if items: raise RunBlocked("rolled-back receipt must not claim durable candidate effects") allowed_requested_counts = {0, count} if outcome == "rolled_back_blocked" else {count} if requested_count not in allowed_requested_counts: raise RunBlocked("runtime receipt requested_count is inconsistent with its outcome") flags = ( _boolean(receipt["committed"], "runtime committed"), _boolean(receipt["readback_verified"], "runtime readback_verified"), _boolean(receipt["enrollment_allowed"], "runtime enrollment_allowed"), ) if flags != expected_flags: raise RunBlocked("runtime receipt commit/readback/enrollment flags are inconsistent") actual_counts = ( counts["recovered_count"], counts["finalized_link_count"], counts["consumed_evidence_count"], counts["resolver_match_count"], ) if actual_counts != expected_counts: raise RunBlocked("runtime receipt effect counts are inconsistent with its outcome") if blocked != (result.returncode != 0): raise RunBlocked("runtime adapter exit status disagrees with its receipt outcome") return ExecutionResult(receipt=receipt, blocked=blocked) def load_dry_run_receipt( path: Path, expected_sha256: str, plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, ) -> ValidatedDryRunReceipt: path = path.expanduser().resolve() raw = _read_bytes(path, "dry-run receipt") actual = _digest(raw) if actual != _sha(expected_sha256, "expected dry-run receipt SHA-256"): raise RunBlocked("dry-run receipt SHA-256 mismatch") receipt = _parse_json(raw, "dry-run receipt") validated = validate_runtime_receipt( RunnerResult(receipt=receipt, returncode=0), plan, runtime_binding, mode="dry_run", ) if validated.blocked or receipt["outcome"] != "dry_run_verified": raise RunBlocked("apply requires a successful runtime dry-run receipt") return ValidatedDryRunReceipt(path=path, sha256=actual, receipt=receipt) def _sanitize_runtime_items(value: Any, plan: ValidatedPlan) -> list[dict[str, str]]: if not isinstance(value, list): return [] expected = _expected_runtime_items(plan) sanitized: list[dict[str, str]] = [] seen: set[str] = set() for raw in value: if not isinstance(raw, dict) or set(raw) != { "remote_id", "source_object", "source_record_id", }: continue remote_id = raw.get("remote_id") source_object = raw.get("source_object") source_record_id = raw.get("source_record_id") if ( not isinstance(remote_id, str) or not isinstance(source_object, str) or not isinstance(source_record_id, str) ): continue item = { "remote_id": remote_id, "source_object": source_object, "source_record_id": source_record_id, } if item not in expected or remote_id in seen: continue seen.add(remote_id) sanitized.append(item) return sorted( sanitized, key=lambda item: (item["remote_id"], item["source_object"], item["source_record_id"]), ) def sanitize_committed_invalid_receipt( raw: dict[str, Any], plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, dry_run_receipt_sha256: str | None, backup_marker_sha256: str | None, ) -> dict[str, Any]: count = len(plan.manifest["candidates"]) def safe_count(key: str) -> int | None: value = raw.get(key) if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= count: return None return value receipt: dict[str, Any] = { "schema_version": 1, "command": COMMAND, "mode": mode, "manifest_sha256": plan.manifest_sha256, "outcome": "committed_readback_blocked", "committed": True, "readback_verified": False, "enrollment_allowed": False, "requested_count": safe_count("requested_count"), "recovered_count": safe_count("recovered_count"), "finalized_link_count": safe_count("finalized_link_count"), "consumed_evidence_count": safe_count("consumed_evidence_count"), "resolver_match_count": safe_count("resolver_match_count"), "items": _sanitize_runtime_items(raw.get("items"), plan), "runtime_binding_sha256": runtime_binding.runtime_binding_sha256, "release_json_sha256": runtime_binding.release_json.sha256, "runtime_env_sha256": runtime_binding.runtime_env.sha256, "setup_binary_sha256": runtime_binding.setup_binary.sha256, "deployed_tuple_sha256": runtime_binding.deployed_tuple_sha256, "result_contract_valid": False, "block_reason": "RUNTIME_RESULT_INVALID", } if mode == "apply": receipt["dry_run_receipt_sha256"] = dry_run_receipt_sha256 receipt["backup_marker"] = backup_marker_sha256 return receipt class SubprocessRuntimeRunner: def __init__(self, *, timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS) -> None: self.timeout_seconds = timeout_seconds def build_command( self, plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, dry_run_receipt: ValidatedDryRunReceipt | None, backup_marker: ValidatedBackupMarker | None, ) -> list[str]: command = [ runtime_binding.setup_binary.inherited_path(), COMMAND, "--manifest", str(plan.manifest_path), "--manifest-sha256", plan.manifest_sha256, "--release-json", runtime_binding.release_json.inherited_path(), "--release-json-sha256", runtime_binding.release_json.sha256, "--runtime-env", runtime_binding.runtime_env.inherited_path(), "--runtime-env-sha256", runtime_binding.runtime_env.sha256, "--setup-binary-sha256", runtime_binding.setup_binary.sha256, ] if mode == "apply": if dry_run_receipt is None or backup_marker is None: raise RunBlocked("apply requires validated dry-run and backup inputs") command.extend( [ "--apply", "--dry-run-receipt", str(dry_run_receipt.path), "--dry-run-receipt-sha256", dry_run_receipt.sha256, # L2 validates the marker file and digest; L0 accepts only the # verified digest as its opaque backup authorization proof. "--backup-marker", backup_marker.sha256, ] ) elif mode != "dry_run": raise RunBlocked("unsupported runtime mode") return command def run( self, plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, dry_run_receipt: ValidatedDryRunReceipt | None, backup_marker: ValidatedBackupMarker | None, stderr_fd: int, ) -> RunnerResult: command = self.build_command( plan, runtime_binding, mode=mode, dry_run_receipt=dry_run_receipt, backup_marker=backup_marker, ) try: runtime_env = os.environ.copy() runtime_env[RECOVERY_RELEASE_PACK_ROOT_ENV] = str( runtime_binding.recovery_release_pack_root ) runtime_env[OFFICIAL_APP_PACKS_ROOT_ENV] = str( runtime_binding.official_app_packs_root ) completed = subprocess.run( command, check=False, stdout=subprocess.PIPE, stderr=stderr_fd, text=True, timeout=self.timeout_seconds, pass_fds=runtime_binding.pass_fds(), env=runtime_env, ) except (OSError, subprocess.TimeoutExpired) as exc: raise RunBlocked("unable to execute the protected L0 setup adapter") from exc try: os.fsync(stderr_fd) except OSError as exc: raise RunBlocked("unable to sync protected runtime stderr") from exc try: receipt = _parse_json(completed.stdout.encode("utf-8"), "runtime adapter receipt") except RunBlocked as exc: raise RunBlocked( f"runtime adapter exited {completed.returncode} without a valid JSON receipt" ) from exc return RunnerResult(receipt=receipt, returncode=completed.returncode) def execute( plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, runner: RuntimeRunner, *, mode: str = "dry_run", dry_run_receipt: ValidatedDryRunReceipt | None = None, backup_marker: ValidatedBackupMarker | None = None, stderr_fd: int, ) -> ExecutionResult: if mode == "apply" and (dry_run_receipt is None or backup_marker is None): raise RunBlocked("apply requires validated dry-run receipt and backup marker files") if mode == "dry_run" and (dry_run_receipt is not None or backup_marker is not None): raise RunBlocked("dry-run does not accept apply-only authorization artifacts") revalidate_runtime_binding(runtime_binding) # The protected L0 command revalidates these authorities again, acquires the # shared deployment cutover lock, and holds it through the database batch. result = runner.run( plan, runtime_binding, mode=mode, dry_run_receipt=dry_run_receipt, backup_marker=backup_marker, stderr_fd=stderr_fd, ) dry_sha = dry_run_receipt.sha256 if dry_run_receipt else None backup_sha = backup_marker.sha256 if backup_marker else None try: return validate_runtime_receipt( result, plan, runtime_binding, mode=mode, dry_run_receipt_sha256=dry_sha, backup_marker_sha256=backup_sha, ) except RunBlocked: if result.receipt.get("committed") is not True: raise sanitized = sanitize_committed_invalid_receipt( result.receipt, plan, runtime_binding, mode=mode, dry_run_receipt_sha256=dry_sha, backup_marker_sha256=backup_sha, ) return validate_runtime_receipt( RunnerResult(receipt=sanitized, returncode=2), plan, runtime_binding, mode=mode, dry_run_receipt_sha256=dry_sha, backup_marker_sha256=backup_sha, ) def _fsync_parent(path: Path) -> None: directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) try: directory_fd = os.open(path.parent, directory_flags) try: os.fsync(directory_fd) finally: os.close(directory_fd) except OSError as exc: raise RunBlocked("unable to sync artifact parent directory") from exc def reserve_artifact(path: Path, context: str) -> ReservedArtifact: expanded = path.expanduser() if not expanded.is_absolute(): expanded = Path.cwd() / expanded try: parent = expanded.parent.resolve(strict=True) except OSError as exc: raise RunBlocked(f"{context} parent directory does not exist") from exc target = parent / expanded.name if not parent.is_dir(): raise RunBlocked(f"{context} parent directory does not exist") if target.is_symlink(): raise RunBlocked(f"{context} must not be a symlink") flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW try: descriptor = os.open(target, flags, 0o600) except FileExistsError as exc: raise RunBlocked(f"{context} already exists") from exc except OSError as exc: raise RunBlocked(f"unable to reserve {context}") from exc try: os.fchmod(descriptor, 0o600) os.fsync(descriptor) _fsync_parent(target) except Exception as exc: os.close(descriptor) try: target.unlink() _fsync_parent(target) except Exception: pass if isinstance(exc, RunBlocked): raise raise RunBlocked(f"unable to durably reserve {context}") from exc return ReservedArtifact(path=target, fd=descriptor) def reserve_execution_artifacts( receipt_path: Path, stderr_log_path: Path, ) -> tuple[ReservedArtifact, ReservedArtifact]: def output_target(path: Path) -> Path: expanded = path.expanduser() if not expanded.is_absolute(): expanded = Path.cwd() / expanded return expanded.parent.resolve(strict=True) / expanded.name if output_target(receipt_path) == output_target(stderr_log_path): raise RunBlocked("receipt and stderr log paths must differ") receipt: ReservedArtifact | None = None try: receipt = reserve_artifact(receipt_path, "receipt output") stderr_log = reserve_artifact(stderr_log_path, "runtime stderr log") return receipt, stderr_log except Exception: if receipt is not None: receipt.close() try: receipt.path.unlink() _fsync_parent(receipt.path) except OSError: pass raise def host_execution_receipt( plan: ValidatedPlan, runtime_binding: ValidatedRuntimeBinding, *, mode: str, outcome: str, block_reason: str, ) -> dict[str, Any]: return { "schema_version": 1, "host_receipt_kind": "hubspot_authored_identity_recovery_guard", "command": COMMAND, "mode": mode, "manifest_sha256": plan.manifest_sha256, "runtime_binding_sha256": runtime_binding.runtime_binding_sha256, "release_json_sha256": runtime_binding.release_json.sha256, "runtime_env_sha256": runtime_binding.runtime_env.sha256, "setup_binary_sha256": runtime_binding.setup_binary.sha256, "deployed_tuple_sha256": runtime_binding.deployed_tuple_sha256, "outcome": outcome, "committed": None, "readback_verified": False, "enrollment_allowed": False, "result_contract_valid": False, "block_reason": block_reason, } def write_receipt(path: Path, receipt: dict[str, Any]) -> None: artifact = reserve_artifact(path, "receipt output") try: artifact.overwrite_json(receipt) finally: artifact.close() def _resolve_setup_bin(value: str | None) -> Path: raw = (value or os.environ.get("NEXUSCRM_SETUP_BIN", "")).strip() if not raw: raise RunBlocked("--setup-bin or NEXUSCRM_SETUP_BIN is required") path = Path(raw).expanduser() if not path.is_absolute(): raise RunBlocked("setup binary path must be absolute") path = path.resolve() if not path.is_file() or not os.access(path, os.X_OK): raise RunBlocked("setup binary path is not an executable file") return path def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run the manifest-bound NexusCRM #3765 recovery batch (dry-run by default)." ) parser.add_argument("--manifest", required=True) parser.add_argument("--manifest-sha256", required=True) parser.add_argument("--release-json", required=True) parser.add_argument("--runtime-env", required=True) parser.add_argument("--setup-bin") parser.add_argument("--apply", action="store_true") parser.add_argument("--dry-run-receipt") parser.add_argument("--dry-run-receipt-sha256") parser.add_argument("--backup-marker") parser.add_argument("--backup-marker-sha256") parser.add_argument("--receipt-out", required=True) parser.add_argument("--stderr-log", required=True) return parser.parse_args(argv) def _validate_mode_args(args: argparse.Namespace) -> None: names = ( "dry_run_receipt", "dry_run_receipt_sha256", "backup_marker", "backup_marker_sha256", ) present = [bool(getattr(args, name)) for name in names] if args.apply and not all(present): raise RunBlocked("--apply requires dry-run receipt and backup marker files plus both raw SHA-256 values") if not args.apply and any(present): raise RunBlocked("dry-run rejects apply-only receipt and backup arguments") def main(argv: list[str] | None = None) -> int: args = parse_args(argv) runtime_binding: ValidatedRuntimeBinding | None = None try: _validate_mode_args(args) plan = load_plan( Path(args.manifest), args.manifest_sha256, ) setup_bin = _resolve_setup_bin(args.setup_bin) runtime_binding = load_runtime_binding( plan, release_json_path=Path(args.release_json), runtime_env_path=Path(args.runtime_env), setup_binary_path=setup_bin, ) runner = SubprocessRuntimeRunner() dry_run_receipt = None backup_marker = None mode = "apply" if args.apply else "dry_run" if args.apply: dry_run_receipt = load_dry_run_receipt( Path(args.dry_run_receipt), args.dry_run_receipt_sha256, plan, runtime_binding, ) backup_marker = load_backup_marker( Path(args.backup_marker), args.backup_marker_sha256, plan, ) revalidate_runtime_binding(runtime_binding) receipt_artifact, stderr_artifact = reserve_execution_artifacts( Path(args.receipt_out), Path(args.stderr_log) ) try: receipt_artifact.overwrite_json( host_execution_receipt( plan, runtime_binding, mode=mode, outcome="runtime_started_outcome_unknown", block_reason="RUNTIME_UNAVAILABLE", ) ) try: result = execute( plan, runtime_binding, runner, mode=mode, dry_run_receipt=dry_run_receipt, backup_marker=backup_marker, stderr_fd=stderr_artifact.fd, ) except Exception: unknown = host_execution_receipt( plan, runtime_binding, mode=mode, outcome="runtime_outcome_unknown", block_reason="RUNTIME_RESULT_INVALID", ) receipt_artifact.overwrite_json(unknown) stderr_artifact.sync() print(json.dumps(unknown, indent=2, sort_keys=True, ensure_ascii=True)) print( "BLOCK: runtime outcome is unknown; targeted enrollment is forbidden", file=sys.stderr, ) return 2 stderr_artifact.sync() receipt_artifact.overwrite_json(result.receipt) print(json.dumps(result.receipt, indent=2, sort_keys=True, ensure_ascii=True)) if result.blocked: print( f"BLOCK: runtime outcome {result.receipt['outcome']}; targeted enrollment is forbidden", file=sys.stderr, ) return 2 return 0 finally: receipt_artifact.close() stderr_artifact.close() except RunBlocked as exc: print(f"BLOCK: {exc}", file=sys.stderr) return 2 finally: if runtime_binding is not None: runtime_binding.close() if __name__ == "__main__": raise SystemExit(main())