"""Lifecycle operations for the libvirt-backed GovOPlaN Kubernetes lab.""" from __future__ import annotations from dataclasses import asdict import getpass import hashlib import http.client import ipaddress import json import os from pathlib import Path import secrets import shlex import shutil import socket import ssl import stat import subprocess import sys import time from typing import Any, Iterable, Mapping, Sequence from urllib.parse import quote, urlsplit from urllib.request import Request, urlopen from .config import LabConfig, LabNode from .render import ( render_caddyfile, render_cloud_init, render_garage_config, render_hosts, render_k3s_config, render_meta_data, render_network_config, render_registry_config, render_secret_manifest, render_state_compose, render_state_environment, render_tls_secret_manifest, write_private, ) REPOSITORY_ROOT = Path(__file__).resolve().parents[3] DEPLOYER = REPOSITORY_ROOT / "tools" / "deployment" / "govoplan-deploy.py" _SSH_TIMEOUT_SECONDS = 600 _STATE_REMOTE_DIRECTORY = "govoplan-lab-state" _SYSTEM_LIBVIRT_URI = "qemu:///system" _SECRET_KEYS = ( "MASTER_KEY_B64", "DATABASE_URL", "GOVOPLAN_DATABASE_URL_PGTOOLS", "REDIS_URL", "FILE_STORAGE_S3_ACCESS_KEY_ID", "FILE_STORAGE_S3_SECRET_ACCESS_KEY", ) class LabOperationError(RuntimeError): """Raised when a lifecycle operation cannot complete safely.""" class CommandRunner: def __init__(self, config: LabConfig, *, verbose: bool = False) -> None: self.config = config self.verbose = verbose self.known_hosts = config.state_directory / "ssh_known_hosts" def run( self, argv: Sequence[str], *, input_bytes: bytes | None = None, capture: bool = False, check: bool = True, timeout: float | None = None, sensitive: bool = False, env: Mapping[str, str] | None = None, ) -> subprocess.CompletedProcess[bytes]: if self.verbose: rendered = "" if sensitive else shlex.join(argv) print(f"+ {rendered}", file=sys.stderr) try: return subprocess.run( list(argv), input=input_bytes, stdout=subprocess.PIPE if capture else None, stderr=subprocess.PIPE if capture else None, check=check, timeout=timeout, env=dict(env) if env is not None else None, ) except FileNotFoundError as exc: raise LabOperationError(f"required command is unavailable: {argv[0]}") from exc except subprocess.CalledProcessError as exc: stderr = (exc.stderr or b"").decode(errors="replace").strip() detail = f": {stderr}" if stderr else "" raise LabOperationError(f"command failed ({argv[0]}){detail}") from exc except subprocess.TimeoutExpired as exc: raise LabOperationError(f"command timed out ({argv[0]})") from exc def ssh_arguments(self, target: str) -> list[str]: return [ "ssh", "-i", str(self.config.ssh_private_key), "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-o", f"UserKnownHostsFile={self.known_hosts}", "-o", "StrictHostKeyChecking=accept-new", target, ] def vm( self, node: LabNode, argv: Sequence[str], *, sudo: bool = False, input_bytes: bytes | None = None, capture: bool = False, check: bool = True, timeout: float | None = None, sensitive: bool = False, ) -> subprocess.CompletedProcess[bytes]: target = f"{self.config.ssh_user}@{node.address}" command = [*(("sudo", "-n", "--") if sudo else ()), *argv] return self.run( [*self.ssh_arguments(target), shlex.join(command)], input_bytes=input_bytes, capture=capture, check=check, timeout=timeout, sensitive=sensitive, ) def hypervisor( self, node: LabNode, argv: Sequence[str], *, capture: bool = False, check: bool = True, timeout: float | None = None, ) -> subprocess.CompletedProcess[bytes]: if node.is_local: return self.run( _local_hypervisor_command(argv), capture=capture, check=check, timeout=timeout, ) command = ["sudo", "-n", "--", *argv] return self.run( [*self.ssh_arguments(node.hypervisor), shlex.join(command)], capture=capture, check=check, timeout=timeout, ) def copy_to_vm(self, node: LabNode, source: Path, destination: str) -> None: target = f"{self.config.ssh_user}@{node.address}:{destination}" self.run(["scp", *self.ssh_arguments("")[1:-1], str(source), target]) def copy_to_hypervisor(self, node: LabNode, source: Path, destination: str) -> None: if node.is_local: self.run(["install", "-m", "0600", str(source), destination]) return upload = f"/tmp/{self.config.name}-{node.name}-{source.name}.upload" target = f"{node.hypervisor}:{upload}" self.run(["scp", *self.ssh_arguments("")[1:-1], str(source), target]) self.hypervisor( node, ["install", "-m", "0600", upload, destination], ) self.hypervisor(node, ["rm", "-f", upload]) def _local_hypervisor_command(argv: Sequence[str]) -> list[str]: command = list(argv) if command and command[0] in {"virsh", "virt-install"}: return [command[0], "--connect", _SYSTEM_LIBVIRT_URI, *command[1:]] return command def doctor(config: LabConfig, *, online: bool = False, verbose: bool = False) -> int: checks: list[tuple[str, bool, str]] = [] for command in ("ssh", "scp", "ssh-keygen", "openssl", "curl"): checks.append((f"local.{command}", shutil.which(command) is not None, command)) checks.extend( [ ( "ssh.private_key", config.ssh_private_key.is_file(), str(config.ssh_private_key), ), ( "ssh.public_key", config.ssh_public_key.is_file(), str(config.ssh_public_key), ), ( "topology.workers", len(config.workers) >= 2, f"{len(config.workers)} worker(s)", ), ( "topology.state", len([node for node in config.nodes if node.role == "state"]) == 1, str(config.state_node.address), ), ( "topology.evidence", config.mode == "rehearsal" or config.evidence_capable, "evidence-capable" if config.evidence_capable else "same-host rehearsal", ), ] ) if config.ssh_private_key.exists(): mode = stat.S_IMODE(config.ssh_private_key.stat().st_mode) checks.append(("ssh.private_key_mode", mode & 0o077 == 0, oct(mode))) if online: runner = CommandRunner(config, verbose=verbose) for hypervisor, node in _representative_hypervisors(config): result = runner.hypervisor( node, [ "sh", "-c", "command -v virsh && command -v virt-install && " "command -v qemu-img && command -v cloud-localds && command -v curl", ], capture=True, check=False, timeout=30, ) checks.append( ( f"hypervisor.{hypervisor}", result.returncode == 0, (result.stderr or result.stdout or b"").decode(errors="replace").strip(), ) ) storage_exists = runner.hypervisor( node, ["test", "-d", config.vm_image_directory], capture=True, check=False, timeout=30, ) storage_writable = runner.hypervisor( node, ["test", "-w", config.vm_image_directory], capture=True, check=False, timeout=30, ) checks.append( ( f"hypervisor.{hypervisor}.vm_image_directory", storage_exists.returncode == 0 and storage_writable.returncode == 0, config.vm_image_directory, ) ) bridge = runner.hypervisor( node, ["test", "-d", f"/sys/class/net/{config.network.bridge}"], capture=True, check=False, timeout=30, ) checks.append( ( f"hypervisor.{hypervisor}.bridge", bridge.returncode == 0, config.network.bridge, ) ) for check_id, passed, detail in checks: print(f"{'OK' if passed else 'FAIL'} {check_id}: {detail}") failed = [check for check in checks if not check[1]] if not failed: print( "Lab configuration is valid for " + ("target evidence." if config.evidence_capable else "a same-host rehearsal.") ) return 1 if failed else 0 def create(config: LabConfig, *, apply: bool, verbose: bool = False) -> None: if not apply: _print_mutation_preview( "create", config, [f"create or reuse VM {node.name} on {node.hypervisor}" for node in config.nodes], ) return _prepare_local_state(config) runner = CommandRunner(config, verbose=verbose) public_key = config.ssh_public_key.read_text(encoding="utf-8").strip() if len(public_key.splitlines()) != 1 or not public_key.startswith( ("ssh-ed25519 ", "ssh-rsa ", "ecdsa-sha2-", "sk-ssh-ed25519@openssh.com ") ): raise LabOperationError("ssh_public_key is not an OpenSSH public key") for node in config.nodes: _create_node(config, runner, node, public_key) for node in config.nodes: _wait_for_ssh(config, runner, node) runner.vm( node, ["cloud-init", "status", "--wait"], sudo=True, timeout=_SSH_TIMEOUT_SECONDS, ) _write_inventory(config) print(f"Created {len(config.nodes)} VM(s) for lab {config.name}.") def _create_node( config: LabConfig, runner: CommandRunner, node: LabNode, public_key: str, ) -> None: domain = _domain_name(config, node) existing = runner.hypervisor( node, ["virsh", "dominfo", domain], capture=True, check=False, timeout=20, ) if existing.returncode == 0: _assert_domain_owned(config, runner, node) print(f"Reusing existing domain {domain} on {node.hypervisor}.") runner.hypervisor(node, ["virsh", "start", domain], check=False) return _forget_vm_host_key(config, runner, node) rendered = config.state_directory / "rendered" / node.name rendered.mkdir(parents=True, exist_ok=True, mode=0o700) write_private(rendered / "user-data", render_cloud_init(config, node, public_key)) write_private(rendered / "meta-data", render_meta_data(config, node)) write_private(rendered / "network-config", render_network_config(config, node)) base_directory = f"{config.vm_image_directory}/base" node_directory = f"{config.vm_image_directory}/{config.name}/{node.name}" base_image = f"{base_directory}/{config.image.sha256}.qcow2" disk_image = f"{node_directory}/root.qcow2" seed_image = f"{node_directory}/seed.img" runner.hypervisor(node, ["install", "-d", "-m", "0750", base_directory, node_directory]) current = runner.hypervisor( node, ["sha256sum", base_image], capture=True, check=False, timeout=120, ) current_digest = ( current.stdout.decode(errors="replace").split(maxsplit=1)[0] if current.returncode == 0 and current.stdout else "" ) if current_digest != config.image.sha256: temporary = f"{base_image}.download" runner.hypervisor( node, [ "curl", "--proto", "=https", "--tlsv1.2", "--fail", "--location", "--output", temporary, config.image.url, ], timeout=1800, ) downloaded = runner.hypervisor( node, ["sha256sum", temporary], capture=True, timeout=1800, ) digest = downloaded.stdout.decode().split(maxsplit=1)[0] if digest != config.image.sha256: runner.hypervisor(node, ["rm", "-f", temporary], check=False) raise LabOperationError(f"cloud image digest mismatch on {node.hypervisor}") runner.hypervisor(node, ["mv", temporary, base_image]) runner.hypervisor( node, [ "qemu-img", "create", "-f", "qcow2", "-F", "qcow2", "-b", base_image, disk_image, f"{node.disk_gib}G", ], ) for name in ("user-data", "meta-data", "network-config"): runner.copy_to_hypervisor(node, rendered / name, f"{node_directory}/{name}") runner.hypervisor( node, [ "cloud-localds", "--network-config", f"{node_directory}/network-config", seed_image, f"{node_directory}/user-data", f"{node_directory}/meta-data", ], ) runner.hypervisor( node, [ "virt-install", "--name", domain, "--description", _domain_description(config, node), "--memory", str(node.memory_mib), "--vcpus", str(node.cpus), "--cpu", "host-passthrough", "--os-variant", "ubuntu24.04", "--disk", f"path={disk_image},format=qcow2,bus=virtio", "--disk", f"path={seed_image},device=cdrom", "--network", f"bridge={config.network.bridge},model=virtio,mac={node.mac_address}", "--graphics", "none", "--console", "pty,target_type=serial", "--import", "--noautoconsole", ], timeout=120, ) runner.hypervisor(node, ["virsh", "autostart", "--disable", domain], check=False) def _wait_for_ssh(config: LabConfig, runner: CommandRunner, node: LabNode) -> None: deadline = time.monotonic() + _SSH_TIMEOUT_SECONDS last_error = "" while time.monotonic() < deadline: try: with socket.create_connection((str(node.address), 22), timeout=3): pass result = runner.vm(node, ["true"], capture=True, check=False, timeout=15) if result.returncode == 0: return last_error = (result.stderr or b"").decode(errors="replace").strip() except OSError as exc: last_error = str(exc) time.sleep(5) raise LabOperationError(f"SSH did not become ready on {node.name}: {last_error}") def _prepare_local_state(config: LabConfig) -> None: config.state_directory.mkdir(parents=True, exist_ok=True, mode=0o700) config.state_directory.chmod(0o700) (config.state_directory / "ssh_known_hosts").touch(mode=0o600, exist_ok=True) (config.state_directory / "ssh_known_hosts").chmod(0o600) def _write_inventory(config: LabConfig) -> None: payload = { "schema_version": 1, "name": config.name, "mode": config.mode, "evidence_capable": config.evidence_capable, "nodes": [ { **asdict(node), "address": str(node.address), } for node in config.nodes ], } write_private( config.state_directory / "inventory.json", json.dumps(payload, indent=2, sort_keys=True) + "\n", ) def _representative_hypervisors( config: LabConfig, ) -> Iterable[tuple[str, LabNode]]: seen: set[str] = set() for node in config.nodes: if node.hypervisor in seen: continue seen.add(node.hypervisor) yield node.hypervisor, node def _print_mutation_preview(operation: str, config: LabConfig, actions: Sequence[str]) -> None: print(f"Dry run: {operation} lab {config.name} ({config.mode})") for action in actions: print(f"- {action}") print("Re-run with --apply to perform these mutations.") def _domain_name(config: LabConfig, node: LabNode) -> str: return f"{config.name}-{node.name}" def _domain_description(config: LabConfig, node: LabNode) -> str: return ( f"GovOPlaN managed lab; lab={config.name}; node={node.name}; " f"schema={config.schema_version}" ) def _assert_domain_owned( config: LabConfig, runner: CommandRunner, node: LabNode, ) -> None: domain = _domain_name(config, node) description = runner.hypervisor( node, ["virsh", "desc", domain], capture=True, check=False, timeout=20, ) if ( description.returncode != 0 or description.stdout.decode(errors="replace").strip() != _domain_description(config, node) ): raise LabOperationError( f"domain {domain} does not carry the expected GovOPlaN lab ownership marker" ) block_devices = runner.hypervisor( node, ["virsh", "domblklist", "--details", domain], capture=True, timeout=20, ).stdout.decode(errors="replace") node_directory = f"{config.vm_image_directory}/{config.name}/{node.name}" expected_paths = { f"{node_directory}/root.qcow2", f"{node_directory}/seed.img", } missing = sorted(path for path in expected_paths if path not in block_devices) if missing: raise LabOperationError( f"domain {domain} does not use the expected lab-owned disks: " + ", ".join(missing) ) def deploy(config: LabConfig, *, apply: bool, verbose: bool = False) -> None: actions = [ "verify and adopt the signed GovOPlaN release", f"start PostgreSQL, Redis, Garage and test mail on {config.state_node.name}", f"install pinned K3s {config.k3s.version} on control and worker nodes", "apply runtime, TLS and CA secrets without printing them", "render and apply the supported stateless Kubernetes profile", ] if not apply: _print_mutation_preview("deploy", config, actions) return _prepare_local_state(config) runner = CommandRunner(config, verbose=verbose) secrets_value = _load_or_create_secrets(config) manifest, installation = _prepare_release(config, runner, secrets_value) _ensure_certificates(config, runner) _deploy_state_services(config, runner, manifest, secrets_value) _deploy_k3s(config, runner, secrets_value, update=False) _deploy_application(config, runner, installation) print(f"GovOPlaN lab deployment is ready at {config.public_url}.") print(f"Trust the private lab CA at {config.state_directory / 'pki' / 'ca.crt'}.") print(f"Host mappings are recorded in {config.state_directory / 'hosts'}.") def update(config: LabConfig, *, apply: bool, verbose: bool = False) -> None: actions = [ f"serially reconcile K3s to {config.k3s.version}", "pull the newly pinned shared-state images", "verify/adopt the configured GovOPlaN release and apply its migration/runtime manifest", ] if not apply: _print_mutation_preview("update", config, actions) return _prepare_local_state(config) runner = CommandRunner(config, verbose=verbose) secrets_value = _load_or_create_secrets(config) manifest, installation = _prepare_release(config, runner, secrets_value) _ensure_certificates(config, runner) _deploy_state_services(config, runner, manifest, secrets_value) _deploy_k3s(config, runner, secrets_value, update=True) _deploy_application(config, runner, installation) print(f"Updated lab {config.name} to the configured immutable inputs.") def _prepare_release( config: LabConfig, runner: CommandRunner, lab_secrets: Mapping[str, str], ) -> tuple[dict[str, Any], Path]: release_directory = config.state_directory / "release" release_directory.mkdir(parents=True, exist_ok=True, mode=0o700) manifest_path = release_directory / "distribution-manifest.json" keyring_path = release_directory / "distribution-keyring.json" _download_verified( config.release.manifest_url, config.release.manifest_sha256, manifest_path, maximum_bytes=4 * 1024 * 1024, ) _download_verified( config.release.keyring_url, config.release.keyring_sha256, keyring_path, maximum_bytes=256 * 1024, ) try: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: raise LabOperationError("downloaded release manifest is not valid JSON") from exc if not isinstance(manifest, dict): raise LabOperationError("downloaded release manifest must be a JSON object") installation = config.state_directory / "installation" if not (installation / "installation.json").is_file(): runner.run( [ sys.executable, str(DEPLOYER), "init", "--directory", str(installation), "--non-interactive", "--installation-id", config.name, "--profile", "self-hosted", "--public-url", config.public_url, "--postgres", "external", "--database-url", "postgresql://placeholder:placeholder@127.0.0.1/govoplan", "--redis", "external", "--redis-url", "redis://:placeholder@127.0.0.1:6379/0", "--mail", "external-relay", "--storage", "s3", "--s3-endpoint-url", config.s3_url, "--s3-region", "garage", "--s3-access-key-id", "GK00000000000000000000000000000000", "--s3-secret-access-key", "lab-placeholder-not-a-secret", "--s3-bucket", "govoplan-files", "--ingress", "existing-proxy", "--trusted-proxy-cidr", str( ipaddress.ip_network( f"{config.network.gateway}/{config.network.prefix_length}", strict=False, ) ), "--api-replicas", str(config.api_replicas), "--web-replicas", str(config.web_replicas), "--worker-replicas", str(config.worker_replicas), "--module-set", config.module_set, "--release-channel", config.release.channel, ], timeout=120, ) _reconcile_installation_environment(config, installation, lab_secrets) runner.run( [ sys.executable, str(DEPLOYER), "verify-release", "--directory", str(installation), "--manifest", str(manifest_path), "--manifest-sha256", config.release.manifest_sha256, "--trusted-keyring", str(keyring_path), "--adopt", ], timeout=120, ) _reconcile_installation_environment(config, installation, lab_secrets) return manifest, installation def _reconcile_installation_environment( config: LabConfig, installation: Path, lab_secrets: Mapping[str, str] ) -> None: deployment_path = REPOSITORY_ROOT / "tools" / "deployment" if str(deployment_path) not in sys.path: sys.path.insert(0, str(deployment_path)) from govoplan_deploy.bundle import read_env, write_env # type: ignore[import-not-found] env_path = installation / "secrets.env" values = read_env(env_path) database_url, pgtools_url = _database_urls(config, lab_secrets) values.update( { "DATABASE_URL": database_url, "GOVOPLAN_DATABASE_URL_PGTOOLS": pgtools_url, "REDIS_URL": _redis_url(config, lab_secrets), "FILE_STORAGE_S3_ENDPOINT_URL": config.s3_url, "FILE_STORAGE_S3_REGION": "garage", "FILE_STORAGE_S3_ACCESS_KEY_ID": lab_secrets["garage_access_key"], "FILE_STORAGE_S3_SECRET_ACCESS_KEY": lab_secrets["garage_secret_key"], "FILE_STORAGE_S3_BUCKET": "govoplan-files", "FILE_STORAGE_S3_DEPLOYMENT_MANAGED": "false", "FILE_STORAGE_S3_ENDPOINT_TRUSTED": "true", "GOVOPLAN_DB_CONNECTION_LIMIT": str(config.db_connection_limit), } ) write_env(env_path, values) def _download_verified( url: str, expected_sha256: str, output: Path, *, maximum_bytes: int, ) -> None: if output.is_file() and _sha256_file(output) == expected_sha256: return request = Request(url, headers={"User-Agent": "GovOPlaN-lab/1"}) try: with urlopen(request, timeout=60) as response: if urlsplit(response.geturl()).scheme != "https": raise LabOperationError(f"pinned input redirected outside HTTPS: {url}") payload = response.read(maximum_bytes + 1) except OSError as exc: raise LabOperationError(f"could not download pinned input: {url}") from exc if len(payload) > maximum_bytes: raise LabOperationError(f"downloaded input exceeds the size limit: {url}") digest = hashlib.sha256(payload).hexdigest() if digest != expected_sha256: raise LabOperationError(f"downloaded input digest mismatch: {url}") write_private(output, payload) def _load_or_create_secrets(config: LabConfig) -> dict[str, str]: path = config.state_directory / "lab-secrets.json" if path.is_file(): try: value = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise LabOperationError("lab-secrets.json is not valid JSON") from exc if not isinstance(value, dict) or not all( isinstance(key, str) and isinstance(item, str) and item for key, item in value.items() ): raise LabOperationError("lab-secrets.json has an invalid shape") missing = sorted(_generated_secret_names() - set(value)) if missing: raise LabOperationError("lab-secrets.json is missing: " + ", ".join(missing)) return dict(value) value = { "cluster_token": secrets.token_urlsafe(48), "postgres_password": secrets.token_urlsafe(36), "redis_password": secrets.token_urlsafe(36), "garage_access_key": f"GK{secrets.token_hex(16)}", "garage_secret_key": secrets.token_hex(32), "garage_rpc_secret": secrets.token_hex(32), "garage_admin_token": secrets.token_urlsafe(48), "garage_metrics_token": secrets.token_urlsafe(48), } write_private(path, json.dumps(value, indent=2, sort_keys=True) + "\n") return value def _generated_secret_names() -> set[str]: return { "cluster_token", "postgres_password", "redis_password", "garage_access_key", "garage_secret_key", "garage_rpc_secret", "garage_admin_token", "garage_metrics_token", } def _database_urls( config: LabConfig, lab_secrets: Mapping[str, str] ) -> tuple[str, str]: password = quote(lab_secrets["postgres_password"], safe="") host = config.state_node.address return ( f"postgresql+psycopg://govoplan:{password}@{host}:5432/govoplan?sslmode=disable", f"postgresql://govoplan:{password}@{host}:5432/govoplan?sslmode=disable", ) def _redis_url(config: LabConfig, lab_secrets: Mapping[str, str]) -> str: password = quote(lab_secrets["redis_password"], safe="") return f"redis://:{password}@{config.state_node.address}:6379/0" def _ensure_certificates(config: LabConfig, runner: CommandRunner) -> None: pki = config.state_directory / "pki" pki.mkdir(parents=True, exist_ok=True, mode=0o700) ca_key = pki / "ca.key" ca_cert = pki / "ca.crt" server_key = pki / "server.key" server_csr = pki / "server.csr" server_cert = pki / "server.crt" extension = pki / "server.ext" server_inputs = pki / "server-inputs.json" expected_server_inputs = json.dumps( { "public_host": config.public_host, "public_address": str(config.primary_control.address), "s3_host": config.s3_host, "s3_address": str(config.state_node.address), }, indent=2, sort_keys=True, ) + "\n" ca_valid = ( ca_key.is_file() and ca_cert.is_file() and runner.run( [ "openssl", "x509", "-checkend", str(30 * 24 * 60 * 60), "-noout", "-in", str(ca_cert), ], capture=True, check=False, ).returncode == 0 ) ca_rotated = not ca_valid if ca_rotated: runner.run( [ "openssl", "genpkey", "-algorithm", "RSA", "-pkeyopt", "rsa_keygen_bits:3072", "-out", str(ca_key), ], sensitive=True, ) runner.run( [ "openssl", "req", "-x509", "-new", "-sha256", "-days", "3650", "-key", str(ca_key), "-subj", f"/CN={config.name} private lab CA", "-out", str(ca_cert), ], ) server_valid = ( not ca_rotated and server_key.is_file() and server_cert.is_file() and server_inputs.is_file() and server_inputs.read_text(encoding="utf-8") == expected_server_inputs and runner.run( [ "openssl", "x509", "-checkend", str(30 * 24 * 60 * 60), "-noout", "-in", str(server_cert), ], capture=True, check=False, ).returncode == 0 ) if not server_valid: runner.run( [ "openssl", "genpkey", "-algorithm", "RSA", "-pkeyopt", "rsa_keygen_bits:3072", "-out", str(server_key), ], sensitive=True, ) runner.run( [ "openssl", "req", "-new", "-sha256", "-key", str(server_key), "-subj", f"/CN={config.public_host}", "-out", str(server_csr), ], ) write_private( extension, "basicConstraints=critical,CA:FALSE\n" "keyUsage=critical,digitalSignature,keyEncipherment\n" "extendedKeyUsage=serverAuth\n" f"subjectAltName=DNS:{config.public_host},DNS:{config.s3_host}," f"IP:{config.primary_control.address},IP:{config.state_node.address}\n", ) runner.run( [ "openssl", "x509", "-req", "-sha256", "-days", "397", "-in", str(server_csr), "-CA", str(ca_cert), "-CAkey", str(ca_key), "-CAcreateserial", "-extfile", str(extension), "-out", str(server_cert), ], ) write_private(server_inputs, expected_server_inputs) for path in (ca_key, server_key): path.chmod(0o600) write_private(config.state_directory / "hosts", render_hosts(config)) def _deploy_state_services( config: LabConfig, runner: CommandRunner, manifest: Mapping[str, Any], lab_secrets: Mapping[str, str], ) -> None: dependencies = manifest.get("dependencies") if not isinstance(dependencies, dict) or not all( isinstance(key, str) and isinstance(value, str) for key, value in dependencies.items() ): raise LabOperationError("release manifest dependencies have an invalid shape") local = config.state_directory / "rendered" / "state" tls = local / "tls" tls.mkdir(parents=True, exist_ok=True, mode=0o700) write_private(local / "compose.json", render_state_compose(dependencies)) write_private(local / "garage.toml", render_garage_config()) write_private(local / "Caddyfile", render_caddyfile()) write_private( local / ".env", render_state_environment( { "POSTGRES_DB": "govoplan", "POSTGRES_USER": "govoplan", "POSTGRES_PASSWORD": lab_secrets["postgres_password"], "REDIS_PASSWORD": lab_secrets["redis_password"], "GARAGE_DEFAULT_ACCESS_KEY": lab_secrets["garage_access_key"], "GARAGE_DEFAULT_SECRET_KEY": lab_secrets["garage_secret_key"], "GARAGE_DEFAULT_BUCKET": "govoplan-files", "GARAGE_RPC_SECRET": lab_secrets["garage_rpc_secret"], "GARAGE_ADMIN_TOKEN": lab_secrets["garage_admin_token"], "GARAGE_METRICS_TOKEN": lab_secrets["garage_metrics_token"], } ), ) write_private(tls / "server.crt", (config.state_directory / "pki/server.crt").read_bytes()) write_private(tls / "server.key", (config.state_directory / "pki/server.key").read_bytes()) node = config.state_node remote = f"/home/{config.ssh_user}/{_STATE_REMOTE_DIRECTORY}" runner.vm(node, ["mkdir", "-p", f"{remote}/tls"]) for source, destination in ( (local / "compose.json", f"{remote}/compose.json"), (local / "garage.toml", f"{remote}/garage.toml"), (local / "Caddyfile", f"{remote}/Caddyfile"), (local / ".env", f"{remote}/.env"), (tls / "server.crt", f"{remote}/tls/server.crt"), (tls / "server.key", f"{remote}/tls/server.key"), ): runner.copy_to_vm(node, source, destination) runner.vm( node, [ "docker", "compose", "--env-file", f"{remote}/.env", "--file", f"{remote}/compose.json", "pull", ], sudo=True, timeout=1800, ) runner.vm( node, [ "docker", "compose", "--env-file", f"{remote}/.env", "--file", f"{remote}/compose.json", "up", "--detach", "--remove-orphans", ], sudo=True, timeout=300, ) runner.vm( node, [ "docker", "compose", "--env-file", f"{remote}/.env", "--file", f"{remote}/compose.json", "restart", "s3-tls", ], sudo=True, timeout=120, ) _wait_for_tcp(str(node.address), 5432, label="PostgreSQL") _wait_for_tcp(str(node.address), 6379, label="Redis") _wait_for_tcp(str(node.address), 9443, label="Garage TLS endpoint") _wait_for_https( config.s3_url, resolve_address=str(node.address), ca_certificate=config.state_directory / "pki/ca.crt", label="Garage TLS endpoint", require_success_status=False, ) def _deploy_k3s( config: LabConfig, runner: CommandRunner, lab_secrets: Mapping[str, str], *, update: bool, ) -> None: username = os.environ.get("GOVOPLAN_LAB_REGISTRY_USERNAME", "").strip() password = os.environ.get("GOVOPLAN_LAB_REGISTRY_PASSWORD", "").strip() registry = render_registry_config(username, password) ordered = [config.primary_control, *config.controls[1:], *config.workers] for node in ordered: if update and node.role == "worker": _kubectl( config, runner, ["cordon", node.name], timeout=60, ) _kubectl( config, runner, [ "drain", node.name, "--ignore-daemonsets", "--delete-emptydir-data", "--timeout=10m", ], timeout=630, ) _install_k3s_node( config, runner, node, cluster_token=lab_secrets["cluster_token"], registry=registry, update=update, ) if node.role == "control": _wait_for_k3s_api(config, runner) _wait_for_k3s_node(config, runner, node) if update and node.role == "worker": _kubectl(config, runner, ["uncordon", node.name], timeout=60) def _install_k3s_node( config: LabConfig, runner: CommandRunner, node: LabNode, *, cluster_token: str, registry: str, update: bool, ) -> None: rendered = config.state_directory / "rendered" / "k3s" / node.name rendered.mkdir(parents=True, exist_ok=True, mode=0o700) config_path = rendered / "config.yaml" registry_path = rendered / "registries.yaml" write_private( config_path, render_k3s_config(config, node, cluster_token=cluster_token), ) if registry: write_private(registry_path, registry) temporary_prefix = f"/tmp/{config.name}-{node.name}" runner.copy_to_vm(node, config_path, f"{temporary_prefix}-config.yaml") if registry: runner.copy_to_vm(node, registry_path, f"{temporary_prefix}-registries.yaml") runner.vm(node, ["install", "-d", "-m", "0700", "/etc/rancher/k3s"], sudo=True) runner.vm( node, [ "install", "-m", "0600", f"{temporary_prefix}-config.yaml", "/etc/rancher/k3s/config.yaml", ], sudo=True, ) if registry: runner.vm( node, [ "install", "-m", "0600", f"{temporary_prefix}-registries.yaml", "/etc/rancher/k3s/registries.yaml", ], sudo=True, sensitive=True, ) else: runner.vm( node, ["rm", "-f", "/etc/rancher/k3s/registries.yaml"], sudo=True, ) runner.vm( node, [ "rm", "-f", f"{temporary_prefix}-config.yaml", f"{temporary_prefix}-registries.yaml", ], ) _install_verified_remote( runner, node, url=config.k3s.binary_url, sha256=config.k3s.binary_sha256, destination="/usr/local/bin/k3s", mode="0755", ) _install_verified_remote( runner, node, url=config.k3s.install_script_url, sha256=config.k3s.install_script_sha256, destination=f"{temporary_prefix}-install.sh", mode="0700", ) service = "k3s" if node.role == "control" else "k3s-agent" role = "server" if node.role == "control" else "agent" runner.vm( node, [ "env", "INSTALL_K3S_SKIP_DOWNLOAD=true", "INSTALL_K3S_SKIP_START=true", f"INSTALL_K3S_VERSION={config.k3s.version}", f"{temporary_prefix}-install.sh", role, ], sudo=True, timeout=180, ) runner.vm(node, ["systemctl", "enable", service], sudo=True) runner.vm( node, ["systemctl", "restart" if update else "start", service], sudo=True, timeout=120, ) runner.vm( node, ["systemctl", "is-active", "--quiet", service], sudo=True, timeout=60, ) runner.vm( node, ["rm", "-f", f"{temporary_prefix}-install.sh"], sudo=True, check=False, ) def _install_verified_remote( runner: CommandRunner, node: LabNode, *, url: str, sha256: str, destination: str, mode: str, ) -> None: current = runner.vm( node, ["sha256sum", destination], sudo=True, capture=True, check=False, timeout=60, ) current_digest = ( current.stdout.decode(errors="replace").split(maxsplit=1)[0] if current.returncode == 0 and current.stdout else "" ) if current_digest == sha256: return temporary = f"{destination}.download" runner.vm( node, [ "curl", "--proto", "=https", "--tlsv1.2", "--fail", "--location", "--output", temporary, url, ], sudo=True, timeout=900, ) downloaded = runner.vm( node, ["sha256sum", temporary], sudo=True, capture=True, timeout=900, ) digest = downloaded.stdout.decode().split(maxsplit=1)[0] if digest != sha256: runner.vm(node, ["rm", "-f", temporary], sudo=True, check=False) raise LabOperationError(f"pinned download digest mismatch on {node.name}: {url}") runner.vm( node, ["install", "-m", mode, temporary, destination], sudo=True, ) runner.vm(node, ["rm", "-f", temporary], sudo=True, check=False) def _wait_for_k3s_api(config: LabConfig, runner: CommandRunner) -> None: _wait_for_tcp(str(config.primary_control.address), 6443, label="K3s API") deadline = time.monotonic() + 300 while time.monotonic() < deadline: result = runner.vm( config.primary_control, ["k3s", "kubectl", "get", "--raw=/readyz"], sudo=True, capture=True, check=False, timeout=20, ) if result.returncode == 0 and result.stdout.strip() == b"ok": return time.sleep(5) raise LabOperationError("K3s API did not become ready") def _wait_for_k3s_node( config: LabConfig, runner: CommandRunner, node: LabNode ) -> None: deadline = time.monotonic() + 300 while time.monotonic() < deadline: result = _kubectl( config, runner, [ "get", "node", node.name, "-o", "jsonpath={.status.conditions[?(@.type=='Ready')].status}", ], capture=True, check=False, timeout=30, ) if result.returncode == 0 and result.stdout.strip() == b"True": return time.sleep(5) raise LabOperationError(f"K3s node did not become Ready: {node.name}") def _deploy_application( config: LabConfig, runner: CommandRunner, installation: Path ) -> None: deployment_path = REPOSITORY_ROOT / "tools" / "deployment" if str(deployment_path) not in sys.path: sys.path.insert(0, str(deployment_path)) from govoplan_deploy.bundle import read_env # type: ignore[import-not-found] values = read_env(installation / "secrets.env") missing = [key for key in _SECRET_KEYS if not values.get(key)] if missing: raise LabOperationError("installation secrets are missing: " + ", ".join(missing)) namespace_manifest = ( json.dumps( { "apiVersion": "v1", "kind": "Namespace", "metadata": {"name": config.namespace}, } ) + "\n" ).encode() _kubectl(config, runner, ["apply", "-f", "-"], input_bytes=namespace_manifest) runtime_secret = render_secret_manifest( namespace=config.namespace, name="govoplan-runtime", values={key: values[key] for key in _SECRET_KEYS}, ) ca_certificate = (config.state_directory / "pki/ca.crt").read_bytes() ca_secret_name = f"{config.name}-s3-ca" ca_secret = render_secret_manifest( namespace=config.namespace, name=ca_secret_name, values={"ca.crt": ca_certificate}, ) tls_secret = render_tls_secret_manifest( namespace=config.namespace, name="govoplan-tls", certificate=(config.state_directory / "pki/server.crt").read_bytes(), private_key=(config.state_directory / "pki/server.key").read_bytes(), ) for secret_manifest in (runtime_secret, ca_secret, tls_secret): _kubectl( config, runner, ["apply", "-f", "-"], input_bytes=secret_manifest, sensitive=True, ) kubernetes_manifest = installation / "kubernetes.json" runner.run( [ sys.executable, str(DEPLOYER), "render-kubernetes", "--directory", str(installation), "--namespace", config.namespace, "--secret-name", "govoplan-runtime", "--tls-secret-name", "govoplan-tls", "--ingress-class-name", config.ingress_class, "--s3-ca-secret-name", ca_secret_name, "--output", str(kubernetes_manifest), ], timeout=120, ) _kubectl( config, runner, ["apply", "-f", "-"], input_bytes=kubernetes_manifest.read_bytes(), timeout=180, ) rendered = json.loads(kubernetes_manifest.read_text(encoding="utf-8")) jobs = [ str(item["metadata"]["name"]) for item in rendered.get("items", []) if item.get("kind") == "Job" ] for job in jobs: _kubectl( config, runner, [ "wait", "--namespace", config.namespace, "--for=condition=complete", f"job/{job}", "--timeout=15m", ], timeout=930, ) _kubectl( config, runner, [ "wait", "--namespace", config.namespace, "--for=condition=available", "deployment", "--all", "--timeout=15m", ], timeout=930, ) _write_kubectl_wrapper(config, runner) _wait_for_https( config.public_url, resolve_address=str(config.primary_control.address), ca_certificate=config.state_directory / "pki/ca.crt", label="GovOPlaN ingress", require_success_status=True, ) def _kubectl( config: LabConfig, runner: CommandRunner, arguments: Sequence[str], *, input_bytes: bytes | None = None, capture: bool = False, check: bool = True, timeout: float | None = None, sensitive: bool = False, ) -> subprocess.CompletedProcess[bytes]: return runner.vm( config.primary_control, ["k3s", "kubectl", *arguments], sudo=True, input_bytes=input_bytes, capture=capture, check=check, timeout=timeout, sensitive=sensitive, ) def status(config: LabConfig, *, verbose: bool = False) -> int: _prepare_local_state(config) runner = CommandRunner(config, verbose=verbose) failed = False for node in config.nodes: result = runner.hypervisor( node, ["virsh", "domstate", _domain_name(config, node)], capture=True, check=False, timeout=20, ) state = ( result.stdout.decode(errors="replace").strip() if result.returncode == 0 else "missing" ) print(f"VM {node.name}: {state} ({node.hypervisor}, {node.failure_domain})") failed = failed or result.returncode != 0 cluster = _kubectl( config, runner, ["get", "nodes", "-o", "wide"], capture=True, check=False, timeout=30, ) if cluster.returncode == 0: print(cluster.stdout.decode(errors="replace").rstrip()) pods = _kubectl( config, runner, ["get", "pods", "--namespace", config.namespace, "-o", "wide"], capture=True, check=False, timeout=30, ) print(pods.stdout.decode(errors="replace").rstrip()) failed = failed or pods.returncode != 0 else: print("Kubernetes API is not reachable.") failed = True return 1 if failed else 0 def pause(config: LabConfig, *, apply: bool, verbose: bool = False) -> None: if not apply: _print_mutation_preview( "pause", config, ["gracefully shut down workers, control planes, then shared state"], ) return runner = CommandRunner(config, verbose=verbose) ordered = [*config.workers, *reversed(config.controls), config.state_node] for node in ordered: domain = _domain_name(config, node) current = runner.hypervisor( node, ["virsh", "domstate", domain], capture=True, check=False, ) if current.returncode != 0 or b"shut off" in current.stdout.lower(): continue _assert_domain_owned(config, runner, node) runner.hypervisor(node, ["virsh", "shutdown", domain]) _wait_for_domain_state(config, runner, node, "shut off", timeout=180) print(f"Paused lab {config.name}; disks and evidence remain in place.") def resume(config: LabConfig, *, apply: bool, verbose: bool = False) -> None: if not apply: _print_mutation_preview( "resume", config, ["start shared state, control planes, then workers and wait for readiness"], ) return _prepare_local_state(config) runner = CommandRunner(config, verbose=verbose) ordered = [config.state_node, *config.controls, *config.workers] for node in ordered: domain = _domain_name(config, node) _assert_domain_owned(config, runner, node) runner.hypervisor(node, ["virsh", "start", domain], check=False) _wait_for_ssh(config, runner, node) if node.role == "state": _wait_for_tcp(str(node.address), 5432, label="PostgreSQL") elif node.role == "control": _wait_for_k3s_api(config, runner) else: _wait_for_k3s_node(config, runner, node) print(f"Resumed lab {config.name}.") def destroy( config: LabConfig, *, apply: bool, confirmation: str, purge_local_state: bool, verbose: bool = False, ) -> None: actions = [ f"destroy and undefine {len(config.nodes)} lab-owned libvirt domains", "delete their lab-owned qcow2 overlays and cloud-init seeds", "preserve local evidence" if not purge_local_state else "delete local lab state and evidence", ] if not apply: _print_mutation_preview("destroy", config, actions) return if confirmation != config.name: raise LabOperationError( f"destruction requires --confirm {config.name}; received {confirmation!r}" ) runner = CommandRunner(config, verbose=verbose) for node in [*config.workers, *reversed(config.controls), config.state_node]: domain = _domain_name(config, node) existing = runner.hypervisor( node, ["virsh", "dominfo", domain], capture=True, check=False, timeout=20, ) if existing.returncode != 0: print(f"Skipping missing domain {domain}; no disk directory was removed.") _forget_vm_host_key(config, runner, node) continue _assert_domain_owned(config, runner, node) runner.hypervisor(node, ["virsh", "destroy", domain], check=False) runner.hypervisor( node, ["virsh", "undefine", domain, "--nvram"], check=False, ) node_directory = f"{config.vm_image_directory}/{config.name}/{node.name}" runner.hypervisor( node, ["rm", "--recursive", "--force", "--one-file-system", node_directory], ) _forget_vm_host_key(config, runner, node) if purge_local_state: resolved = config.state_directory.resolve() home = Path.home().resolve() if resolved in {Path("/"), home} or len(resolved.parts) < 4: raise LabOperationError("refusing to purge an unsafe state_directory") shutil.rmtree(resolved) print(f"Destroyed VM resources for lab {config.name}.") def verify( config: LabConfig, *, exercise_api_pod_loss: bool, verbose: bool = False, ) -> None: api_key = os.environ.get("GOVOPLAN_OPS_API_KEY", "").strip() if not api_key: raise LabOperationError( "GOVOPLAN_OPS_API_KEY must contain a short-lived key authorized " "to read Ops status" ) runner = CommandRunner(config, verbose=verbose) wrapper = _write_kubectl_wrapper(config, runner) environment = dict(os.environ) environment["PATH"] = f"{wrapper.parent}:{environment.get('PATH', '')}" environment["SSL_CERT_FILE"] = str(config.state_directory / "pki/ca.crt") environment["GOVOPLAN_OPS_API_KEY"] = api_key output = config.state_directory / "evidence" / "kubernetes-multi-host.json" output.parent.mkdir(parents=True, exist_ok=True, mode=0o700) command = [ sys.executable, str(DEPLOYER), "verify-kubernetes", "--directory", str(config.state_directory / "installation"), "--namespace", config.namespace, "--output", str(output), ] if exercise_api_pod_loss: command.append("--exercise-api-pod-loss") runner.run(command, timeout=900, env=environment) print(f"Wrote sanitized Kubernetes evidence to {output}.") if not config.evidence_capable: print("This is rehearsal evidence only: the worker VMs do not prove independent failure domains.") def enroll_admin( config: LabConfig, *, email: str, display_name: str | None, tenant_slug: str, tenant_name: str, apply: bool, ) -> None: artifact = config.state_directory / "first-admin-enrollment.json" if not apply: _print_mutation_preview( "enroll-admin", config, [ f"consume {artifact} over the private lab CA", f"create the first administrator {email!r} in tenant {tenant_slug!r}", "delete the local enrollment artifact after a successful response", ], ) return if not artifact.is_file(): raise LabOperationError(f"first-administrator artifact is missing: {artifact}") mode = stat.S_IMODE(artifact.stat().st_mode) if mode & 0o077: raise LabOperationError( f"first-administrator artifact must be owner-only; found mode {oct(mode)}" ) try: credential = json.loads(artifact.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise LabOperationError( "first-administrator artifact is not valid JSON" ) from exc if not isinstance(credential, dict): raise LabOperationError("first-administrator artifact must be a JSON object") endpoint = str(credential.get("endpoint") or "") header = str(credential.get("header") or "") token = str(credential.get("enrollment_token") or "") if endpoint != "/api/v1/bootstrap/first-admin": raise LabOperationError("first-administrator artifact has an unexpected endpoint") if header != "X-GovOPlaN-Enrollment-Token" or len(token) < 32: raise LabOperationError("first-administrator artifact has an invalid token contract") password = getpass.getpass("First administrator password: ") confirmation = getpass.getpass("Confirm password: ") if password != confirmation: raise LabOperationError("password confirmation does not match") if len(password) < 12: raise LabOperationError("first administrator password must contain at least 12 characters") public = urlsplit(config.public_url) if public.scheme != "https" or not public.hostname: raise LabOperationError("first-administrator enrollment requires an HTTPS public URL") body = json.dumps( { "email": email, "display_name": display_name or None, "password": password, "tenant_slug": tenant_slug, "tenant_name": tenant_name, }, separators=(",", ":"), ).encode("utf-8") context = ssl.create_default_context( cafile=str(config.state_directory / "pki" / "ca.crt") ) connection = http.client.HTTPSConnection( public.hostname, public.port or 443, timeout=30, context=context, ) try: connection.request( "POST", endpoint, body=body, headers={ "Accept": "application/json", "Content-Type": "application/json", header: token, }, ) response = connection.getresponse() response_body = response.read(64 * 1024 + 1) except (OSError, http.client.HTTPException) as exc: raise LabOperationError(f"first-administrator enrollment failed: {exc}") from exc finally: connection.close() if len(response_body) > 64 * 1024: raise LabOperationError("first-administrator enrollment response was too large") if response.status != 201: detail = response_body.decode("utf-8", errors="replace").strip() raise LabOperationError( f"first-administrator enrollment returned HTTP {response.status}: {detail}" ) artifact.unlink() print(f"Enrolled first administrator {email!r} and removed {artifact}.") def _write_kubectl_wrapper(config: LabConfig, runner: CommandRunner) -> Path: binary_directory = config.state_directory / "bin" binary_directory.mkdir(parents=True, exist_ok=True, mode=0o700) path = binary_directory / "kubectl" target = f"{config.ssh_user}@{config.primary_control.address}" ssh = runner.ssh_arguments(target) write_private(path, _render_kubectl_wrapper(ssh)) path.chmod(0o700) return path def _render_kubectl_wrapper(ssh: Sequence[str]) -> str: return ( "#!/usr/bin/env python3\n" "import os\n" "import shlex\n" "import sys\n\n" f"_SSH = {list(ssh)!r}\n" '_REMOTE = ["sudo", "--", "k3s", "kubectl", *sys.argv[1:]]\n' "os.execvp(_SSH[0], [*_SSH, shlex.join(_REMOTE)])\n" ) def _wait_for_domain_state( config: LabConfig, runner: CommandRunner, node: LabNode, desired: str, *, timeout: float, ) -> None: deadline = time.monotonic() + timeout domain = _domain_name(config, node) while time.monotonic() < deadline: result = runner.hypervisor( node, ["virsh", "domstate", domain], capture=True, check=False, timeout=20, ) if result.returncode == 0 and result.stdout.decode().strip().lower() == desired: return time.sleep(3) raise LabOperationError(f"domain {domain} did not reach state {desired!r}") def _wait_for_tcp(host: str, port: int, *, label: str, timeout: float = 300) -> None: deadline = time.monotonic() + timeout last_error = "" while time.monotonic() < deadline: try: with socket.create_connection((host, port), timeout=3): return except OSError as exc: last_error = str(exc) time.sleep(3) raise LabOperationError(f"{label} did not become reachable at {host}:{port}: {last_error}") def _wait_for_https( url: str, *, resolve_address: str, ca_certificate: Path, label: str, require_success_status: bool, timeout: float = 300, ) -> None: parsed = urlsplit(url) if parsed.scheme != "https" or parsed.hostname is None: raise LabOperationError(f"{label} URL must be HTTPS") port = parsed.port or 443 command = [ "curl", "--silent", "--show-error", "--output", "/dev/null", "--connect-timeout", "5", "--max-time", "15", "--cacert", str(ca_certificate), "--resolve", f"{parsed.hostname}:{port}:{resolve_address}", ] if require_success_status: command.append("--fail") command.append(url) deadline = time.monotonic() + timeout last_error = "" while time.monotonic() < deadline: result = subprocess.run(command, capture_output=True, check=False) if result.returncode == 0: return last_error = result.stderr.decode(errors="replace").strip() time.sleep(3) raise LabOperationError(f"{label} did not become ready: {last_error}") 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 _forget_vm_host_key( config: LabConfig, runner: CommandRunner, node: LabNode, ) -> None: runner.run( [ "ssh-keygen", "-f", str(config.state_directory / "ssh_known_hosts"), "-R", str(node.address), ], check=False, capture=True, )