#!/usr/bin/env python3 """Workstation orchestrator for Overdeck K3s Phase 1. The command converges a non-secret control-plane configuration contract, proves K3s health, creates an encrypted off-host datastore+token backup, verifies that backup offline, proves API/CA reachability from every registry-reachable node, and installs a recurring workstation backup timer. """ from __future__ import annotations import argparse import hashlib import ipaddress import json import os import pwd import secrets import shlex import shutil import signal import subprocess import sys import tempfile import threading import time from pathlib import Path from typing import Any, Mapping, Sequence from urllib.parse import urlparse SCRIPT = Path(__file__).resolve() K3S_DIR = SCRIPT.parent LIB_DIR = K3S_DIR / "lib" sys.path.insert(0, str(LIB_DIR)) from phase1_common import ( # noqa: E402 HostAccess, Phase1Error, absolute_path_without_symlink_resolution, atomic_write_json, build_ssh_command, eligible_reachable_hosts, exclusive_lock, load_host_registry, normalize_path, prune_backup_pairs, read_json, require_commands, resolve_host_access, run_command, safe_name, sha256_file, utc_now, utc_stamp, ) REMOTE_INSTALL_PATH = "/usr/local/libexec/overdeck/k3s-phase1-server.py" REMOTE_HELPER = K3S_DIR / "remote/phase1-server.py" VERIFY_BACKUP = K3S_DIR / "verify-backup.py" SYSTEMD_SERVICE = K3S_DIR / "systemd/overdeck-k3s-backup.service" SYSTEMD_TIMER = K3S_DIR / "systemd/overdeck-k3s-backup.timer" class Logger: def __init__(self, log_path: Path) -> None: log_path.parent.mkdir(parents=True, exist_ok=True) self.log_path = log_path def __call__(self, message: str) -> None: line = f"[phase1 {utc_now()}] {message}" print(line, file=sys.stderr, flush=True) with self.log_path.open("a", encoding="utf-8") as handle: handle.write(line + "\n") class Orchestrator: def __init__(self, args: argparse.Namespace) -> None: self.args = args self.repo_root = normalize_path(args.repo_root) self.registry_path = normalize_path(args.host_registry or self.repo_root / "modules/workstation/claude/buildbox-hosts.json") self.kubeconfig = normalize_path(args.kubeconfig) self.backup_dir = normalize_path(args.backup_dir) self.age_identity = absolute_path_without_symlink_resolution(args.age_identity) self.lock_file = normalize_path( args.lock_file or Path.home() / ".local/state/overdeck" / f"k3s-phase1-{safe_name(args.server)}.lock" ) self.receipt_dir = normalize_path(args.receipt_dir or self.default_receipt_dir()) self.receipt_dir.mkdir(parents=True, exist_ok=False) os.chmod(self.receipt_dir, 0o700) (self.receipt_dir / "logs").mkdir(mode=0o700) self.log = Logger(self.receipt_dir / "logs/phase1.log") self.registry = load_host_registry(self.registry_path) self.server_access = resolve_host_access( self.registry, args.server, preferred_door=args.ssh_door, require_reachable=True, ) self.ssh = build_ssh_command(self.server_access, timeout=args.ssh_timeout) self.remote_installed = False self.remote_helper_path = REMOTE_INSTALL_PATH self.remote_temp_path: str | None = None self.qualification: dict[str, Any] = { "schema_version": 2, "status": "running", "started_utc": utc_now(), "checks": [], } self.backup_artifacts: dict[str, dict[str, Any]] = {} self.current_backup_purpose: str | None = None self.converge_tx: str | None = None self.converge_changed = False self.backup_tx: str | None = None self.remote_archive: str | None = None self.local_backup: Path | None = None self.backup_verified = False self.timer_rollback: dict[str, Any] | None = None self.linger_enabled_by_phase = False self.plan_doc: dict[str, Any] | None = None self.result: dict[str, Any] = { "schema_version": 2, "mode": args.mode, "status": "running", "started_utc": utc_now(), "server": args.server, "repo_root": str(self.repo_root), "kubeconfig": str(self.kubeconfig), "backup_dir": str(self.backup_dir), "receipt_dir": str(self.receipt_dir), "lock_file": str(self.lock_file), "steps": [], } def default_receipt_dir(self) -> Path: if self.args.mode == "backup": base = self.backup_dir / "receipts" prefix = "scheduled" else: base = Path.cwd() prefix = "live" return base / f"overdeck-k3s-phase1-{prefix}-{utc_stamp()}-{os.getpid()}" def add_step(self, name: str, status: str, **detail: Any) -> None: item = {"name": name, "status": status, "at_utc": utc_now()} item.update(detail) self.result["steps"].append(item) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) def save_json(self, relative: str, value: Any) -> None: atomic_write_json(self.receipt_dir / relative, value) def remote_call(self, *arguments: str, timeout: int = 300) -> dict[str, Any]: remote_command = shlex.join(["sudo", "-n", "/usr/bin/python3", self.remote_helper_path, *arguments]) command = self.ssh + [remote_command] completed = run_command(command, check=False, timeout=timeout) stdout = completed.stdout if isinstance(completed.stdout, str) else completed.stdout.decode(errors="replace") stderr = completed.stderr if isinstance(completed.stderr, str) else completed.stderr.decode(errors="replace") try: payload = json.loads(stdout) except json.JSONDecodeError as exc: raise Phase1Error( f"remote helper returned invalid JSON (exit {completed.returncode})\n" f"stdout:\n{stdout[-3000:]}\nstderr:\n{stderr[-3000:]}" ) from exc if completed.returncode != 0 or payload.get("status") == "error": command_name = safe_name(arguments[0]) if arguments else "unknown" # The root helper emits only explicitly sanitized diagnostics. Persist # that structured payload so a live preflight failure can be repaired # without asking the operator to run ad-hoc secret-bearing commands. self.save_json(f"remote-{command_name}-error.json", payload) raise Phase1Error(str(payload.get("error") or stderr or f"remote helper exited {completed.returncode}")) return payload def install_remote_helper(self) -> None: """Upload a checksum-verified ephemeral helper for this invocation. Qualification must not persist a privileged helper on the server. The recurring workstation service repeats this same upload on every backup, so no successful Phase 1 path depends on stale server-side code. """ if not REMOTE_HELPER.is_file() or REMOTE_HELPER.is_symlink(): raise Phase1Error(f"remote helper is missing or unsafe: {REMOTE_HELPER}") require_commands(["ssh", "python3"]) remote_tmp = f"/tmp/overdeck-k3s-phase1-{os.getpid()}-{secrets.token_hex(4)}.py" upload = self.ssh + ["umask 077; cat > " + shlex.quote(remote_tmp)] run_command(upload, input_bytes=REMOTE_HELPER.read_bytes(), timeout=60) run_command(self.ssh + [shlex.join(["chmod", "0700", remote_tmp])], timeout=30) expected = sha256_file(REMOTE_HELPER) actual = run_command( self.ssh + [shlex.join(["sha256sum", remote_tmp])], timeout=30 ).stdout.split()[0] if actual != expected: raise Phase1Error(f"remote helper SHA-256 mismatch: expected {expected}, got {actual}") self.remote_helper_path = remote_tmp self.remote_temp_path = remote_tmp self.remote_installed = True self.add_step( "install-remote-helper", "passed", sha256=expected, path=remote_tmp, persistence="ephemeral", ) def cleanup_ephemeral_helper(self) -> None: if not self.remote_temp_path: return completed = run_command( self.ssh + [shlex.join(["rm", "-f", self.remote_temp_path])], check=False, timeout=30, ) if completed.returncode != 0: self.result["ephemeral_helper_cleanup_error"] = completed.stderr[-2000:] self.remote_temp_path = None def endpoint_from_kubeconfig(self, fallback_ip: str) -> str: if not self.kubeconfig.is_file(): raise Phase1Error(f"K3s kubeconfig is missing: {self.kubeconfig}") require_commands(["kubectl"]) completed = run_command( ["kubectl", "--kubeconfig", str(self.kubeconfig), "config", "view", "--minify", "-o", "json"], timeout=30, ) try: document = json.loads(completed.stdout) server = document["clusters"][0]["cluster"]["server"] except (json.JSONDecodeError, KeyError, IndexError, TypeError) as exc: raise Phase1Error("cannot determine API endpoint from kubeconfig") from exc parsed = urlparse(str(server)) if parsed.scheme != "https" or not parsed.hostname: raise Phase1Error(f"unsupported Kubernetes API endpoint: {server!r}") if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}: raise Phase1Error("Kubernetes API endpoint must not contain credentials, path, query, or fragment") try: port = parsed.port or 443 except ValueError as exc: raise Phase1Error(f"Kubernetes API endpoint has an invalid port: {server!r}") from exc if not (1 <= port <= 65535): raise Phase1Error(f"Kubernetes API endpoint port is out of range: {port}") host = parsed.hostname if host in {"127.0.0.1", "localhost", "::1"}: # Preserve the kubeconfig's explicit API port. Only the loopback # host is remapped to the server's verified Tailscale identity. host = fallback_ip try: host = str(ipaddress.ip_address(host)) except ValueError: labels = host.rstrip(".").split(".") if len(host) > 253 or any( not label or len(label) > 63 or not label[0].isalnum() or not label[-1].isalnum() or any(not (char.isalnum() or char == "-") for char in label) for label in labels ): raise Phase1Error(f"Kubernetes API endpoint hostname is invalid: {host!r}") host = host.rstrip(".").lower() rendered_host = f"[{host}]" if ":" in host else host return f"https://{rendered_host}:{port}" @staticmethod def san_values(endpoint: str, inspect: Mapping[str, Any], access: HostAccess) -> list[str]: parsed = urlparse(endpoint) candidates = [ parsed.hostname, inspect.get("tailscale", {}).get("ipv4"), inspect.get("tailscale", {}).get("dns_name"), inspect.get("hostname"), access.host, access.magic_dns, ] values: list[str] = [] seen: set[str] = set() for raw in candidates: if not isinstance(raw, str): continue value = raw.strip().rstrip(".") if not value or value in seen: continue seen.add(value) values.append(value) return values def inspect_server(self) -> dict[str, Any]: inspect = self.remote_call("inspect", timeout=150) self.save_json("server-inspect.json", inspect) registry_ip = self.server_access.host if self.server_access.door == "tailscale_ip" else None observed_ip = inspect.get("tailscale", {}).get("ipv4") if registry_ip and observed_ip != registry_ip: raise Phase1Error( f"server Tailscale identity drift: registry={registry_ip!r} observed={observed_ip!r}" ) if not isinstance(observed_ip, str) or not observed_ip: raise Phase1Error("server did not report exactly one Tailscale IPv4 address") readiness = inspect.get("service", {}).get("ready") or {} if inspect.get("service", {}).get("state") != "active" or readiness.get("ready") is not True: raise Phase1Error( "server K3s readiness failed: " f"classification={readiness.get('classification')} detail={readiness.get('detail')}" ) node_probe = inspect.get("node_probe") or {} if node_probe.get("ok") is not True or int(node_probe.get("ready_count") or 0) < 1: raise Phase1Error( "server Node inventory qualification failed: " f"classification={node_probe.get('classification')} detail={node_probe.get('detail')}" ) launcher = inspect.get("k3s", {}).get("launcher") or {} if not isinstance(launcher.get("sha256"), str) or len(launcher["sha256"]) != 64: raise Phase1Error("server did not report a valid outer K3s launcher digest") self.add_step( "inspect-server", "passed", launcher=launcher.get("invocation_path"), runtime=(inspect.get("k3s", {}).get("runtime") or {}).get("resolved_path"), datastore=inspect.get("datastore", {}).get("type"), ) return inspect def plan_control_plane(self, inspect: Mapping[str, Any]) -> tuple[str, list[str], dict[str, Any]]: observed_ip = str(inspect.get("tailscale", {}).get("ipv4")) endpoint = self.endpoint_from_kubeconfig(observed_ip) sans = self.san_values(endpoint, inspect, self.server_access) plan = self.remote_call("plan", "--endpoint", endpoint, "--sans-json", json.dumps(sans), timeout=180) self.save_json("control-plane-plan.json", plan) self.plan_doc = plan self.result["endpoint"] = endpoint self.result["tls_sans"] = sans self.result["k3s_version"] = inspect.get("k3s", {}).get("version") self.result["k3s_launcher_sha256"] = (inspect.get("k3s", {}).get("launcher") or {}).get("sha256") self.result["k3s_runtime_sha256"] = (inspect.get("k3s", {}).get("runtime") or {}).get("sha256") self.result["datastore"] = inspect.get("datastore", {}).get("type") self.add_step("plan-control-plane", "passed", changed=plan.get("changed"), endpoint=endpoint) return endpoint, sans, plan def inspect_and_plan(self) -> tuple[dict[str, Any], str, list[str]]: inspect = self.inspect_server() endpoint, sans, _plan = self.plan_control_plane(inspect) return inspect, endpoint, sans def ensure_age(self) -> str: if shutil.which("age") is None or shutil.which("age-keygen") is None: if not self.args.install_age: raise Phase1Error("age and age-keygen are required; rerun with --install-age") self.install_age_package() require_commands(["age", "age-keygen"]) if self.age_identity.is_symlink(): raise Phase1Error(f"age identity must not be a symlink: {self.age_identity}") if self.age_identity.parent.is_symlink(): raise Phase1Error(f"age identity parent must not be a symlink: {self.age_identity.parent}") if not self.age_identity.exists(): self.age_identity.parent.mkdir(parents=True, exist_ok=True) os.chmod(self.age_identity.parent, 0o700) run_command(["age-keygen", "-o", str(self.age_identity)], timeout=30) os.chmod(self.age_identity, 0o600) if self.age_identity.is_symlink() or not self.age_identity.is_file(): raise Phase1Error(f"age identity must be a regular non-symlink file: {self.age_identity}") info = self.age_identity.stat() if info.st_uid != os.getuid(): raise Phase1Error(f"age identity is not owned by the current user: {self.age_identity}") if info.st_mode & 0o077: raise Phase1Error(f"age identity permissions are too broad: {self.age_identity}") recipient = run_command(["age-keygen", "-y", str(self.age_identity)], timeout=30).stdout.strip() if not recipient.startswith("age1"): raise Phase1Error("age-keygen did not return a valid recipient") self.add_step("age-identity", "passed", identity_path=str(self.age_identity), recipient=recipient) return recipient def install_age_package(self) -> None: manager: list[str] if shutil.which("apt-get"): manager = ["sudo", "-n", "apt-get", "install", "-y", "age"] elif shutil.which("dnf"): manager = ["sudo", "-n", "dnf", "install", "-y", "age"] elif shutil.which("pacman"): manager = ["sudo", "-n", "pacman", "-S", "--noconfirm", "age"] elif shutil.which("zypper"): manager = ["sudo", "-n", "zypper", "--non-interactive", "install", "age"] else: raise Phase1Error("age is missing and no supported package manager was found") self.log("installing the age package on the workstation") run_command(manager, timeout=600) def converge(self, endpoint: str, sans: list[str]) -> dict[str, Any]: tx = f"phase1-{utc_stamp()}-{secrets.token_hex(3)}" # Arm rollback before the remote submit so an SSH disconnect after the # server committed the transaction is treated as ambiguous and recovered. self.converge_tx = tx self.converge_changed = True result = self.remote_call( "converge", "--transaction", tx, "--endpoint", endpoint, "--sans-json", json.dumps(sans), timeout=360, ) self.save_json("converge.json", result) transaction_id = result.get("transaction_id") self.converge_tx = str(transaction_id) if transaction_id else None self.converge_changed = bool(result.get("changed")) self.add_step( "converge-control-plane", "passed", changed=self.converge_changed, transaction_id=self.converge_tx, ) return result @staticmethod def stop_process(process: subprocess.Popen[Any]) -> None: if process.poll() is not None: return process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) def stream_backup(self, recipient: str, purpose: str) -> tuple[Path, dict[str, Any]]: if purpose not in {"prechange", "postchange", "scheduled"}: raise Phase1Error(f"unsupported backup purpose: {purpose}") tx = f"backup-{purpose}-{utc_stamp()}-{secrets.token_hex(3)}" self.backup_tx = tx self.current_backup_purpose = purpose self.backup_verified = False staged = self.remote_call( "backup", "--transaction", tx, "--purpose", purpose, timeout=900 ) self.remote_archive = str(staged["archive_path"]) self.save_json(f"backup-{purpose}-staged.json", staged) self.backup_dir.mkdir(parents=True, exist_ok=True) os.chmod(self.backup_dir, 0o700) prefix = f"overdeck-k3s-{safe_name(self.args.server)}" backup_suffix = tx.removeprefix(f"backup-{purpose}-") backup = self.backup_dir / f"{prefix}-{purpose}-{backup_suffix}.tar.age" if backup.exists() or backup.is_symlink(): raise Phase1Error(f"refusing to overwrite an existing backup path: {backup}") temporary = backup.with_name(f".{backup.name}.{os.getpid()}.tmp") command = self.ssh + [shlex.join(["sudo", "-n", "cat", self.remote_archive])] self.log(f"streaming and encrypting {purpose} server backup to {backup}") digest = hashlib.sha256() plaintext_size = 0 pump_errors: list[str] = [] with tempfile.TemporaryFile() as ssh_error, tempfile.TemporaryFile() as age_error: ssh_proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=ssh_error) try: age_proc = subprocess.Popen( ["age", "-r", recipient, "-o", str(temporary)], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=age_error, ) except Exception: self.stop_process(ssh_proc) temporary.unlink(missing_ok=True) raise if ssh_proc.stdout is None or age_proc.stdin is None: self.stop_process(ssh_proc) self.stop_process(age_proc) raise Phase1Error("failed to create the encrypted backup pipeline") def pump() -> None: nonlocal plaintext_size try: while True: chunk = ssh_proc.stdout.read(1024 * 1024) if not chunk: break digest.update(chunk) plaintext_size += len(chunk) age_proc.stdin.write(chunk) age_proc.stdin.close() except (BrokenPipeError, OSError) as exc: pump_errors.append(f"{type(exc).__name__}: {exc}") try: age_proc.stdin.close() except OSError: pass thread = threading.Thread(target=pump, name="phase1-backup-stream", daemon=True) thread.start() thread.join(timeout=1200) if thread.is_alive(): self.stop_process(ssh_proc) self.stop_process(age_proc) thread.join(timeout=10) temporary.unlink(missing_ok=True) raise Phase1Error("encrypted backup stream exceeded the 1200-second deadline") try: ssh_rc = ssh_proc.wait(timeout=60) age_rc = age_proc.wait(timeout=60) except subprocess.TimeoutExpired as exc: self.stop_process(ssh_proc) self.stop_process(age_proc) temporary.unlink(missing_ok=True) raise Phase1Error("encrypted backup pipeline did not terminate cleanly") from exc ssh_error.seek(0) age_error.seek(0) ssh_stderr = ssh_error.read().decode(errors="replace") age_stderr = age_error.read().decode(errors="replace") expected_size = staged.get("archive_size") expected_sha = staged.get("archive_sha256") actual_sha = digest.hexdigest() if ( ssh_rc != 0 or age_rc != 0 or pump_errors or not isinstance(expected_size, int) or plaintext_size != expected_size or not isinstance(expected_sha, str) or actual_sha != expected_sha ): temporary.unlink(missing_ok=True) raise Phase1Error( f"backup stream validation failed: ssh_rc={ssh_rc} age_rc={age_rc} " f"size={plaintext_size}/{expected_size} sha256={actual_sha}/{expected_sha} " f"pump={pump_errors}\nssh stderr:\n{ssh_stderr[-3000:]}\n" f"age stderr:\n{age_stderr[-3000:]}" ) os.chmod(temporary, 0o600) with temporary.open("rb") as handle: os.fsync(handle.fileno()) os.replace(temporary, backup) directory_fd = os.open(backup.parent, os.O_DIRECTORY) try: os.fsync(directory_fd) finally: os.close(directory_fd) self.local_backup = backup encrypted = { "schema_version": 2, "status": "encrypted", "purpose": purpose, "created_utc": utc_now(), "server": self.args.server, "path": str(backup), "size": backup.stat().st_size, "sha256": sha256_file(backup), "recipient": recipient, "remote_manifest": staged.get("manifest"), "remote_archive_sha256": staged.get("archive_sha256"), } self.save_json(f"backup-{purpose}-encrypted.json", encrypted) self.add_step( f"off-host-encrypted-{purpose}-backup", "passed", path=str(backup), sha256=encrypted["sha256"] ) return backup, encrypted def verify_encrypted_backup(self, backup: Path, purpose: str) -> dict[str, Any]: with tempfile.TemporaryDirectory(prefix=f"overdeck-k3s-phase1-{purpose}-decrypt-") as tmp: decrypted = Path(tmp) / "backup.tar" restore_root = Path(tmp) / "isolated-restore" run_command( ["age", "-d", "-i", str(self.age_identity), "-o", str(decrypted), str(backup)], timeout=1200, ) verified = run_command( [ sys.executable, str(VERIFY_BACKUP), "--archive", str(decrypted), "--restore-root", str(restore_root), ], timeout=600, ) try: document = json.loads(verified.stdout) except json.JSONDecodeError as exc: raise Phase1Error("backup verifier returned invalid JSON") from exc if document.get("status") != "verified": raise Phase1Error(f"backup verifier did not return verified: {document}") isolated = document.get("isolated_restore") if not isinstance(isolated, dict) or isolated.get("status") != "materialized": raise Phase1Error("backup verifier did not complete isolated restore materialization") isolated["target_root"] = "" isolated["retained_after_verification"] = False document["archive"] = "" self.save_json(f"backup-{purpose}-verify.json", document) metadata_path = backup.with_suffix("").with_suffix(".json") metadata = { "schema_version": 2, "status": "verified", "purpose": purpose, "verified_utc": document.get("verified_utc"), "server": document.get("server"), "k3s": document.get("k3s"), "datastore": document.get("datastore"), "agent_token": document.get("isolated_restore", {}).get("agent_token"), "canonical_config": document.get("canonical_config"), "encrypted_path": str(backup), "encrypted_size": backup.stat().st_size, "encrypted_sha256": sha256_file(backup), "age_identity_path": str(self.age_identity), "secret_values_recorded": False, } atomic_write_json(metadata_path, metadata) self.backup_verified = True self.backup_artifacts[purpose] = {"encrypted": str(backup), "metadata": str(metadata_path), "verification": document} self.add_step(f"offline-{purpose}-restore-proof", "passed", datastore=document.get("datastore", {}).get("type")) return document def cleanup_remote_backup(self) -> None: if not self.backup_tx: return tx = self.backup_tx purpose = self.current_backup_purpose or "unknown" result = self.remote_call("cleanup-backup", "--transaction", tx, timeout=120) self.save_json(f"backup-{purpose}-cleanup.json", result) self.add_step(f"remote-{purpose}-backup-cleanup", "passed", transaction_id=tx) self.remote_archive = None self.backup_tx = None def probe_reachable_nodes(self, endpoint: str, expected_ca_hash: str) -> list[dict[str, Any]]: script = r'''import hashlib, json, os, socket, ssl, sys, tempfile, urllib.request from urllib.parse import urlparse endpoint=sys.argv[1].rstrip('/') url=endpoint+'/cacerts' parsed=urlparse(endpoint) port=parsed.port or 443 ctx=ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE try: with urllib.request.urlopen(urllib.request.Request(url, headers={'User-Agent':'overdeck-k3s-phase1'}), context=ctx, timeout=10) as r: body=r.read() fd, ca_path=tempfile.mkstemp(prefix='overdeck-k3s-ca-', suffix='.pem') try: os.fchmod(fd, 0o600) with os.fdopen(fd, 'wb') as f: f.write(body) verify=ssl.create_default_context(cafile=ca_path) with socket.create_connection((parsed.hostname, port), timeout=10) as raw: with verify.wrap_socket(raw, server_hostname=parsed.hostname) as tls: peer=tls.getpeercert() sans=[value for kind,value in peer.get('subjectAltName',[]) if kind in {'DNS','IP Address'}] finally: try: os.unlink(ca_path) except OSError: pass print(json.dumps({'status':'reachable','url':url,'sha256':hashlib.sha256(body).hexdigest(),'size':len(body),'tls_verified':True,'peer_sans':sans})) except Exception as exc: print(json.dumps({'status':'error','url':url,'tls_verified':False,'error':f'{type(exc).__name__}: {exc}'})) raise SystemExit(2) ''' results: list[dict[str, Any]] = [] for name in eligible_reachable_hosts(self.registry): access = resolve_host_access(self.registry, name, preferred_door=self.args.ssh_door, require_reachable=True) remote_command = shlex.join(["python3", "-", endpoint]) command = build_ssh_command(access, timeout=self.args.ssh_timeout) + [remote_command] completed = run_command(command, input_text=script, check=False, timeout=45) try: payload = json.loads(completed.stdout) except json.JSONDecodeError: payload = {"status": "error", "error": "probe returned invalid JSON"} payload["host"] = name payload["ssh_exit_code"] = completed.returncode payload["ca_match"] = payload.get("sha256") == expected_ca_hash results.append(payload) self.save_json("node-api-probes.json", results) failed = [ item for item in results if item.get("status") != "reachable" or not item.get("ca_match") or item.get("tls_verified") is not True ] if failed: raise Phase1Error(f"API/CA/TLS reachability failed for: {', '.join(str(item['host']) for item in failed)}") self.add_step("node-api-ca-tls-reachability", "passed", hosts=[item["host"] for item in results]) return results def kubectl_ready(self) -> dict[str, Any]: completed = run_command( ["kubectl", "--kubeconfig", str(self.kubeconfig), "--request-timeout=15s", "get", "--raw=/readyz"], timeout=30, ) detail = completed.stdout.strip() if "ok" not in detail.lower(): raise Phase1Error(f"Kubernetes API readyz did not return ok: {detail!r}") result = {"status": "ready", "detail": detail, "kubeconfig": str(self.kubeconfig)} self.add_step("workstation-kubectl-readyz", "passed", detail=detail) return result @staticmethod def atomic_copy(source: Path, target: Path, *, mode: int | None = None) -> None: target.parent.mkdir(parents=True, exist_ok=True) fd, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) temporary = Path(temporary_name) try: with os.fdopen(fd, "wb") as destination, source.open("rb") as incoming: shutil.copyfileobj(incoming, destination, length=1024 * 1024) destination.flush() os.fsync(destination.fileno()) os.chmod(temporary, mode if mode is not None else source.stat().st_mode & 0o777) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) @staticmethod def atomic_text(target: Path, value: str, *, mode: int = 0o600) -> None: target.parent.mkdir(parents=True, exist_ok=True) fd, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) temporary = Path(temporary_name) try: os.fchmod(fd, mode) with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(value) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) def capture_timer_rollback(self, paths: Sequence[Path]) -> dict[str, Any]: rollback_dir = self.receipt_dir / ".timer-rollback" if rollback_dir.exists(): raise Phase1Error(f"timer rollback directory already exists: {rollback_dir}") try: files_dir = rollback_dir / "files" files_dir.mkdir(parents=True, mode=0o700) records: list[dict[str, Any]] = [] for index, target in enumerate(paths): record: dict[str, Any] = {"index": index, "path": str(target)} if target.is_symlink(): record.update({"type": "symlink", "target": os.readlink(target)}) elif target.is_file(): record.update({"type": "file", "mode": target.stat().st_mode & 0o777}) shutil.copy2(target, files_dir / str(index), follow_symlinks=False) elif target.exists(): raise Phase1Error(f"timer-managed path is not a regular file: {target}") else: record["type"] = "absent" records.append(record) enabled = run_command( ["systemctl", "--user", "is-enabled", SYSTEMD_TIMER.name], check=False, timeout=20 ).returncode == 0 active = run_command( ["systemctl", "--user", "is-active", SYSTEMD_TIMER.name], check=False, timeout=20 ).returncode == 0 document = { "schema_version": 1, "paths": records, "timer_was_enabled": enabled, "timer_was_active": active, } atomic_write_json(rollback_dir / "transaction.json", document) return {"directory": str(rollback_dir), **document} except Exception: shutil.rmtree(rollback_dir, ignore_errors=True) raise def restore_timer_files(self, transaction: Mapping[str, Any]) -> None: rollback_dir = Path(str(transaction["directory"])) files_dir = rollback_dir / "files" for record in transaction.get("paths", []): target = Path(str(record["path"])) if target.exists() or target.is_symlink(): if target.is_dir() and not target.is_symlink(): raise Phase1Error(f"refusing to replace directory during timer rollback: {target}") target.unlink() prior_type = record.get("type") if prior_type == "absent": continue target.parent.mkdir(parents=True, exist_ok=True) if prior_type == "symlink": os.symlink(str(record["target"]), target) elif prior_type == "file": shutil.copy2(files_dir / str(record["index"]), target, follow_symlinks=False) os.chmod(target, int(record.get("mode", 0o600))) else: raise Phase1Error(f"unknown timer rollback record type: {prior_type!r}") def rollback_timer(self) -> dict[str, Any] | None: if self.timer_rollback is None: return None errors: list[str] = [] run_command( ["systemctl", "--user", "disable", "--now", SYSTEMD_TIMER.name], check=False, timeout=60, ) try: self.restore_timer_files(self.timer_rollback) except Exception as exc: errors.append(f"file restore: {exc}") run_command(["systemctl", "--user", "daemon-reload"], check=False, timeout=30) if self.timer_rollback.get("timer_was_enabled"): completed = run_command( ["systemctl", "--user", "enable", SYSTEMD_TIMER.name], check=False, timeout=60 ) if completed.returncode != 0: errors.append(f"restore enable: {completed.stderr[-1000:]}") if self.timer_rollback.get("timer_was_active"): completed = run_command( ["systemctl", "--user", "start", SYSTEMD_TIMER.name], check=False, timeout=60 ) if completed.returncode != 0: errors.append(f"restore start: {completed.stderr[-1000:]}") user_name = pwd.getpwuid(os.getuid()).pw_name if self.linger_enabled_by_phase: completed = run_command( ["sudo", "-n", "loginctl", "disable-linger", user_name], check=False, timeout=60 ) if completed.returncode != 0: errors.append(f"disable linger: {completed.stderr[-1000:]}") result = {"status": "rolled-back" if not errors else "rollback-incomplete", "errors": errors} self.save_json("timer-rollback.json", result) self.timer_rollback = None return result def commit_timer(self) -> dict[str, Any] | None: if self.timer_rollback is None: return None rollback_dir = Path(str(self.timer_rollback["directory"])) # Crossing the commit point means the installed recurring timer must no # longer be rolled back even if receipt cleanup later fails. self.timer_rollback = None try: shutil.rmtree(rollback_dir) return {"status": "committed", "rollback_material_removed": True} except FileNotFoundError: return {"status": "committed", "rollback_material_removed": True} except OSError as exc: return { "status": "committed-with-cleanup-warning", "rollback_material_removed": False, "rollback_directory": str(rollback_dir), "error": str(exc), } @staticmethod def systemd_escape(value: Path | str) -> str: rendered = str(value) if any(character in rendered for character in "\r\n\0"): raise Phase1Error("systemd-managed paths must not contain control characters") output: list[str] = [] safe = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789/_.:+-") for character in rendered: if character == "%": output.append("%%") elif character in safe: output.append(character) else: output.extend(f"\\x{byte:02x}" for byte in character.encode("utf-8")) return "".join(output) def timer_preflight(self) -> dict[str, Any]: if self.args.no_timer: return {"status": "skipped", "reason": "--no-timer"} require_commands(["systemctl"]) for source in [ K3S_DIR / "phase1-control-plane.py", K3S_DIR / "phase1-control-plane.sh", K3S_DIR / "verify-backup.py", K3S_DIR / "lib/phase1_common.py", K3S_DIR / "remote/phase1-server.py", self.registry_path, SYSTEMD_SERVICE, SYSTEMD_TIMER, ]: if source.is_symlink() or not source.is_file(): raise Phase1Error(f"timer source is missing or unsafe: {source}") session = run_command(["systemctl", "--user", "show-environment"], check=False, timeout=20) if session.returncode != 0: raise Phase1Error("systemd user manager is unavailable for recurring backups") install_root = normalize_path(self.args.install_root) config_path = normalize_path(self.args.timer_config) for target in [install_root, config_path.parent, self.backup_dir, self.lock_file.parent]: if target.exists() and (target.is_symlink() or not target.is_dir()): raise Phase1Error(f"timer target parent is unsafe: {target}") with tempfile.TemporaryDirectory(prefix="overdeck-k3s-timer-preflight-") as tmp: stage = Path(tmp) staged_install = stage / "install" staged_install.mkdir() staged_script = staged_install / "phase1-control-plane.sh" shutil.copy2(K3S_DIR / "phase1-control-plane.sh", staged_script) staged_config = stage / "backup.json" staged_config.write_text("{}\n") service = ( SYSTEMD_SERVICE.read_text() .replace("@INSTALL_ROOT@", self.systemd_escape(staged_install)) .replace("@CONFIG_PATH@", self.systemd_escape(staged_config)) .replace("@BACKUP_DIR@", self.systemd_escape(stage / "backups")) .replace("@CONFIG_DIR@", self.systemd_escape(stage)) .replace("@LOCK_DIR@", self.systemd_escape(stage / "locks")) ) service_path = stage / SYSTEMD_SERVICE.name timer_path = stage / SYSTEMD_TIMER.name service_path.write_text(service) timer_path.write_text(SYSTEMD_TIMER.read_text()) if shutil.which("systemd-analyze"): run_command(["systemd-analyze", "--user", "verify", str(service_path), str(timer_path)], timeout=60) linger = {"requested": bool(self.args.enable_linger), "current": None, "sudo_available": None} if self.args.enable_linger: require_commands(["loginctl", "sudo"]) user_name = pwd.getpwuid(os.getuid()).pw_name probe = run_command(["loginctl", "show-user", user_name, "-p", "Linger", "--value"], check=False, timeout=30) if probe.returncode != 0: raise Phase1Error(f"cannot inspect user linger state: {probe.stderr[-1000:]}") linger["current"] = probe.stdout.strip().lower() or "unknown" if linger["current"] != "yes": sudo_probe = run_command(["sudo", "-n", "true"], check=False, timeout=20) linger["sudo_available"] = sudo_probe.returncode == 0 if sudo_probe.returncode != 0: raise Phase1Error("enabling user linger will require sudo, but non-interactive sudo is unavailable") result = { "status": "qualified", "install_root": str(install_root), "config_path": str(config_path), "backup_dir": str(self.backup_dir), "linger": linger, "persistent_server_helper_required": False, } self.save_json("timer-preflight.json", result) return result def install_timer(self, recipient: str) -> dict[str, Any]: if self.args.no_timer: result = {"status": "skipped", "reason": "--no-timer"} self.save_json("timer.json", result) self.add_step("recurring-backup-timer", "skipped", reason="--no-timer") return result require_commands(["systemctl"]) install_root = normalize_path(self.args.install_root) config_path = normalize_path(self.args.timer_config) unit_dir = Path.home() / ".config/systemd/user" copies = { K3S_DIR / "phase1-control-plane.py": install_root / "phase1-control-plane.py", K3S_DIR / "phase1-control-plane.sh": install_root / "phase1-control-plane.sh", K3S_DIR / "verify-backup.py": install_root / "verify-backup.py", K3S_DIR / "lib/phase1_common.py": install_root / "lib/phase1_common.py", K3S_DIR / "remote/phase1-server.py": install_root / "remote/phase1-server.py", self.registry_path: install_root / "config/buildbox-hosts.json", } service_path = unit_dir / SYSTEMD_SERVICE.name timer_path = unit_dir / SYSTEMD_TIMER.name managed_paths = [*copies.values(), config_path, service_path, timer_path] self.timer_rollback = self.capture_timer_rollback(managed_paths) try: run_command(["systemctl", "--user", "stop", SYSTEMD_TIMER.name], check=False, timeout=30) running_service = run_command( ["systemctl", "--user", "is-active", SYSTEMD_SERVICE.name], check=False, timeout=20 ) if running_service.returncode == 0: raise Phase1Error("an existing recurring K3s backup service is still active") executable_sources = { K3S_DIR / "phase1-control-plane.py", K3S_DIR / "phase1-control-plane.sh", K3S_DIR / "verify-backup.py", K3S_DIR / "remote/phase1-server.py", } for source, target in copies.items(): mode = 0o755 if source in executable_sources else 0o644 self.atomic_copy(source, target, mode=mode) config = { "schema_version": 1, "repo_root": str(self.repo_root), "server": self.args.server, "host_registry": str(install_root / "config/buildbox-hosts.json"), "kubeconfig": str(self.kubeconfig), "backup_dir": str(self.backup_dir), "age_identity": str(self.age_identity), "retention": self.args.retention, "ssh_timeout": self.args.ssh_timeout, "ssh_door": self.args.ssh_door, "install_root": str(install_root), "lock_file": str(self.lock_file), "recipient": recipient, } atomic_write_json(config_path, config) service = ( SYSTEMD_SERVICE.read_text() .replace("@INSTALL_ROOT@", self.systemd_escape(install_root)) .replace("@CONFIG_PATH@", self.systemd_escape(config_path)) .replace("@BACKUP_DIR@", self.systemd_escape(self.backup_dir)) .replace("@CONFIG_DIR@", self.systemd_escape(config_path.parent)) .replace("@LOCK_DIR@", self.systemd_escape(self.lock_file.parent)) ) timer = SYSTEMD_TIMER.read_text() self.atomic_text(service_path, service, mode=0o644) self.atomic_text(timer_path, timer, mode=0o644) run_command(["systemctl", "--user", "daemon-reload"], timeout=30) if shutil.which("systemd-analyze"): run_command(["systemd-analyze", "--user", "verify", str(service_path), str(timer_path)], timeout=60) run_command(["systemctl", "--user", "enable", "--now", SYSTEMD_TIMER.name], timeout=60) if self.args.enable_linger: require_commands(["loginctl", "sudo"]) user_name = pwd.getpwuid(os.getuid()).pw_name linger = run_command( ["loginctl", "show-user", user_name, "-p", "Linger", "--value"], check=False, timeout=30, ) if linger.returncode != 0: raise Phase1Error(f"cannot inspect user linger state: {linger.stderr[-1000:]}") if linger.stdout.strip().lower() != "yes": run_command(["sudo", "-n", "loginctl", "enable-linger", user_name], timeout=60) self.linger_enabled_by_phase = True status = run_command( [ "systemctl", "--user", "show", SYSTEMD_TIMER.name, "-p", "ActiveState", "-p", "UnitFileState", "-p", "NextElapseUSecRealtime", ], timeout=30, ).stdout result = { "status": "installed", "install_root": str(install_root), "config_path": str(config_path), "service_path": str(service_path), "timer_path": str(timer_path), "systemctl": status.strip().splitlines(), "linger_enabled_by_phase": self.linger_enabled_by_phase, } self.save_json("timer.json", result) self.add_step("recurring-backup-timer", "passed", timer=SYSTEMD_TIMER.name) return result except Exception: self.rollback_timer() raise def prune(self) -> list[str]: prefix = f"overdeck-k3s-{safe_name(self.args.server)}" removed = prune_backup_pairs(self.backup_dir, prefix=prefix, retain=self.args.retention) self.add_step("backup-retention", "passed", retain=self.args.retention, removed=removed) return removed def finalize_remote(self) -> dict[str, Any] | None: timer_commit: dict[str, Any] | None = None if self.converge_tx: tx = self.converge_tx try: result = self.remote_call("finalize", "--transaction", tx, timeout=120) except Phase1Error: # Finalization is idempotent. A single retry resolves the common # ambiguous-response case where the server committed before SSH # disconnected. time.sleep(1) result = self.remote_call("finalize", "--transaction", tx, timeout=120) if result.get("status") != "finalized": raise Phase1Error(f"control-plane transaction did not finalize: {result}") # Remote config and the installed timer are now committed. Clear # rollback state before any local receipt write can fail. self.converge_tx = None self.converge_changed = False timer_commit = self.commit_timer() self.save_json("converge-finalize.json", result) self.add_step("finalize-control-plane-transaction", "passed", transaction_id=tx) else: timer_commit = self.commit_timer() if timer_commit is not None: self.save_json("timer-commit.json", timer_commit) if timer_commit.get("status") != "committed": self.result["timer_commit_warning"] = timer_commit return timer_commit def rollback_remote(self, reason: str) -> dict[str, Any] | None: if not self.converge_tx or not self.converge_changed: return None try: result = self.remote_call( "rollback", "--transaction", self.converge_tx, "--reason", reason[:500], timeout=300, ) self.save_json("rollback.json", result) return result except Exception as exc: failure = {"status": "rollback-call-failed", "error": str(exc)} self.save_json("rollback.json", failure) return failure def mark_failed_backup(self, error: str) -> str | None: if self.local_backup is None or not self.local_backup.exists(): return None source = self.local_backup stem = source.name[:-8] if source.name.endswith(".tar.age") else source.name qualifier = "verified-phase-failed" if self.backup_verified else "unverified-phase-failed" failed = source.with_name(f"{stem}.{qualifier}.tar.age") os.replace(source, failed) old_metadata = source.with_suffix("").with_suffix(".json") new_metadata = failed.with_suffix("").with_suffix(".json") metadata: dict[str, Any] = {} if old_metadata.is_file(): try: loaded = read_json(old_metadata) if isinstance(loaded, dict): metadata.update(loaded) except Phase1Error: pass old_metadata.unlink(missing_ok=True) metadata.update( { "schema_version": 2, "status": qualifier, "purpose": self.current_backup_purpose, "phase_failed_utc": utc_now(), "phase_error": error[:2000], "encrypted_path": str(failed), "encrypted_size": failed.stat().st_size, "encrypted_sha256": sha256_file(failed), "backup_verified_before_phase_failure": self.backup_verified, "secret_values_recorded": False, } ) atomic_write_json(new_metadata, metadata) self.local_backup = failed return str(failed) def qualification_record(self, name: str, status: str, **detail: Any) -> None: item = {"name": name, "status": status, "at_utc": utc_now()} item.update(detail) self.qualification["checks"].append(item) atomic_write_json(self.receipt_dir / "qualification.json", self.qualification) def qualification_attempt(self, name: str, callback: Any) -> Any: try: value = callback() except Exception as exc: self.qualification_record(name, "failed", error=(str(exc) or type(exc).__name__)[:3000]) return None summary: dict[str, Any] = {} if isinstance(value, dict): for key in ("changed", "purpose", "path"): if key in value: summary[key] = value[key] if "status" in value: summary["result_status"] = value["status"] self.qualification_record(name, "passed", **summary) return value def qualification_blocked(self, name: str, dependencies: Sequence[str]) -> None: self.qualification_record(name, "blocked", dependencies=list(dependencies)) def backup_cycle(self, recipient: str, purpose: str) -> dict[str, Any]: backup, encrypted = self.stream_backup(recipient, purpose) try: verified = self.verify_encrypted_backup(backup, purpose) finally: # Remote plaintext must be removed even when offline verification fails. if self.backup_tx: self.cleanup_remote_backup() artifact = {"encrypted": encrypted, "verified": verified} self.backup_artifacts[purpose] = artifact self.local_backup = None self.backup_verified = False self.current_backup_purpose = None return artifact def run_qualification(self) -> dict[str, Any]: self.qualification.update({"status": "running", "server": self.args.server, "mode": self.args.mode}) recipient = self.qualification_attempt("age-identity-and-tools", self.ensure_age) timer_preflight = self.qualification_attempt("recurring-backup-preflight", self.timer_preflight) inspect = self.qualification_attempt("server-inspect-readiness-nodes", self.inspect_server) endpoint: str | None = None sans: list[str] | None = None plan: dict[str, Any] | None = None if isinstance(inspect, dict): planned = self.qualification_attempt("control-plane-plan", lambda: self.plan_control_plane(inspect)) if isinstance(planned, tuple): endpoint, sans, plan = planned else: self.qualification_blocked("control-plane-plan", ["server-inspect-readiness-nodes"]) workstation_ready = None if endpoint: workstation_ready = self.qualification_attempt("workstation-kubectl-readyz", self.kubectl_ready) else: self.qualification_blocked("workstation-kubectl-readyz", ["control-plane-plan"]) peers = None expected_ca = inspect.get("cacerts_sha256") if isinstance(inspect, dict) else None if endpoint and isinstance(expected_ca, str) and len(expected_ca) == 64: peers = self.qualification_attempt( "registry-host-api-ca-tls", lambda: self.probe_reachable_nodes(endpoint, expected_ca) ) else: self.qualification_blocked("registry-host-api-ca-tls", ["server-inspect-readiness-nodes", "control-plane-plan"]) prechange = None if recipient and inspect is not None and plan is not None: prechange = self.qualification_attempt( "prechange-backup-encrypt-verify-materialize", lambda: self.backup_cycle(str(recipient), "prechange"), ) else: self.qualification_blocked( "prechange-backup-encrypt-verify-materialize", ["age-identity-and-tools", "server-inspect-readiness-nodes", "control-plane-plan"], ) failures = [item for item in self.qualification["checks"] if item["status"] in {"failed", "blocked"}] self.qualification.update( { "status": "passed" if not failures else "failed", "finished_utc": utc_now(), "required_failures": failures, "prechange_backup_retained": bool(prechange), "cluster_configuration_mutated": False, } ) atomic_write_json(self.receipt_dir / "qualification.json", self.qualification) if failures: rendered = "; ".join(f"{item['name']}={item['status']}" for item in failures) raise Phase1Error(f"Phase 1 qualification did not pass: {rendered}") assert isinstance(inspect, dict) and endpoint and sans is not None and isinstance(plan, dict) and recipient return { "recipient": str(recipient), "timer_preflight": timer_preflight, "inspect": inspect, "endpoint": endpoint, "sans": sans, "plan": plan, "workstation_ready": workstation_ready, "peers": peers, "prechange": prechange, } def run_locked(self) -> int: try: self.log(f"starting mode={self.args.mode} server={self.args.server}") self.install_remote_helper() if self.args.mode == "recover": if self.args.recovery_action == "rollback": recovery = self.remote_call( "rollback", "--transaction", self.args.transaction, "--reason", "explicit Phase 1 recovery requested by trusted workstation", timeout=300, ) if recovery.get("status") != "rolled-back" or not (recovery.get("readiness") or {}).get("ready"): raise Phase1Error(f"recovery rollback did not return K3s Ready: {recovery}") else: recovery = self.remote_call("finalize", "--transaction", self.args.transaction, timeout=120) if recovery.get("status") != "finalized": raise Phase1Error(f"recovery finalization failed: {recovery}") post_recovery = self.remote_call("inspect", timeout=150) self.save_json("recovery.json", recovery) self.save_json("server-post-recovery.json", post_recovery) self.result.update( { "status": "success", "finished_utc": utc_now(), "recovery_action": self.args.recovery_action, "transaction": self.args.transaction, "recovery": recovery, } ) self.add_step("phase1-recovery", "passed", action=self.args.recovery_action) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) return 0 if self.args.mode == "backup": inspect = self.inspect_server() recipient = self.ensure_age() endpoint = self.endpoint_from_kubeconfig(str(inspect["tailscale"]["ipv4"])) expected_ca = inspect.get("cacerts_sha256") artifact = self.backup_cycle(recipient, "scheduled") self.kubectl_ready() if not isinstance(expected_ca, str) or len(expected_ca) != 64: raise Phase1Error("server did not expose a valid public CA bundle SHA-256") self.probe_reachable_nodes(endpoint, expected_ca) self.prune() self.result.update( {"status": "success", "finished_utc": utc_now(), "backup": artifact, "control_plane_changed": False} ) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) return 0 qualification = self.run_qualification() if self.args.mode in {"qualify", "plan"}: self.result.update( { "status": "qualified", "finished_utc": utc_now(), "qualification": qualification, "control_plane_changed": False, } ) self.add_step("phase1-qualification", "passed", mutation=False) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) return 0 endpoint = str(qualification["endpoint"]) sans = list(qualification["sans"]) recipient = str(qualification["recipient"]) self.converge(endpoint, sans) control_plane_changed = self.converge_changed control_plane_transaction = self.converge_tx post = self.remote_call("inspect", timeout=180) self.save_json("server-post-converge.json", post) readiness = post.get("service", {}).get("ready") or {} node_probe = post.get("node_probe") or {} if readiness.get("ready") is not True or node_probe.get("ok") is not True or int(node_probe.get("ready_count") or 0) < 1: raise Phase1Error("post-converge server readiness or Node inventory failed") expected_ca = post.get("cacerts_sha256") if not isinstance(expected_ca, str) or len(expected_ca) != 64: raise Phase1Error("post-converge server did not expose a valid public CA bundle SHA-256") self.kubectl_ready() self.probe_reachable_nodes(endpoint, expected_ca) postchange = self.backup_cycle(recipient, "postchange") self.install_timer(recipient) self.finalize_remote() try: self.prune() except Exception as prune_exc: self.result["retention_warning"] = str(prune_exc) self.add_step("backup-retention", "warning", error=str(prune_exc)) self.result.update( { "status": "success", "finished_utc": utc_now(), "qualification": qualification, "backups": { "prechange": qualification["prechange"], "postchange": postchange, }, "control_plane_changed": control_plane_changed, "control_plane_transaction": control_plane_transaction, "git_publication_allowed": True, } ) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) self.log("Phase 1 qualification and live apply gates completed") return 0 except (Exception, KeyboardInterrupt) as exc: error = str(exc) or type(exc).__name__ self.log(f"failure: {error}") try: timer_rollback = self.rollback_timer() except Exception as timer_exc: timer_rollback = { "status": "rollback-call-failed", "error": str(timer_exc) or type(timer_exc).__name__, } try: if self.backup_tx: self.cleanup_remote_backup() except Exception as cleanup_exc: self.result["backup_cleanup_error"] = str(cleanup_exc) rollback = self.rollback_remote(error) failed_backup = self.mark_failed_backup(error) self.result.update( { "status": "failed", "finished_utc": utc_now(), "error": error, "qualification": self.qualification, "rollback": rollback, "timer_rollback": timer_rollback, "failed_backup_path": failed_backup, "git_publication_allowed": False, } ) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) return 2 def run(self) -> int: try: with exclusive_lock(self.lock_file): return self.run_locked() except (Exception, KeyboardInterrupt) as exc: error = str(exc) or type(exc).__name__ self.log(f"failure before locked execution: {error}") self.result.update({"status": "failed", "finished_utc": utc_now(), "error": error}) atomic_write_json(self.receipt_dir / "phase1-result.json", self.result) return 2 finally: self.cleanup_ephemeral_helper() def load_config_defaults(path: Path | None) -> dict[str, Any]: if path is None: return {} document = read_json(normalize_path(path)) if not isinstance(document, dict) or document.get("schema_version") != 1: raise Phase1Error(f"unsupported Phase 1 config schema: {path}") return document def build_parser(config: Mapping[str, Any] | None = None) -> argparse.ArgumentParser: cfg = dict(config or {}) parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", type=Path, help="Load defaults from a Phase 1 JSON config") parser.add_argument("--mode", choices=["qualify", "plan", "apply", "backup", "recover"], default=cfg.get("mode", "apply")) parser.add_argument("--repo-root", default=cfg.get("repo_root", str(K3S_DIR.parent.parent))) parser.add_argument("--server", default=cfg.get("server", "debian3")) parser.add_argument("--host-registry", default=cfg.get("host_registry")) parser.add_argument("--kubeconfig", default=cfg.get("kubeconfig", str(Path.home() / ".kube/config-buildboxes"))) parser.add_argument("--backup-dir", default=cfg.get("backup_dir", str(Path.home() / ".local/state/overdeck/k3s-backups/debian3"))) parser.add_argument("--age-identity", default=cfg.get("age_identity", str(Path.home() / ".config/overdeck/k3s-backup.agekey"))) parser.add_argument("--lock-file", default=cfg.get("lock_file")) parser.add_argument("--retention", type=int, default=int(cfg.get("retention", 14))) parser.add_argument("--receipt-dir", default=None) parser.add_argument("--ssh-timeout", type=int, default=int(cfg.get("ssh_timeout", 12))) parser.add_argument("--ssh-door", choices=["tailscale_ip", "tailscale_ssh", "lan"], default=cfg.get("ssh_door", "tailscale_ip")) parser.add_argument("--install-age", action=argparse.BooleanOptionalAction, default=bool(cfg.get("install_age", False))) parser.add_argument("--no-timer", action="store_true") parser.add_argument("--install-root", default=cfg.get("install_root", str(Path.home() / ".local/lib/overdeck/k3s"))) parser.add_argument("--timer-config", default=cfg.get("timer_config", str(Path.home() / ".config/overdeck/k3s-backup.json"))) parser.add_argument("--enable-linger", action=argparse.BooleanOptionalAction, default=bool(cfg.get("enable_linger", True))) parser.add_argument("--non-interactive", action="store_true", help="Reserved for timer invocation; all operations are already non-interactive") parser.add_argument("--transaction", help="Unfinished server transaction ID for --mode recover") parser.add_argument( "--recovery-action", choices=["rollback", "finalize"], help="Explicit action for --mode recover; finalize only after reviewing a complete success receipt", ) return parser def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: initial = argparse.ArgumentParser(add_help=False) initial.add_argument("--config", type=Path) known, _ = initial.parse_known_args(argv) config = load_config_defaults(known.config) parser = build_parser(config) args = parser.parse_args(argv) if args.retention < 1: parser.error("--retention must be at least 1") if args.ssh_timeout < 1: parser.error("--ssh-timeout must be at least 1") if args.mode == "recover" and (not args.transaction or not args.recovery_action): parser.error("--mode recover requires --transaction and --recovery-action") if args.mode != "recover" and (args.transaction or args.recovery_action): parser.error("--transaction/--recovery-action are valid only with --mode recover") return args def main(argv: Sequence[str] | None = None) -> int: try: args = parse_args(argv) def interrupted(signum: int, _frame: Any) -> None: raise Phase1Error(f"received {signal.Signals(signum).name}; initiating rollback") for signal_name in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT): signal.signal(signal_name, interrupted) return Orchestrator(args).run() except Phase1Error as exc: print(f"phase1-control-plane: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())