#!/usr/bin/env python3 """Root-side helper for Overdeck K3s Phase 1. This program is installed on the K3s server by the workstation orchestrator. It never prints token values, kubeconfig contents, datastore credentials, or systemd environment values. Backup payloads contain the required sensitive material only inside the root-owned tar stream that the workstation immediately age-encrypts. """ from __future__ import annotations import argparse import base64 import hashlib import ipaddress import json import os import re import shutil import sqlite3 import ssl import stat import subprocess import sys import tarfile import tempfile import time import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable, Sequence from urllib.parse import urlparse STATE_ROOT = Path("/var/lib/overdeck/k3s-phase1") TRANSACTION_ROOT = STATE_ROOT / "transactions" BACKUP_ROOT = STATE_ROOT / "backups" MANAGED_CONFIG = Path("/etc/rancher/k3s/config.yaml.d/90-overdeck-control-plane.yaml") CONTRACT_FILE = Path("/etc/rancher/k3s/overdeck/control-plane.json") VERSION_LOCK_FILE = Path("/etc/rancher/k3s/overdeck/version-lock.json") SERVER_TOKEN = Path("/var/lib/rancher/k3s/server/token") AGENT_TOKEN = Path("/var/lib/rancher/k3s/server/agent-token") DATA_DIR = Path("/var/lib/rancher/k3s") DB_DIR = DATA_DIR / "server/db" SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") SENSITIVE_FLAGS = { "--token", "--agent-token", "--datastore-endpoint", "--etcd-s3-access-key", "--etcd-s3-secret-key", "--etcd-s3-session-token", } SENSITIVE_FLAG_TERMS = ( "token", "password", "secret", "credential", "access-key", "private-key", "datastore-endpoint", ) ACTIVE_TRANSACTION_STATES = {"prepared", "files-published", "converged", "finalizing", "rollback-failed"} DNS_LABEL_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$") SAFE_EXECUTABLE_DIRS = ( Path("/usr/local/sbin"), Path("/usr/local/bin"), Path("/usr/sbin"), Path("/usr/bin"), Path("/sbin"), Path("/bin"), Path("/opt/bin"), Path("/opt/k3s/bin"), ) SAFE_EXECUTABLE_PATH = os.pathsep.join(str(path) for path in SAFE_EXECUTABLE_DIRS) K3S_CANONICAL_BINARY_PATHS = tuple(path / "k3s" for path in SAFE_EXECUTABLE_DIRS) SYSTEMD_EXEC_PATH_RE = re.compile(r"(?:^|[;{ ]+)(?:path|argv\[\])=(/[^ ;}]+)") CONTROL_PLANE_SCHEMA = 2 VERSION_LOCK_SCHEMA = 2 BACKUP_MANIFEST_SCHEMA = 2 class RemoteError(RuntimeError): pass class K3sDiscoveryError(RemoteError): """K3s discovery failure with a safe, receipt-ready diagnostic.""" def __init__(self, message: str, diagnostics: dict[str, Any]): super().__init__(message) self.diagnostics = diagnostics def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def emit(value: Any) -> None: json.dump(value, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") def require_root() -> None: if os.geteuid() != 0: raise RemoteError("phase1-server must run as root") def safe_id(value: str, label: str = "identifier") -> str: if not SAFE_ID.fullmatch(value): raise RemoteError(f"invalid {label}: {value!r}") return value def run( argv: Sequence[str], *, check: bool = True, timeout: int | None = None, input_text: str | None = None, ) -> subprocess.CompletedProcess[str]: environment = os.environ.copy() # The helper runs through sudo. Hardened sudoers policies may omit # /usr/local/bin even though the official K3s installer uses it. Child # commands therefore receive a deterministic root-controlled search path. environment["PATH"] = SAFE_EXECUTABLE_PATH completed = subprocess.run( list(argv), input=input_text, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout, check=False, env=environment, ) if check and completed.returncode != 0: raise RemoteError( f"command failed ({completed.returncode}): {' '.join(argv)}\n" f"stdout:\n{completed.stdout[-3000:]}\n" f"stderr:\n{completed.stderr[-3000:]}" ) return completed def command_path(name: str, *, required: bool = True) -> str | None: path = shutil.which(name, path=SAFE_EXECUTABLE_PATH) if path: return path if required: raise RemoteError( f"required command is missing: {name}; searched approved PATH: {SAFE_EXECUTABLE_PATH}" ) return None def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def atomic_write(path: Path, payload: bytes, mode: int = 0o600) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) tmp = Path(tmp_name) try: os.fchmod(fd, mode) with os.fdopen(fd, "wb") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) os.replace(tmp, path) os.chmod(path, mode) dir_fd = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(dir_fd) finally: os.close(dir_fd) finally: tmp.unlink(missing_ok=True) def atomic_json(path: Path, value: Any, mode: int = 0o600) -> None: atomic_write(path, (json.dumps(value, indent=2, sort_keys=True) + "\n").encode(), mode) def file_metadata(path: Path, *, hash_contents: bool = True) -> dict[str, Any]: if not path.exists() and not path.is_symlink(): return {"exists": False, "path": str(path)} info = path.lstat() result: dict[str, Any] = { "exists": True, "path": str(path), "mode": f"{stat.S_IMODE(info.st_mode):04o}", "uid": info.st_uid, "gid": info.st_gid, "size": info.st_size, "type": "symlink" if path.is_symlink() else "directory" if path.is_dir() else "file", } if hash_contents and path.is_file() and not path.is_symlink(): result["sha256"] = sha256_file(path) if path.is_symlink(): result["target"] = os.readlink(path) return result def secret_file_metadata(path: Path) -> dict[str, Any]: """Return proof that a secret file exists without fingerprinting its value.""" result = file_metadata(path, hash_contents=False) if result.get("exists"): try: result["mtime_ns"] = path.lstat().st_mtime_ns except OSError: pass result["content_hash_recorded"] = False return result def _trusted_directory_chain(path: Path) -> bool: current = path while True: try: metadata = current.stat() except OSError: return False if metadata.st_uid != 0 or metadata.st_mode & 0o022: return False if current == current.parent: return True current = current.parent def validated_root_executable(candidate: Path) -> dict[str, Any]: """Validate a privileged executable while preserving its invocation path. K3s is a self-extracting multi-call binary. The systemd launcher path and the executable behind the running process are related, but they are not interchangeable command entrypoints. Callers receive both the trusted invocation path and the resolved file used for integrity measurement. """ if not candidate.is_absolute(): raise RemoteError(f"executable candidate is not absolute: {candidate}") if candidate.name != "k3s": raise RemoteError(f"executable candidate has an unexpected name: {candidate}") try: resolved = candidate.resolve(strict=True) metadata = resolved.stat() except (OSError, RuntimeError) as exc: raise RemoteError(f"cannot inspect executable candidate {candidate}: {type(exc).__name__}") from exc if not stat.S_ISREG(metadata.st_mode): raise RemoteError(f"executable candidate is not a regular file: {resolved}") if not metadata.st_mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH): raise RemoteError(f"executable candidate is not executable: {resolved}") if os.geteuid() == 0: if metadata.st_uid != 0: raise RemoteError(f"refusing non-root-owned executable while privileged: {resolved}") if metadata.st_mode & 0o022: raise RemoteError(f"refusing group/world-writable executable while privileged: {resolved}") if not _trusted_directory_chain(candidate.parent) or not _trusted_directory_chain(resolved.parent): raise RemoteError(f"refusing executable reached through a writable directory: {candidate}") return { "invocation_path": str(candidate), "resolved_path": str(resolved), "sha256": sha256_file(resolved), "mode": f"{stat.S_IMODE(metadata.st_mode):04o}", "uid": metadata.st_uid, "gid": metadata.st_gid, } def systemd_k3s_launcher_candidates() -> list[Path]: """Return only the outer K3s launcher from systemd ExecStart. ``/proc//exe`` is deliberately excluded here. On a running K3s server it can resolve to an extracted internal applet that reports the K3s version but cannot dispatch ``k3s kubectl`` or ``k3s etcd-snapshot``. """ candidates: list[Path] = [] exec_start = run( ["systemctl", "show", "k3s", "--property=ExecStart", "--value", "--no-pager"], check=False, timeout=20, ).stdout for match in SYSTEMD_EXEC_PATH_RE.finditer(exec_start): candidate = Path(match.group(1)) if candidate.name == "k3s" and candidate not in candidates: candidates.append(candidate) return candidates def runtime_k3s_candidate() -> tuple[Path | None, str | None]: main_pid = run( ["systemctl", "show", "k3s", "--property=MainPID", "--value", "--no-pager"], check=False, timeout=20, ).stdout.strip() if not main_pid.isdigit() or int(main_pid) <= 0: return None, None try: return Path(os.readlink(Path("/proc") / main_pid / "exe")), main_pid except OSError: return None, main_pid def safe_k3s_service_diagnostics() -> dict[str, Any]: """Collect non-secret unit facts suitable for a returned failure receipt.""" properties = ("LoadState", "ActiveState", "SubState", "FragmentPath") completed = run( [ "systemctl", "show", "k3s", *(f"--property={item}" for item in properties), "--no-pager", ], check=False, timeout=20, ) result: dict[str, Any] = {"returncode": completed.returncode} for line in completed.stdout.splitlines(): key, separator, value = line.partition("=") if separator and key in properties: result[key] = value or None return result def discover_k3s_launcher() -> dict[str, Any]: """Locate the trusted outer K3s command launcher. Candidate order intentionally prefers systemd's ExecStart path, then the official installer locations, then the deterministic privileged PATH. The running process executable is never used as a launcher candidate. """ candidates: list[tuple[str, Path]] = [] candidates.extend(("systemd-execstart", item) for item in systemd_k3s_launcher_candidates()) candidates.extend(("canonical", item) for item in K3S_CANONICAL_BINARY_PATHS) discovered = command_path("k3s", required=False) if discovered: candidates.append(("approved-path", Path(discovered))) attempts: list[dict[str, str]] = [] seen: set[str] = set() for source, candidate in candidates: rendered = str(candidate) if rendered in seen: continue seen.add(rendered) try: inventory = validated_root_executable(candidate) version, output = k3s_version(Path(inventory["invocation_path"])) except RemoteError as exc: attempts.append({"source": source, "path": rendered, "status": str(exc)}) continue inventory.update({"source": source, "version": version, "version_output": output}) return inventory diagnostics = { "schema_version": 1, "service": safe_k3s_service_diagnostics(), "attempts": attempts, "approved_path": SAFE_EXECUTABLE_PATH, "secret_values_recorded": False, } service = diagnostics["service"] raise K3sDiscoveryError( "trusted K3s launcher could not be located from k3s.service ExecStart, canonical installer paths, " f"or the approved root PATH (load={service.get('LoadState')!r}, " f"active={service.get('ActiveState')!r}, fragment={service.get('FragmentPath')!r})", diagnostics, ) def discover_k3s_runtime() -> dict[str, Any] | None: candidate, main_pid = runtime_k3s_candidate() if candidate is None: return {"present": False, "main_pid": main_pid, "source": "proc-mainpid"} try: inventory = validated_root_executable(candidate) except RemoteError as exc: return { "present": True, "main_pid": main_pid, "source": "proc-mainpid", "path": str(candidate), "trusted": False, "error": str(exc), } inventory.update({"present": True, "main_pid": main_pid, "source": "proc-mainpid", "trusted": True}) return inventory def k3s_executable_inventory() -> dict[str, Any]: return {"launcher": discover_k3s_launcher(), "runtime": discover_k3s_runtime()} def k3s_launcher() -> Path: return Path(discover_k3s_launcher()["invocation_path"]) def k3s_version(launcher: Path) -> tuple[str, str]: output = run([str(launcher), "--version"], timeout=20).stdout.strip() match = re.search(r"\bk3s version (v[^\s]+)", output) if not match: raise RemoteError(f"cannot parse K3s version from launcher {launcher}: {output!r}") return match.group(1), output def service_state() -> str: return run(["systemctl", "is-active", "k3s"], check=False, timeout=20).stdout.strip() or "unknown" def classify_cli_failure(stdout: str, stderr: str) -> str: rendered = (stdout + "\n" + stderr).lower() if 'unknown command "kubectl"' in rendered: return "launcher-dispatch-invalid" if "connection refused" in rendered or "unable to connect" in rendered: return "api-unreachable" if "forbidden" in rendered or "unauthorized" in rendered: return "api-authentication-failed" return "cli-probe-failed" def wait_ready(timeout_sec: int = 120, launcher: Path | None = None) -> dict[str, Any]: deadline = time.monotonic() + timeout_sec last: dict[str, Any] = { "ready": False, "service": service_state(), "classification": "not-probed", "detail": "", "launcher_role": "outer-command-launcher", } launcher = launcher or k3s_launcher() while time.monotonic() < deadline: active = service_state() if active == "active": probe = run( [str(launcher), "kubectl", "--request-timeout=15s", "get", "--raw=/readyz"], check=False, timeout=25, ) detail = (probe.stdout + probe.stderr).strip() if probe.returncode == 0 and "ok" in probe.stdout.lower(): return { "ready": True, "service": active, "classification": "ready", "detail": probe.stdout.strip(), "launcher_role": "outer-command-launcher", "launcher_path": str(launcher), } last = { "ready": False, "service": active, "classification": classify_cli_failure(probe.stdout, probe.stderr), "detail": detail[-2000:], "returncode": probe.returncode, "launcher_role": "outer-command-launcher", "launcher_path": str(launcher), } else: last = { "ready": False, "service": active, "classification": "service-not-active", "detail": f"service={active}", "launcher_role": "outer-command-launcher", "launcher_path": str(launcher), } if timeout_sec <= 5: break time.sleep(2) last["service"] = service_state() return last def public_cacerts() -> bytes: context = ssl.create_default_context() context.check_hostname = False context.verify_mode = ssl.CERT_NONE request = urllib.request.Request("https://127.0.0.1:6443/cacerts", headers={"User-Agent": "overdeck-k3s-phase1"}) try: with urllib.request.urlopen(request, context=context, timeout=10) as response: return response.read() except Exception as exc: # urllib exposes several environment-specific subclasses raise RemoteError(f"cannot fetch local K3s CA bundle: {exc}") from exc def tailscale_facts() -> dict[str, Any]: facts: dict[str, Any] = {"interface_present": Path("/sys/class/net/tailscale0").exists()} tailscale = command_path("tailscale", required=False) if not tailscale: facts.update({"command_present": False, "ipv4": None, "dns_name": None}) return facts facts["command_present"] = True ip_result = run([tailscale, "ip", "-4"], check=False, timeout=15) ips = [line.strip() for line in ip_result.stdout.splitlines() if line.strip()] facts["ipv4"] = ips[0] if len(ips) == 1 else None facts["ipv4_candidates"] = ips status = run([tailscale, "status", "--json"], check=False, timeout=15) dns_name = None if status.returncode == 0: try: payload = json.loads(status.stdout) self_doc = payload.get("Self") or {} raw = self_doc.get("DNSName") if isinstance(raw, str) and raw: dns_name = raw.rstrip(".") except json.JSONDecodeError: pass facts["dns_name"] = dns_name return facts def redacted_process_argv() -> list[str]: pid_raw = run(["systemctl", "show", "k3s", "--property=MainPID", "--value"], check=False, timeout=20).stdout.strip() try: pid = int(pid_raw) except ValueError: return [] if pid <= 0: return [] path = Path(f"/proc/{pid}/cmdline") try: argv = [item.decode(errors="replace") for item in path.read_bytes().split(b"\0") if item] except OSError: return [] redacted: list[str] = [] hide_next = False for arg in argv: if hide_next: redacted.append("") hide_next = False continue flag_name = arg.split("=", 1)[0] sensitive = flag_name in SENSITIVE_FLAGS or any(term in flag_name.lower() for term in SENSITIVE_FLAG_TERMS) if sensitive and "=" not in arg: redacted.append(arg) hide_next = True continue if sensitive and "=" in arg: redacted.append(flag_name + "=") else: redacted.append(arg) return redacted def environment_key_inventory() -> list[str]: keys: set[str] = set() for path in [Path("/etc/systemd/system/k3s.service.env"), Path("/etc/default/k3s")]: if not path.is_file(): continue try: lines = path.read_text(errors="replace").splitlines() except OSError: continue for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key = stripped.split("=", 1)[0].strip().removeprefix("export ").strip() if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key): keys.add(key) return sorted(keys) def detect_datastore(argv: Iterable[str], env_keys: Iterable[str]) -> str: argv_list = list(argv) if "K3S_DATASTORE_ENDPOINT" in set(env_keys): return "external" for index, arg in enumerate(argv_list): if arg == "--datastore-endpoint" or arg.startswith("--datastore-endpoint="): return "external" if arg == "--cluster-init" or arg.startswith("--cluster-init="): # Existing etcd on disk is the authoritative signal below; this flag may # remain in the unit after initialization. pass if (DB_DIR / "etcd").exists(): return "etcd" if (DB_DIR / "state.db").exists(): return "sqlite" return "unknown" def node_inventory_probe(launcher: Path) -> dict[str, Any]: result = run( [str(launcher), "kubectl", "--request-timeout=20s", "get", "nodes", "-o", "json"], check=False, timeout=35, ) if result.returncode != 0: return { "ok": False, "classification": classify_cli_failure(result.stdout, result.stderr), "returncode": result.returncode, "detail": (result.stdout + result.stderr).strip()[-2000:], "nodes": [], } try: payload = json.loads(result.stdout) except json.JSONDecodeError as exc: return { "ok": False, "classification": "invalid-json", "detail": f"{type(exc).__name__}: {exc}", "nodes": [], } items = payload.get("items") if not isinstance(items, list): return {"ok": False, "classification": "invalid-node-list", "detail": "items is not a list", "nodes": []} nodes: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): return {"ok": False, "classification": "invalid-node-item", "detail": "node item is not an object", "nodes": []} metadata = item.get("metadata") or {} status_doc = item.get("status") or {} addresses = { entry.get("type"): entry.get("address") for entry in status_doc.get("addresses", []) if isinstance(entry, dict) } conditions = { entry.get("type"): entry.get("status") for entry in status_doc.get("conditions", []) if isinstance(entry, dict) } nodes.append( { "name": metadata.get("name"), "uid": metadata.get("uid"), "internal_ip": addresses.get("InternalIP"), "hostname": addresses.get("Hostname"), "ready": conditions.get("Ready"), "labels": sorted((metadata.get("labels") or {}).keys()), } ) ready_count = sum(1 for node in nodes if node.get("ready") == "True") if not nodes: return { "ok": False, "classification": "empty-node-inventory", "detail": "Kubernetes returned a valid but empty NodeList", "nodes": [], "ready_count": 0, } return { "ok": True, "classification": "ok", "nodes": nodes, "node_count": len(nodes), "ready_count": ready_count, } def node_inventory(launcher: Path) -> list[dict[str, Any]]: probe = node_inventory_probe(launcher) if not probe.get("ok"): raise RemoteError( "Kubernetes Node inventory probe failed: " f"classification={probe.get('classification')} detail={probe.get('detail')}" ) return list(probe["nodes"]) def serving_certificate_inventory() -> dict[str, Any]: candidates = [ DATA_DIR / "server/tls/serving-kube-apiserver.crt", DATA_DIR / "server/tls/serving-kube-apiserver.crt.tmp", ] certificate = next((path for path in candidates if path.is_file() and not path.is_symlink()), None) if certificate is None: return {"present": False, "path": None, "sans": []} result: dict[str, Any] = { "present": True, "path": str(certificate), "sha256": sha256_file(certificate), "sans": [], } try: decoded = ssl._ssl._test_decode_cert(str(certificate)) # type: ignore[attr-defined] sans: list[str] = [] for kind, value in decoded.get("subjectAltName", []): if kind in {"DNS", "IP Address"} and isinstance(value, str) and value not in sans: sans.append(value.rstrip(".")) result["sans"] = sorted(sans) result["not_after"] = decoded.get("notAfter") except Exception as exc: result["decode_error"] = f"{type(exc).__name__}: {exc}" return result def active_control_plane_transactions() -> list[dict[str, Any]]: active: list[dict[str, Any]] = [] if not TRANSACTION_ROOT.is_dir(): return active for tx_dir in sorted(TRANSACTION_ROOT.iterdir()): if not tx_dir.is_dir() or not SAFE_ID.fullmatch(tx_dir.name): continue metadata = tx_dir / "transaction.json" try: document = json.loads(metadata.read_text()) except (OSError, json.JSONDecodeError): active.append({"transaction_id": tx_dir.name, "status": "invalid-metadata"}) continue status = str(document.get("status") or "unknown") if status in ACTIVE_TRANSACTION_STATES or status not in {"finalized", "rolled-back"}: active.append( { "transaction_id": tx_dir.name, "status": status, "created_utc": document.get("created_utc"), "updated_utc": document.get("updated_utc"), } ) return active def classify_server_token() -> dict[str, Any]: metadata = secret_file_metadata(SERVER_TOKEN) if not metadata.get("exists"): metadata.update({"supported": False, "classification": "missing"}) return metadata if SERVER_TOKEN.is_symlink() or not SERVER_TOKEN.is_file(): metadata.update({"supported": False, "classification": "unsafe-non-regular"}) return metadata info = SERVER_TOKEN.stat() if info.st_uid != 0 or info.st_mode & 0o077: metadata.update({"supported": False, "classification": "unsafe-owner-or-mode"}) return metadata metadata.update({"supported": True, "classification": "regular"}) return metadata def classify_agent_token() -> dict[str, Any]: metadata = secret_file_metadata(AGENT_TOKEN) if not metadata.get("exists"): metadata.update({"supported": True, "classification": "absent", "restore": "absent"}) return metadata if AGENT_TOKEN.is_symlink(): raw_target = os.readlink(AGENT_TOKEN) try: resolved = AGENT_TOKEN.resolve(strict=True) canonical_server = SERVER_TOKEN.resolve(strict=True) except (OSError, RuntimeError) as exc: metadata.update( { "supported": False, "classification": "broken-or-cyclic-symlink", "error": f"{type(exc).__name__}: {exc}", } ) return metadata if resolved != canonical_server: metadata.update( { "supported": False, "classification": "symlink-target-not-server-token", "resolved_target": str(resolved), } ) return metadata metadata.update( { "supported": True, "classification": "symlink-to-server-token", "target": raw_target, "resolved_target": str(resolved), "restore": "symlink-to-token", } ) return metadata if not AGENT_TOKEN.is_file(): metadata.update({"supported": False, "classification": "unsafe-non-regular"}) return metadata info = AGENT_TOKEN.stat() if info.st_uid != 0 or info.st_mode & 0o077: metadata.update({"supported": False, "classification": "unsafe-owner-or-mode"}) return metadata metadata.update({"supported": True, "classification": "regular", "restore": "regular-file"}) return metadata def inspect_state() -> dict[str, Any]: executable = k3s_executable_inventory() launcher_inventory = executable["launcher"] launcher = Path(launcher_inventory["invocation_path"]) version = str(launcher_inventory["version"]) version_output = str(launcher_inventory["version_output"]) argv = redacted_process_argv() env_keys = environment_key_inventory() datastore = detect_datastore(argv, env_keys) config_paths = [Path("/etc/rancher/k3s/config.yaml")] dropin = Path("/etc/rancher/k3s/config.yaml.d") if dropin.is_dir(): config_paths.extend(sorted(dropin.glob("*.yaml"))) unit_fragment = run( ["systemctl", "show", "k3s", "--property=FragmentPath", "--value"], check=False, timeout=20, ).stdout.strip() unit_metadata = ( file_metadata(Path(unit_fragment)) if unit_fragment and (Path(unit_fragment).exists() or Path(unit_fragment).is_symlink()) else {"exists": False, "path": unit_fragment or None} ) service_environment_files = [ secret_file_metadata(path) for path in [Path("/etc/systemd/system/k3s.service.env"), Path("/etc/default/k3s")] if path.exists() or path.is_symlink() ] ready = wait_ready(30, launcher) node_probe = node_inventory_probe(launcher) if ready.get("ready") else { "ok": False, "classification": "blocked-by-readiness", "detail": ready.get("detail"), "nodes": [], } cacerts = b"" cacerts_probe: dict[str, Any] if ready.get("ready"): try: cacerts = public_cacerts() cacerts_probe = {"ok": True, "size": len(cacerts), "sha256": sha256_bytes(cacerts)} except RemoteError as exc: cacerts_probe = {"ok": False, "error": str(exc)} else: cacerts_probe = {"ok": False, "error": "blocked-by-readiness"} runtime = executable.get("runtime") return { "schema_version": 2, "generated_utc": utc_now(), "hostname": os.uname().nodename, "machine_id": Path("/etc/machine-id").read_text().strip() if Path("/etc/machine-id").is_file() else None, "service": {"state": service_state(), "ready": ready}, "k3s": { "version": version, "version_output": version_output, "launcher": launcher_inventory, "runtime": runtime, "argv_redacted": argv, "environment_keys": env_keys, "unit_fragment": unit_fragment or None, "unit_fragment_metadata": unit_metadata, "service_environment_files": service_environment_files, }, "datastore": {"type": datastore, "db_dir": str(DB_DIR)}, "server_token": classify_server_token(), "agent_token": classify_agent_token(), "config_files": [file_metadata(path) for path in config_paths if path.exists() or path.is_symlink()], "managed_files": [file_metadata(MANAGED_CONFIG), file_metadata(CONTRACT_FILE), file_metadata(VERSION_LOCK_FILE)], "tailscale": tailscale_facts(), "nodes": list(node_probe.get("nodes") or []), "node_probe": node_probe, "serving_certificate": serving_certificate_inventory(), "active_transactions": active_control_plane_transactions(), "cacerts_probe": cacerts_probe, "cacerts_sha256": sha256_bytes(cacerts) if cacerts else None, "cacerts_size": len(cacerts), } def yaml_quote(value: str) -> str: return json.dumps(value, ensure_ascii=False) def normalize_endpoint(endpoint: str) -> str: parsed = urlparse(endpoint) if parsed.scheme != "https" or not parsed.hostname: raise RemoteError("API endpoint must be an absolute https URL") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise RemoteError("API endpoint must not contain credentials, query, or fragment") if parsed.path not in {"", "/"}: raise RemoteError("API endpoint must not contain a path") try: port = parsed.port or 443 except ValueError as exc: raise RemoteError("API endpoint contains an invalid port") from exc if not (1 <= port <= 65535): raise RemoteError("API endpoint port is out of range") host = normalize_san(parsed.hostname) rendered = f"[{host}]" if ":" in host else host return f"https://{rendered}:{port}" def normalize_san(raw: str) -> str: item = raw.strip().rstrip(".") if not item or any(char in item for char in "\r\n\0"): raise RemoteError("TLS SAN is empty or contains a control character") try: return str(ipaddress.ip_address(item)) except ValueError: pass if len(item) > 253: raise RemoteError(f"TLS SAN DNS name is too long: {item!r}") labels = item.split(".") if any(not DNS_LABEL_RE.fullmatch(label) for label in labels): raise RemoteError(f"TLS SAN is not a valid IP address or DNS name: {item!r}") return item.lower() def normalized_sans(sans: Sequence[str]) -> list[str]: unique: list[str] = [] seen: set[str] = set() for raw in sans: item = normalize_san(raw) if item in seen: continue seen.add(item) unique.append(item) if not unique: raise RemoteError("at least one TLS SAN is required") return unique def managed_config_content(sans: Sequence[str]) -> bytes: unique = normalized_sans(sans) lines = [ "# Managed by Overdeck tools/k3s. Do not hand-edit.", "# CLI flags continue to take precedence; Phase 1 does not remove installer arguments.", 'write-kubeconfig-mode: "0600"', "tls-san+:", ] lines.extend(f" - {yaml_quote(item)}" for item in unique) return ("\n".join(lines) + "\n").encode() def external_configuration_baseline(state: dict[str, Any]) -> dict[str, Any]: """Describe the pre-existing K3s configuration surface without secret values. Phase 1 owns one drop-in and two JSON contracts. Existing installer flags remain in place to avoid translating unknown or secret-bearing arguments during a live control-plane migration. Their exact unit/config files are placed in the encrypted backup; this non-secret baseline makes subsequent runs fail closed when those external inputs drift. """ excluded = {str(MANAGED_CONFIG), str(CONTRACT_FILE), str(VERSION_LOCK_FILE)} config_files = [ item for item in state.get("config_files", []) if isinstance(item, dict) and item.get("path") not in excluded ] return { "external_config_files": sorted(config_files, key=lambda item: str(item.get("path"))), "systemd_unit": state.get("k3s", {}).get("unit_fragment_metadata"), "service_environment_files": state.get("k3s", {}).get("service_environment_files", []), "process_argv_redacted": state.get("k3s", {}).get("argv_redacted", []), "environment_keys": state.get("k3s", {}).get("environment_keys", []), "migration_policy": "preserve-external-installer-inputs-and-drift-lock", "encrypted_backup_contains_exact_external_inputs": True, } def read_canonical_json(path: Path, *, allowed_schemas: set[int] | None = None) -> dict[str, Any] | None: if not path.exists() and not path.is_symlink(): return None if path.is_symlink() or not path.is_file(): raise RemoteError(f"canonical managed path is not a regular file: {path}") try: value = json.loads(path.read_text()) except json.JSONDecodeError as exc: raise RemoteError(f"canonical managed JSON is invalid: {path}: {exc}") from exc if not isinstance(value, dict): raise RemoteError(f"canonical managed JSON must be an object: {path}") schemas = allowed_schemas or {1, 2} if value.get("schema_version") not in schemas: raise RemoteError(f"canonical managed JSON has an unsupported schema: {path}") return value def planned_documents(endpoint: str, sans: Sequence[str]) -> tuple[bytes, bytes, bytes, dict[str, Any]]: state = inspect_state() if state.get("active_transactions"): active = ", ".join( f"{item.get('transaction_id')}:{item.get('status')}" for item in state["active_transactions"] ) raise RemoteError(f"unfinished control-plane transaction requires recovery before proceeding: {active}") readiness = state["service"]["ready"] if state["service"]["state"] != "active" or not readiness.get("ready"): classification = readiness.get("classification") raise RemoteError( "K3s launcher readiness qualification failed: " f"classification={classification!r} detail={readiness.get('detail')!r}" ) node_probe = state.get("node_probe") or {} if not node_probe.get("ok"): raise RemoteError( "Kubernetes Node inventory qualification failed: " f"classification={node_probe.get('classification')!r} detail={node_probe.get('detail')!r}" ) if int(node_probe.get("ready_count") or 0) < 1: raise RemoteError("Kubernetes Node inventory contains no Ready nodes") if state["datastore"]["type"] not in {"sqlite", "etcd"}: raise RemoteError(f"unsupported datastore for automated Phase 1 backup: {state['datastore']['type']}") if not state["server_token"].get("supported"): raise RemoteError(f"server token is missing or unsafe: {state['server_token'].get('classification')}") if not state["agent_token"].get("supported"): raise RemoteError(f"agent token layout is unsupported: {state['agent_token'].get('classification')}") tailscale = state["tailscale"] if not tailscale.get("interface_present") or not tailscale.get("ipv4"): raise RemoteError("tailscale0 and exactly one Tailscale IPv4 address are required") endpoint = normalize_endpoint(endpoint) canonical_sans = normalized_sans(sans) endpoint_host = normalize_san(urlparse(endpoint).hostname or "") if endpoint_host not in canonical_sans: raise RemoteError("API endpoint hostname/IP must be included in the managed TLS SAN list") existing_contract = read_canonical_json(CONTRACT_FILE, allowed_schemas={1, CONTROL_PLANE_SCHEMA}) configuration_baseline = external_configuration_baseline(state) if existing_contract is not None: if existing_contract.get("managed_by") != "overdeck-k3s-phase1": raise RemoteError(f"refusing to replace an unmanaged canonical contract: {CONTRACT_FILE}") prior_server = existing_contract.get("server") or {} if prior_server.get("machine_id") not in {None, state["machine_id"]}: raise RemoteError("canonical contract machine identity does not match this server") if prior_server.get("tailscale_ipv4") not in {None, tailscale.get("ipv4")}: raise RemoteError("canonical contract Tailscale identity does not match this server") prior_baseline = existing_contract.get("configuration_baseline") if prior_baseline != configuration_baseline: raise RemoteError( "external K3s configuration drift differs from the canonical baseline; " "review the unit/config change and use an explicit migration or upgrade workflow" ) config = managed_config_content(canonical_sans) contract = { "schema_version": CONTROL_PLANE_SCHEMA, "managed_by": "overdeck-k3s-phase1", "server": { "hostname": state["hostname"], "machine_id": state["machine_id"], "tailscale_ipv4": tailscale.get("ipv4"), "tailscale_dns_name": tailscale.get("dns_name"), "network_interface": "tailscale0", }, "api": { "endpoint": endpoint, "tls_sans": canonical_sans, "cacerts_sha256": state.get("cacerts_sha256"), }, "network_enforcement": { "server_node_ip": "observed-not-changed-in-phase1", "server_flannel_interface": "observed-not-changed-in-phase1", "new_agent_node_ip": "tailscale-ipv4", "new_agent_flannel_interface": "tailscale0", }, "configuration_baseline": configuration_baseline, "qualification": { "launcher_role_separated_from_runtime": True, "node_inventory_required": True, "prechange_backup_required_before_publish": True, "agent_token_layout": state["agent_token"].get("classification"), }, } version = state["k3s"]["version"] launcher = state["k3s"]["launcher"] runtime = state["k3s"].get("runtime") or {"present": False} lock = { "schema_version": VERSION_LOCK_SCHEMA, "managed_by": "overdeck-k3s-phase1", "version": version, "launcher": { "invocation_path": launcher["invocation_path"], "resolved_path": launcher["resolved_path"], "sha256": launcher["sha256"], "release_url": f"https://github.com/k3s-io/k3s/releases/download/{version}/k3s", }, "runtime": { "present": bool(runtime.get("present")), "invocation_path": runtime.get("invocation_path"), "resolved_path": runtime.get("resolved_path"), "sha256": runtime.get("sha256"), "source": runtime.get("source"), }, "install_script_url": "https://get.k3s.io", "upgrade_policy": "explicit-version-and-launcher-sha256-only", } existing_lock = read_canonical_json(VERSION_LOCK_FILE, allowed_schemas={1, VERSION_LOCK_SCHEMA}) if existing_lock is not None: if existing_lock.get("managed_by") != "overdeck-k3s-phase1": raise RemoteError(f"refusing to replace an unmanaged version lock: {VERSION_LOCK_FILE}") if existing_lock.get("schema_version") == 1: raise RemoteError( "legacy Phase 1 version-lock schema conflates launcher and runtime binaries; " "review and remove that unlanded lock before rerunning qualification" ) prior_launcher = existing_lock.get("launcher") or {} if existing_lock.get("version") != version or prior_launcher.get("sha256") != launcher["sha256"]: raise RemoteError( "installed K3s version/launcher differs from the canonical lock; " "use the explicit upgrade/rollback workflow instead of adopting drift" ) contract_bytes = (json.dumps(contract, indent=2, sort_keys=True) + "\n").encode() lock_bytes = (json.dumps(lock, indent=2, sort_keys=True) + "\n").encode() return config, contract_bytes, lock_bytes, state def content_status(path: Path, expected: bytes) -> dict[str, Any]: if path.is_symlink(): raise RemoteError(f"managed path must not be a symlink: {path}") if path.exists() and not path.is_file(): raise RemoteError(f"managed path must be a regular file: {path}") actual = path.read_bytes() if path.is_file() else None return { "path": str(path), "exists": actual is not None, "changed": actual != expected, "current_sha256": sha256_bytes(actual) if actual is not None else None, "desired_sha256": sha256_bytes(expected), } def create_transaction(tx: str, paths: Sequence[Path], before: dict[str, Any]) -> Path: safe_id(tx, "transaction ID") active = active_control_plane_transactions() if active: rendered = ", ".join(f"{item.get('transaction_id')}:{item.get('status')}" for item in active) raise RemoteError(f"another control-plane transaction is unfinished: {rendered}") TRANSACTION_ROOT.mkdir(parents=True, exist_ok=True) os.chmod(TRANSACTION_ROOT, 0o700) tx_dir = TRANSACTION_ROOT / tx if tx_dir.exists(): raise RemoteError(f"transaction already exists: {tx}") temporary = TRANSACTION_ROOT / f".{tx}.tmp-{os.getpid()}-{os.urandom(3).hex()}" rollback_dir = temporary / "rollback" rollback_dir.mkdir(parents=True, mode=0o700) try: records: list[dict[str, Any]] = [] for index, path in enumerate(paths): record = file_metadata(path) record["index"] = index if path.exists() or path.is_symlink(): backup = rollback_dir / str(index) if path.is_symlink(): record["backup_type"] = "symlink" record["backup_target"] = os.readlink(path) elif path.is_file(): shutil.copy2(path, backup, follow_symlinks=False) record["backup_type"] = "file" else: raise RemoteError(f"managed path is not a regular file: {path}") else: record["backup_type"] = "absent" records.append(record) document = { "schema_version": 1, "transaction_id": tx, "created_utc": utc_now(), "status": "prepared", "managed_paths": records, "before": before, } atomic_json(temporary / "transaction.json", document) os.rename(temporary, tx_dir) directory_fd = os.open(TRANSACTION_ROOT, os.O_DIRECTORY) try: os.fsync(directory_fd) finally: os.close(directory_fd) except Exception: shutil.rmtree(temporary, ignore_errors=True) raise return tx_dir def load_transaction(tx: str) -> tuple[Path, dict[str, Any]]: safe_id(tx, "transaction ID") tx_dir = TRANSACTION_ROOT / tx path = tx_dir / "transaction.json" if not path.is_file(): raise RemoteError(f"transaction does not exist: {tx}") try: document = json.loads(path.read_text()) except json.JSONDecodeError as exc: raise RemoteError(f"transaction metadata is invalid: {tx}") from exc return tx_dir, document def update_transaction(tx_dir: Path, document: dict[str, Any], status: str, **extra: Any) -> None: document = dict(document) document["status"] = status document["updated_utc"] = utc_now() document.update(extra) atomic_json(tx_dir / "transaction.json", document) def restore_transaction_files(tx_dir: Path, document: dict[str, Any]) -> None: rollback_dir = tx_dir / "rollback" for record in document.get("managed_paths", []): path = Path(record["path"]) backup_type = record.get("backup_type") if path.exists() or path.is_symlink(): if path.is_dir() and not path.is_symlink(): raise RemoteError(f"refusing to replace directory during rollback: {path}") path.unlink() if backup_type == "absent": continue path.parent.mkdir(parents=True, exist_ok=True) if backup_type == "symlink": os.symlink(record["backup_target"], path) elif backup_type == "file": source = rollback_dir / str(record["index"]) shutil.copy2(source, path, follow_symlinks=False) os.chown(path, int(record.get("uid", 0)), int(record.get("gid", 0))) os.chmod(path, int(str(record.get("mode", "0600")), 8)) else: raise RemoteError(f"unknown rollback record type: {backup_type}") def rollback_transaction(tx: str, reason: str) -> dict[str, Any]: tx_dir, document = load_transaction(tx) status_before = str(document.get("status") or "unknown") if status_before == "finalized": raise RemoteError(f"transaction is finalized and cannot be rolled back: {tx}") if status_before == "finalizing": raise RemoteError( f"transaction finalization has started and cannot be rolled back safely: {tx}; " "retry finalization instead" ) if status_before == "rolled-back": return { "transaction_id": tx, "status": "rolled-back", "idempotent": True, "readiness": document.get("rollback_readiness"), } restore_transaction_files(tx_dir, document) run(["systemctl", "restart", "k3s"], check=False, timeout=60) readiness = wait_ready(120) status = "rolled-back" if readiness.get("ready") else "rollback-failed" update_transaction(tx_dir, document, status, rollback_reason=reason, rollback_readiness=readiness) return {"transaction_id": tx, "status": status, "readiness": readiness} def require_certificate_sans(state: dict[str, Any], sans: Sequence[str]) -> None: certificate = state.get("serving_certificate") or {} actual_sans = {normalize_san(item) for item in certificate.get("sans", []) if isinstance(item, str)} required_sans = set(normalized_sans(sans)) missing_sans = sorted(required_sans - actual_sans) if not certificate.get("present") or certificate.get("decode_error") or missing_sans: raise RemoteError( "K3s serving certificate does not prove every managed TLS SAN: " f"missing={missing_sans} certificate={certificate}" ) def converge(tx: str, endpoint: str, sans: Sequence[str]) -> dict[str, Any]: config, contract, lock, before = planned_documents(endpoint, sans) desired = [(MANAGED_CONFIG, config), (CONTRACT_FILE, contract), (VERSION_LOCK_FILE, lock)] statuses = [content_status(path, payload) for path, payload in desired] changed = any(item["changed"] for item in statuses) if not changed: after = inspect_state() require_certificate_sans(after, sans) return { "schema_version": 1, "transaction_id": None, "changed": False, "status": "already-converged", "files": statuses, "before": before, "after": after, } tx_dir = create_transaction(tx, [path for path, _ in desired], before) _, document = load_transaction(tx) launcher_sha = before["k3s"]["launcher"]["sha256"] runtime_sha = (before["k3s"].get("runtime") or {}).get("sha256") config_changed = next(item["changed"] for item in statuses if item["path"] == str(MANAGED_CONFIG)) try: for path, payload in desired: if not next(item["changed"] for item in statuses if item["path"] == str(path)): continue atomic_write(path, payload, 0o600) os.chown(path, 0, 0) update_transaction(tx_dir, document, "files-published") if config_changed: run(["systemctl", "restart", "k3s"], timeout=90) readiness = wait_ready(150) if not readiness.get("ready"): raise RemoteError(f"K3s did not return Ready after config convergence: {readiness}") after = inspect_state() if after["k3s"]["launcher"]["sha256"] != launcher_sha: raise RemoteError("K3s launcher changed during Phase 1; version upgrades are forbidden") after_runtime_sha = (after["k3s"].get("runtime") or {}).get("sha256") if runtime_sha is not None and after_runtime_sha != runtime_sha: raise RemoteError("running K3s runtime executable changed during Phase 1") require_certificate_sans(after, sans) update_transaction(tx_dir, document, "converged", readiness=readiness, after=after) return { "schema_version": 1, "transaction_id": tx, "changed": True, "status": "converged", "files": [content_status(path, payload) for path, payload in desired], "before": before, "after": after, "readiness": readiness, "service_restarted": bool(config_changed), } except Exception as exc: rollback = rollback_transaction(tx, f"converge failure: {exc}") raise RemoteError(f"convergence failed and rollback status is {rollback['status']}: {exc}") from exc def copy_regular_tree(source: Path, destination: Path, *, exclude_names: set[str] | None = None) -> None: excluded = exclude_names or set() if source.is_symlink() or not source.is_dir(): raise RemoteError(f"required directory is missing or unsafe: {source}") for root, dirs, files in os.walk(source, followlinks=False): root_path = Path(root) relative = root_path.relative_to(source) target_root = destination / relative target_root.mkdir(parents=True, exist_ok=True) os.chmod(target_root, 0o700) safe_dirs: list[str] = [] for name in dirs: child = root_path / name if child.is_symlink(): raise RemoteError(f"backup source contains a symlinked directory: {child}") if not child.is_dir(): raise RemoteError(f"backup source contains a non-directory entry: {child}") safe_dirs.append(name) dirs[:] = safe_dirs for name in files: if name in excluded: continue src = root_path / name if src.is_symlink() or not src.is_file(): raise RemoteError(f"backup source contains an unsafe file: {src}") dst = target_root / name shutil.copy2(src, dst, follow_symlinks=False) def sqlite_backup(destination: Path) -> dict[str, Any]: source_db = DB_DIR / "state.db" if not source_db.is_file(): raise RemoteError(f"SQLite state database is missing: {source_db}") destination.mkdir(parents=True, exist_ok=True) copy_regular_tree(DB_DIR, destination, exclude_names={"state.db", "state.db-wal", "state.db-shm"}) target_db = destination / "state.db" source_uri = f"file:{source_db}?mode=ro" source = sqlite3.connect(source_uri, uri=True, timeout=30) target = sqlite3.connect(target_db) try: source.backup(target, pages=1024, sleep=0.05) target.execute("PRAGMA wal_checkpoint(TRUNCATE)") result = target.execute("PRAGMA integrity_check").fetchone() verdict = result[0] if result else "missing-result" if verdict != "ok": raise RemoteError(f"SQLite backup integrity_check failed: {verdict}") finally: target.close() source.close() os.chmod(target_db, 0o600) return {"type": "sqlite", "source": str(source_db), "integrity_check": "ok"} def etcd_backup(destination: Path, tx: str) -> tuple[dict[str, Any], Path]: binary = k3s_launcher() name = f"overdeck-{safe_id(tx)}" before = {path.resolve() for path in DATA_DIR.rglob(f"{name}*") if path.is_file()} result = run([str(binary), "etcd-snapshot", "save", "--name", name, "--snapshot-compress"], timeout=300) candidates = [path for path in DATA_DIR.rglob(f"{name}*") if path.is_file() and path.resolve() not in before] if not candidates: candidates = [path for path in DATA_DIR.rglob(f"{name}*") if path.is_file()] if not candidates: raise RemoteError(f"K3s reported an etcd snapshot but no file matching {name!r} was found") snapshot = max(candidates, key=lambda path: path.stat().st_mtime_ns) destination.mkdir(parents=True, exist_ok=True) target = destination / snapshot.name shutil.copy2(snapshot, target) os.chmod(target, 0o600) listing = run([str(binary), "etcd-snapshot", "ls"], check=False, timeout=60) if listing.returncode != 0: target.unlink(missing_ok=True) raise RemoteError(f"K3s could not list the newly-created etcd snapshot: {listing.stderr[-2000:]}") listed = name in (listing.stdout + listing.stderr) or snapshot.name in (listing.stdout + listing.stderr) if not listed: target.unlink(missing_ok=True) raise RemoteError("K3s snapshot listing did not contain the newly-created etcd snapshot") return ( { "type": "etcd", "snapshot_name": snapshot.name, "snapshot_size": target.stat().st_size, "snapshot_sha256": sha256_file(target), "k3s_snapshot_list_verified": True, }, snapshot, ) def copy_if_regular(source: Path, destination: Path) -> None: if not source.exists() and not source.is_symlink(): return if source.is_symlink() or not source.is_file(): raise RemoteError(f"backup source is not a regular non-symlink file: {source}") destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, destination, follow_symlinks=False) def stage_token_material(payload: Path) -> dict[str, Any]: server = classify_server_token() if not server.get("supported"): raise RemoteError(f"server token is unsafe: {server.get('classification')}") agent = classify_agent_token() if not agent.get("supported"): raise RemoteError(f"agent token is unsafe: {agent.get('classification')}") copy_if_regular(SERVER_TOKEN, payload / "server/token") classification = str(agent.get("classification")) if classification == "regular": copy_if_regular(AGENT_TOKEN, payload / "server/agent-token") return { "classification": "regular", "restore": "regular-file", "payload_path": "payload/server/agent-token", } if classification == "symlink-to-server-token": return { "classification": "symlink-to-server-token", "restore": "symlink-to-token", "link_target": "token", "payload_path": None, } if classification == "absent": return {"classification": "absent", "restore": "absent", "payload_path": None} raise RemoteError(f"unsupported agent token classification: {classification}") def backup_manifest( stage: Path, tx: str, purpose: str, datastore: dict[str, Any], state: dict[str, Any], agent_token: dict[str, Any], ) -> dict[str, Any]: files: list[dict[str, Any]] = [] for path in sorted((stage / "payload").rglob("*")): if not path.is_file() or path.is_symlink(): continue info = path.stat() files.append( { "path": path.relative_to(stage).as_posix(), "size": info.st_size, "sha256": sha256_file(path), "mode": f"{stat.S_IMODE(info.st_mode):04o}", "uid": info.st_uid, "gid": info.st_gid, } ) token_rel = "payload/server/token" if token_rel not in {item["path"] for item in files}: raise RemoteError("backup payload is missing the required server token") launcher = state["k3s"]["launcher"] runtime = state["k3s"].get("runtime") or {"present": False} return { "schema_version": BACKUP_MANIFEST_SCHEMA, "created_utc": utc_now(), "transaction_id": tx, "purpose": purpose, "server": { "hostname": state["hostname"], "machine_id": state["machine_id"], "tailscale_ipv4": state["tailscale"].get("ipv4"), }, "k3s": { "version": state["k3s"]["version"], "launcher": { "invocation_path": launcher.get("invocation_path"), "resolved_path": launcher.get("resolved_path"), "sha256": launcher.get("sha256"), }, "runtime": { "present": bool(runtime.get("present")), "resolved_path": runtime.get("resolved_path"), "sha256": runtime.get("sha256"), }, "cacerts_sha256": state.get("cacerts_sha256"), }, "datastore": datastore, "server_token_path": token_rel, "agent_token": agent_token, "systemd": { "unit_fragment": state.get("k3s", {}).get("unit_fragment"), "service_path": "/etc/systemd/system/k3s.service", "environment_path": "/etc/systemd/system/k3s.service.env", }, "files": files, } def cleanup_stale_backup_artifacts(*, max_age_seconds: int = 6 * 60 * 60) -> list[str]: removed: list[str] = [] if not BACKUP_ROOT.is_dir(): return removed cutoff = time.time() - max_age_seconds for path in sorted(BACKUP_ROOT.iterdir()): name = path.name tx_name = name[:-4] if name.endswith(".tar") else name if not tx_name.startswith("backup-") or not SAFE_ID.fullmatch(tx_name): continue try: stale = path.stat().st_mtime < cutoff except OSError: continue if not stale: continue if path.is_dir() and not path.is_symlink(): shutil.rmtree(path, ignore_errors=True) removed.append(name) elif path.is_file() and not path.is_symlink(): path.unlink(missing_ok=True) removed.append(name) return removed def create_backup(tx: str, purpose: str) -> dict[str, Any]: safe_id(tx, "backup transaction ID") if purpose not in {"prechange", "postchange", "scheduled"}: raise RemoteError(f"unsupported backup purpose: {purpose!r}") state = inspect_state() if not state["service"]["ready"].get("ready"): raise RemoteError("refusing backup because K3s is not ready") datastore_type = state["datastore"]["type"] if datastore_type not in {"sqlite", "etcd"}: raise RemoteError(f"unsupported datastore for automated backup: {datastore_type}") if not state["server_token"].get("supported"): raise RemoteError(f"server token is missing or unsafe: {state['server_token'].get('classification')}") if not state["agent_token"].get("supported"): raise RemoteError(f"agent token layout is unsafe: {state['agent_token'].get('classification')}") BACKUP_ROOT.mkdir(parents=True, exist_ok=True) os.chmod(BACKUP_ROOT, 0o700) stale_removed = cleanup_stale_backup_artifacts() stage = BACKUP_ROOT / tx if stage.exists(): raise RemoteError(f"backup transaction already exists: {tx}") payload = stage / "payload" stage.mkdir(parents=True, mode=0o700) generated_snapshot: Path | None = None try: if datastore_type == "sqlite": datastore = sqlite_backup(payload / "datastore/db") else: datastore, generated_snapshot = etcd_backup(payload / "datastore/snapshot", tx) agent_token = stage_token_material(payload) config_source = Path("/etc/rancher/k3s") if config_source.is_dir(): copy_regular_tree(config_source, payload / "config/etc-rancher-k3s") for source in [Path("/etc/systemd/system/k3s.service"), Path("/etc/systemd/system/k3s.service.env")]: copy_if_regular(source, payload / "systemd" / source.name) unit_fragment = state["k3s"].get("unit_fragment") if unit_fragment: source = Path(unit_fragment) if source not in {Path("/etc/systemd/system/k3s.service")}: copy_if_regular(source, payload / "systemd" / "fragment" / source.name) manifest = backup_manifest(stage, tx, purpose, datastore, state, agent_token) atomic_json(stage / "manifest.json", manifest) archive = BACKUP_ROOT / f"{tx}.tar" with tarfile.open(archive, "w", format=tarfile.PAX_FORMAT) as handle: handle.add(stage / "manifest.json", arcname="manifest.json", recursive=False) handle.add(payload, arcname="payload", recursive=True, filter=lambda item: None if (item.issym() or item.islnk() or item.isdev() or item.isfifo()) else item) os.chmod(archive, 0o600) return { "schema_version": 2, "transaction_id": tx, "purpose": purpose, "status": "staged", "archive_path": str(archive), "archive_size": archive.stat().st_size, "archive_sha256": sha256_file(archive), "manifest": { "created_utc": manifest["created_utc"], "server": manifest["server"], "k3s": manifest["k3s"], "datastore": manifest["datastore"], "file_count": len(manifest["files"]), }, "stale_artifacts_removed": stale_removed, } except Exception: shutil.rmtree(stage, ignore_errors=True) (BACKUP_ROOT / f"{tx}.tar").unlink(missing_ok=True) raise finally: # ``k3s etcd-snapshot save`` writes into K3s' local snapshot directory. # The authoritative copy for this transaction is now in the staged TAR; # remove only the uniquely-named snapshot created by this invocation so # recurring workstation backups cannot exhaust server disk space. if generated_snapshot is not None: generated_snapshot.unlink(missing_ok=True) def cleanup_backup(tx: str) -> dict[str, Any]: safe_id(tx, "backup transaction ID") stage = BACKUP_ROOT / tx archive = BACKUP_ROOT / f"{tx}.tar" shutil.rmtree(stage, ignore_errors=True) archive.unlink(missing_ok=True) return {"transaction_id": tx, "status": "cleaned"} def finalize_transaction(tx: str) -> dict[str, Any]: tx_dir, document = load_transaction(tx) status = str(document.get("status") or "unknown") if status == "finalized": return {"transaction_id": tx, "status": "finalized", "idempotent": True} if status not in {"converged", "finalizing"}: raise RemoteError(f"only a converged transaction can be finalized; {tx} is {status!r}") rollback_dir = tx_dir / "rollback" if status == "converged": update_transaction(tx_dir, document, "finalizing") _, document = load_transaction(tx) if rollback_dir.exists(): shutil.rmtree(rollback_dir) if rollback_dir.exists(): raise RemoteError(f"could not remove rollback material while finalizing transaction: {tx}") update_transaction(tx_dir, document, "finalized") return {"transaction_id": tx, "status": "finalized"} def plan(endpoint: str, sans: Sequence[str]) -> dict[str, Any]: config, contract, lock, state = planned_documents(endpoint, sans) files = [ content_status(MANAGED_CONFIG, config), content_status(CONTRACT_FILE, contract), content_status(VERSION_LOCK_FILE, lock), ] return { "schema_version": 1, "generated_utc": utc_now(), "changed": any(item["changed"] for item in files), "files": files, "desired_managed_config_b64": base64.b64encode(config).decode(), "desired_contract": json.loads(contract), "desired_version_lock": json.loads(lock), "observed": state, } def parse_sans(raw: str) -> list[str]: try: value = json.loads(raw) except json.JSONDecodeError as exc: raise RemoteError("--sans-json must be a JSON array") from exc if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise RemoteError("--sans-json must be a JSON array of strings") return value def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("inspect") plan_parser = sub.add_parser("plan") plan_parser.add_argument("--endpoint", required=True) plan_parser.add_argument("--sans-json", required=True) converge_parser = sub.add_parser("converge") converge_parser.add_argument("--transaction", required=True) converge_parser.add_argument("--endpoint", required=True) converge_parser.add_argument("--sans-json", required=True) backup_parser = sub.add_parser("backup") backup_parser.add_argument("--transaction", required=True) backup_parser.add_argument( "--purpose", choices=["prechange", "postchange", "scheduled"], required=True, ) cleanup_parser = sub.add_parser("cleanup-backup") cleanup_parser.add_argument("--transaction", required=True) rollback_parser = sub.add_parser("rollback") rollback_parser.add_argument("--transaction", required=True) rollback_parser.add_argument("--reason", default="workstation-requested rollback") finalize_parser = sub.add_parser("finalize") finalize_parser.add_argument("--transaction", required=True) return parser def main(argv: Sequence[str] | None = None) -> int: require_root() args = build_parser().parse_args(argv) if args.command == "inspect": result = inspect_state() elif args.command == "plan": result = plan(args.endpoint, parse_sans(args.sans_json)) elif args.command == "converge": result = converge(args.transaction, args.endpoint, parse_sans(args.sans_json)) elif args.command == "backup": result = create_backup(args.transaction, args.purpose) elif args.command == "cleanup-backup": result = cleanup_backup(args.transaction) elif args.command == "rollback": result = rollback_transaction(args.transaction, args.reason) elif args.command == "finalize": result = finalize_transaction(args.transaction) else: # pragma: no cover - argparse prevents this raise RemoteError(f"unsupported command: {args.command}") emit(result) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except RemoteError as exc: payload: dict[str, Any] = {"schema_version": 1, "status": "error", "error": str(exc)} diagnostics = getattr(exc, "diagnostics", None) if isinstance(diagnostics, dict): payload["diagnostics"] = diagnostics emit(payload) raise SystemExit(2) except Exception as exc: # fail loud without leaking arbitrary object reprs emit({"schema_version": 1, "status": "error", "error": f"unexpected {type(exc).__name__}: {exc}"}) raise SystemExit(3)