#!/usr/bin/env python3 """Repository-independent KAPTRACS host lifecycle CLI (Ubuntu 24.04/26.04 x86-64).""" from __future__ import annotations import argparse import base64 import contextlib import getpass import hashlib import hmac import ipaddress import json import os import platform import re import secrets import shutil import subprocess import sys import tarfile import tempfile from datetime import datetime, timezone from pathlib import Path from urllib import request CLI_VERSION = "1.0.0" TRUSTED_PUBLIC_KEY = '-----BEGIN PUBLIC KEY-----\nMCowBQYDK2VwAyEASf5DeO/FBo8rhI3ICew5Wt/1G7kUBN/w0F7ThEtivPA=\n-----END PUBLIC KEY-----\n' BASE = "https://releases.kaptracs.com" CONFIG = Path("/etc/kaptracs") RELEASES = Path("/opt/kaptracs/releases") STATE = Path("/var/lib/kaptracs") CURRENT = Path("/opt/kaptracs/current") EXECUTABLE = Path("/usr/local/bin/kaptracs") SERVICES = ["app", "mail-worker", "notification-worker", "backup-worker"] PROFILES = {"local", "lan", "public", "reverse-proxy", "cloudflare"} BACKUP_FORMAT = "kaptracs-recovery-v2" OFFLINE_ROOT = STATE / "offline" def version(value): match = re.fullmatch(r"v?(\d+)\.(\d+)\.(\d+)", value) if not match: raise ValueError("Invalid semantic version") return tuple(map(int, match.groups())) def atomic(path, value): path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = path.with_name(path.name + ".tmp") with temporary.open("w", encoding="utf-8") as handle: os.chmod(temporary, 0o600) handle.write( json.dumps(value, sort_keys=True, indent=2) if not isinstance(value, str) else value ) handle.flush() os.fsync(handle.fileno()) temporary.replace(path) def run(arguments, **kwargs): # Capture diagnostics rather than exposing commands containing credentials. return subprocess.run( arguments, check=True, capture_output=True, timeout=900, **kwargs ) def stream_file(arguments, path, *, restore_input=False): """Stream recovery data on disk instead of loading a host's data into RAM.""" with path.open("rb" if restore_input else "wb") as handle: subprocess.run( arguments, check=True, timeout=3600, stdin=handle if restore_input else None, stdout=subprocess.DEVNULL if restore_input else handle, stderr=subprocess.PIPE, ) def progress(stage): path = STATE / "operation.json" if path.exists(): state = json.loads(path.read_text()) state["stage"] = stage atomic(path, state) class NoRedirect(request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): raise ValueError("Distribution redirects are not allowed") def download(url, limit=1024 * 1024): from urllib.parse import urlsplit parsed = urlsplit(url) if ( parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password ): raise ValueError("Public HTTPS download required") distribution_request = request.Request( url, headers={"User-Agent": "kaptracs-installer/1.0"} ) with request.build_opener(request.ProxyHandler({}), NoRedirect()).open( distribution_request, timeout=60 ) as response: raw = response.read(limit + 1) if len(raw) > limit: raise ValueError("Artifact size limit exceeded") return raw def sha256_file(path): 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 verify(raw, signature, public_key=None): key = public_key if public_key is not None else TRUSTED_PUBLIC_KEY if not key or "PUBLIC KEY" not in key: raise ValueError( "A trusted release verification key must be shipped with this CLI" ) with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / "key.pem").write_text(key) (root / "payload").write_bytes(raw) (root / "signature").write_bytes(signature) run( [ "openssl", "pkeyutl", "-verify", "-pubin", "-inkey", str(root / "key.pem"), "-rawin", "-in", str(root / "payload"), "-sigfile", str(root / "signature"), ] ) payload = json.loads(raw) if not isinstance(payload, dict): raise ValueError("Signed metadata must be an object") return payload def signed(url): raw = download(url) signature = download(url + ".sig", 1024) return verify(raw, signature), raw, signature def validate_manifest(payload): if payload.get("schema_version") != 2: raise ValueError("Unsupported release manifest schema") version(payload["version"]) if version(payload["minimum_cli_version"]) > version(CLI_VERSION): raise ValueError("Upgrade the host CLI first") image = payload["images"]["application"] if not re.fullmatch(r"ghcr\.io/[a-z0-9._/-]+@sha256:[a-f0-9]{64}", image): raise ValueError("Application image must be digest pinned") if not re.fullmatch("[a-f0-9]{64}", payload["bundle"]["sha256"]): raise ValueError("Invalid bundle digest") if payload["database"]["minimum_postgresql_major"] != 17: raise ValueError("Unsupported database major") profiles = payload["supported_profiles"] if not profiles or not set(profiles) <= PROFILES: raise ValueError("Unsupported deployment profile") for source in payload["supported_source_versions"]: version(source) return payload def legacy_recovery_manifest(release_version): descriptor, raw, signature = signed( f"{BASE}/recovery/legacy/{release_version}.json" ) validate_manifest(descriptor) if ( descriptor.get("version") != release_version or descriptor.get("legacy_recovery_descriptor") is not True ): raise ValueError("Invalid legacy recovery descriptor") return descriptor, raw, signature def unpack_offline(archive, metadata_path=None, signature_path=None): """Verify and stage an offline archive without making any network request.""" archive = Path(archive).resolve(strict=True) metadata_path = Path( metadata_path or str(archive).removesuffix(".tar.gz") + ".json" ) signature_path = Path(signature_path or str(metadata_path) + ".sig") raw = metadata_path.read_bytes() metadata = verify(raw, signature_path.read_bytes()) if (metadata.get("schema_version"), metadata.get("platform")) != ( 1, "ubuntu-24.04-amd64", ): raise ValueError("Unsupported offline bundle metadata") release_version = metadata.get("version", "") version(release_version) destination = OFFLINE_ROOT / release_version if destination.exists(): marker = destination / "offline.json" if not marker.exists() or marker.read_bytes() != raw: raise ValueError("Staged offline release differs from immutable metadata") return metadata, destination OFFLINE_ROOT.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with tempfile.TemporaryDirectory(dir=OFFLINE_ROOT.parent) as temporary: temporary_root = Path(temporary) with tarfile.open(archive, "r:gz") as bundle: members = bundle.getmembers() actual = {member.name for member in members} expected = set(metadata.get("files", {})) if actual != expected or len(actual) != len(members): raise ValueError("Offline archive contents differ from signed metadata") for member in members: if ( not member.isfile() or member.name.startswith("/") or ".." in Path(member.name).parts ): raise ValueError("Unsafe offline archive member") descriptor = metadata["files"][member.name] if member.size != descriptor["size"]: raise ValueError("Offline artifact size mismatch") source = bundle.extractfile(member) target = temporary_root / member.name target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(source.read()) if sha256_file(target) != descriptor["sha256"]: raise ValueError("Offline artifact digest mismatch") (temporary_root / "offline.json").write_bytes(raw) (temporary_root / "offline.json.sig").write_bytes(signature_path.read_bytes()) OFFLINE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o700) temporary_root.rename(destination) return metadata, destination def prepare_offline(path): release = platform.freedesktop_os_release() if release.get("ID") != "ubuntu" or release.get("VERSION_ID") != "24.04": raise ValueError("Offline bundles currently require Ubuntu 24.04") metadata, root = unpack_offline(path) packages = sorted((root / "packages").glob("*.deb")) if packages: run(["apt-get", "install", "--yes", "--no-download", *map(str, packages)]) for image in sorted((root / "images").glob("*.tar")): run(["docker", "load", "--input", str(image)]) return metadata, root def prepare_release_bundle(path): """Stage a signed deployment bundle supplied directly by a recovery operator.""" bundle = Path(path).resolve(strict=True) manifest_path = bundle.parent / "manifest.json" signature_path = bundle.parent / "manifest.json.sig" raw, signature = manifest_path.read_bytes(), signature_path.read_bytes() manifest = validate_manifest(verify(raw, signature)) if hashlib.sha256(bundle.read_bytes()).hexdigest() != manifest["bundle"]["sha256"]: raise ValueError("Recovery release bundle digest mismatch") root = STATE / "recovery-bundles" / manifest["version"] release = root / "release" if root.exists(): if (release / "manifest.json").read_bytes() != raw: raise ValueError("Staged recovery release differs from signed metadata") return manifest, root release.mkdir(parents=True, mode=0o700) shutil.copyfile(bundle, release / bundle.name) (release / "manifest.json").write_bytes(raw) (release / "manifest.json.sig").write_bytes(signature) return manifest, root def network_values( profile, hostname, bind_address=None, trusted_proxies=None, accept_lan_http=False ): if profile not in PROFILES: raise ValueError("Unsupported networking profile") bind = bind_address or ("127.0.0.1" if profile == "local" else "0.0.0.0") try: address = ipaddress.ip_address(bind) except ValueError as exc: raise ValueError("--bind-address must be an IP address") from exc if profile == "lan": if ( not address.is_private or address.is_loopback or address.is_unspecified or address.is_multicast or not accept_lan_http ): raise ValueError( "LAN HTTP requires a private --bind-address and --accept-trusted-lan-http" ) if profile == "reverse-proxy": if not hostname or not hostname.startswith("https://"): raise ValueError("Reverse proxy requires an external HTTPS --hostname URL") proxies = [ item.strip() for item in (trusted_proxies or "").split(",") if item.strip() ] if not proxies: raise ValueError("Reverse proxy requires --trusted-proxy addresses") for proxy in proxies: ipaddress.ip_network(proxy, strict=False) else: proxies = [] if profile == "local": external = "http://localhost:8080" elif profile == "lan": external = f"http://{bind}:8080" elif profile == "reverse-proxy": external = hostname.rstrip("/") else: if not hostname or not re.fullmatch(r"[a-zA-Z0-9.-]+", hostname): raise ValueError("Specify a valid --hostname") external = "https://" + hostname return { "KAPTRACS_PROFILE": profile, "KAPTRACS_BIND_ADDRESS": bind, "DEPLOY_HOSTNAME": hostname or "localhost", "APP_BASE_URL": external, "SESSION_COOKIE_SECURE": "false" if profile in {"local", "lan"} else "true", "TRUSTED_PROXY_ADDRESSES": ",".join(proxies), } def channel(name="stable"): payload, _, _ = signed(f"{BASE}/channels/{name}.json") if payload.get("schema_version") != 2 or payload.get("channel") != name: raise ValueError("Invalid release channel") expires = datetime.fromisoformat(payload["expires_at"].replace("Z", "+00:00")) if expires <= datetime.now(timezone.utc): raise ValueError("Release channel has expired") sequence = payload["sequence"] if not isinstance(sequence, int) or sequence < 1: raise ValueError("Invalid channel sequence") path = STATE / f"{name}-channel.json" previous = json.loads(path.read_text()) if path.exists() else {} if sequence < previous.get("sequence", 0): raise ValueError("Release channel rollback rejected") if sequence == previous.get("sequence") and payload != previous: raise ValueError("Channel sequence was reused with different content") available_manifest = validate_manifest(signed(payload["manifest_url"])[0]) atomic(STATE / "available-manifest.json", available_manifest) atomic(path, payload) return payload def extract(bundle, destination): expected = { "kaptracs", "kaptracs-host-agent", "kaptracs-host-agent.service", "kaptracs-backup.service", "kaptracs-backup.timer", "compose.yml", "Caddyfile", "DistributionCaddyfile", "README.md", } with tarfile.open(bundle, "r:gz") as archive: members = archive.getmembers() names = [member.name for member in members] if set(names) != expected or len(names) != len(expected): raise ValueError("Unexpected deployment bundle members") if any( not member.isfile() or member.size > 16 * 1024 * 1024 for member in members ): raise ValueError("Unsafe deployment archive member") destination.mkdir(parents=True, mode=0o700) for member in members: with archive.extractfile(member) as source: data = source.read(member.size + 1) if len(data) != member.size: raise ValueError("Truncated deployment member") target = destination / member.name target.write_bytes(data) target.chmod( 0o755 if member.name in {"kaptracs", "kaptracs-host-agent"} else 0o644 ) def install_host_services(root): required = { "kaptracs-host-agent.service", "kaptracs-backup.service", "kaptracs-backup.timer", } if not all((root / name).is_file() for name in required): return systemd = Path("/etc/systemd/system") for name in sorted(required): atomic(systemd / name, (root / name).read_text()) (systemd / name).chmod(0o644) handoff = STATE / "handoff" (handoff / "requests").mkdir(parents=True, exist_ok=True, mode=0o2770) os.chown(handoff, 0, 10001) os.chown(handoff / "requests", 0, 10001) os.chmod(handoff, 0o2770) os.chmod(handoff / "requests", 0o2770) run(["systemctl", "daemon-reload"]) run( [ "systemctl", "enable", "--now", "kaptracs-host-agent.service", "kaptracs-backup.timer", ] ) def stage(release_version, offline_root=None): progress("verify-and-stage") version(release_version) if offline_root: release = Path(offline_root) / "release" raw = (release / "manifest.json").read_bytes() signature = (release / "manifest.json.sig").read_bytes() manifest = verify(raw, signature) else: url = f"{BASE}/releases/{release_version}/manifest.json" try: manifest, raw, signature = signed(url) except Exception: manifest, raw, signature = legacy_recovery_manifest(release_version) validate_manifest(manifest) if manifest["version"] != release_version: raise ValueError("Release identity mismatch") target = RELEASES / release_version if target.exists(): if (target / "manifest.json").read_bytes() != raw: raise ValueError("Installed immutable release changed") return manifest, target if offline_root: bundles = list( (Path(offline_root) / "release").glob("kaptracs-deployment-*.tar.gz") ) if len(bundles) != 1: raise ValueError("Offline release must contain one deployment bundle") payload = bundles[0].read_bytes() else: payload = download(manifest["bundle"]["url"], 64 * 1024 * 1024) if hashlib.sha256(payload).hexdigest() != manifest["bundle"]["sha256"]: raise ValueError("Deployment bundle digest mismatch") RELEASES.mkdir(parents=True, exist_ok=True, mode=0o700) with tempfile.TemporaryDirectory(dir=RELEASES) as temporary: root = Path(temporary) bundle = root / "bundle.tar.gz" bundle.write_bytes(payload) extract(bundle, root / "release") (root / "release" / "manifest.json").write_bytes(raw) (root / "release" / "manifest.json.sig").write_bytes(signature) (root / "release").rename(target) return manifest, target def env_values(): path = CONFIG / "kaptracs.env" if not path.exists(): return {} return dict( line.split("=", 1) for line in path.read_text().splitlines() if line and not line.startswith("#") ) def save_env(values): if any("\n" in value or "\r" in value for value in values.values()): raise ValueError("Environment values must be single lines") atomic( CONFIG / "kaptracs.env", "".join(f"{key}={value}\n" for key, value in sorted(values.items())), ) def save_network_override(values): """Expose the app only when the selected profile has a host listener.""" path = CONFIG / "network.override.yml" if values["KAPTRACS_PROFILE"] in {"local", "lan", "reverse-proxy"}: bind = values.get("KAPTRACS_BIND_ADDRESS", "127.0.0.1") atomic( path, "services:\n app:\n ports:\n" f' - "{bind}:${{KAPTRACS_LOCAL_PORT:-8080}}:80"\n', ) else: atomic(path, "services: {}\n") def compose(root=None, values=None, env_file=None, extra_override=None): root = root or CURRENT.resolve(strict=True) values = values or env_values() profile = values["KAPTRACS_PROFILE"] command = [ "docker", "compose", "--project-name", values["KAPTRACS_INSTALLATION_ID"], "--env-file", str(env_file or CONFIG / "kaptracs.env"), "-f", str(root / "compose.yml"), ] if profile in {"public", "cloudflare"}: command.extend(["--profile", profile]) if (CONFIG / "compose.override.yml").exists(): command.extend(["-f", str(CONFIG / "compose.override.yml")]) if (CONFIG / "network.override.yml").exists(): command.extend(["-f", str(CONFIG / "network.override.yml")]) if extra_override: command.extend(["-f", str(extra_override)]) return command def preflight(): release = platform.freedesktop_os_release() if ( release.get("ID") != "ubuntu" or release.get("VERSION_ID") not in {"24.04", "26.04"} or platform.machine() not in {"x86_64", "amd64"} ): raise ValueError("Supported online host: Ubuntu 24.04 or 26.04 x86-64") if os.geteuid() != 0: raise ValueError("Run lifecycle commands as root") for command in ("docker", "openssl", "restic"): if not shutil.which(command): raise ValueError(f"Required host dependency: {command}") run(["docker", "compose", "version"]) run(["docker", "info"]) if (os.cpu_count() or 0) < 2 or shutil.disk_usage("/var").free < 10 * 1024**3: raise ValueError("At least 2 CPUs and 10 GiB free disk are required") memory = int( next( line.split()[1] for line in Path("/proc/meminfo").read_text().splitlines() if line.startswith("MemTotal:") ) ) if memory < 4 * 1024**2: raise ValueError("At least 4 GiB RAM is required") def ensure_host_dependencies(): """Install supported Ubuntu packages on a fresh host without replacing Docker.""" if os.geteuid() != 0: raise ValueError("Run installation and recovery commands as root") missing = [name for name in ("openssl", "restic") if not shutil.which(name)] if not shutil.which("docker"): missing.extend(["docker.io", "docker-compose-v2"]) if not missing: return run(["apt-get", "update"]) run(["apt-get", "install", "--yes", *missing]) def prompt_install_network(profile, args): interactive = sys.stdin.isatty() hostname = args.hostname bind_address = args.bind_address trusted_proxy = args.trusted_proxy accept_lan = args.accept_trusted_lan_http if interactive and profile in {"public", "cloudflare"} and not hostname: hostname = input("Public DNS hostname: ").strip() if interactive and profile == "lan": bind_address = bind_address or input("Private interface IP address: ").strip() if not accept_lan: accept_lan = ( input( "Type 'accept' to allow trusted-LAN HTTP on this private address: " ) .strip() .casefold() == "accept" ) if interactive and profile == "reverse-proxy": hostname = hostname or input("External HTTPS URL: ").strip() bind_address = bind_address or input("Private bind IP address: ").strip() trusted_proxy = ( trusted_proxy or input("Trusted proxy IP/CIDR addresses (comma separated): ").strip() ) return hostname, bind_address, trusted_proxy, accept_lan def validate_compose(root, values): model = json.loads( run(compose(root, values) + ["config", "--format", "json"]).stdout ) expected = values["KAPTRACS_IMAGE"] for name, service in model["services"].items(): if service.get("build") or "@sha256:" not in service.get("image", ""): raise ValueError( "Released services must use digest-pinned images without build contexts" ) if name in SERVICES + ["migrate"] and service["image"] != expected: raise ValueError("Application image differs from signed manifest") if any( "/var/run/docker.sock" in str(volume) for volume in service.get("volumes", []) ): raise ValueError("Docker socket access is prohibited") def activate( root, manifest, values, offline=False, acceptance=False, env_file=None, extra_override=None, ): progress("compose-validation") validate_compose(root, values) cmd = compose(root, values, env_file, extra_override) if not offline: run(cmd + ["pull"]) else: run(["docker", "image", "inspect", values["KAPTRACS_IMAGE"]]) progress("database-migration") run(cmd + ["up", "-d", "--wait", "db"]) run(cmd + ["run", "--rm", "migrate"]) progress("readiness") active_services = ["app"] if acceptance else SERVICES run(cmd + ["up", "-d", "--no-deps", "--wait", *active_services]) if not acceptance and values["KAPTRACS_PROFILE"] in {"public", "cloudflare"}: edge_services = ( ["cloudflared", "distribution"] if values["KAPTRACS_PROFILE"] == "cloudflare" else ["caddy"] ) run(cmd + ["up", "-d", "--no-deps", *edge_services]) actual = ( run( cmd + [ "exec", "-T", "app", "python", "-c", "import os,urllib.request; urllib.request.urlopen('http://127.0.0.1/api/health/ready',timeout=5); print(os.environ['RELEASE_VERSION'])", ] ) .stdout.decode() .strip() ) if actual != manifest["version"]: raise ValueError("Running release identity mismatch") image_id = run( ["docker", "image", "inspect", values["KAPTRACS_IMAGE"], "--format", "{{.Id}}"] ).stdout.strip() container = run(cmd + ["ps", "-q", "app"]).stdout.decode().strip() if ( run(["docker", "inspect", container, "--format", "{{.Image}}"]).stdout.strip() != image_id ): raise ValueError("Running image differs from verified release") previous = str(CURRENT.resolve()) if CURRENT.exists() else None if previous == str(root): installed = CONFIG / "installation.json" previous = ( json.loads(installed.read_text()).get("previous") if installed.exists() else None ) progress("atomic-activation") link = CURRENT.with_name("current.next") link.unlink(missing_ok=True) link.symlink_to(root) link.replace(CURRENT) atomic( CONFIG / "installation.json", { "schema_version": 1, "id": values["KAPTRACS_INSTALLATION_ID"], "version": manifest["version"], "profile": values["KAPTRACS_PROFILE"], "previous": previous, }, ) atomic(EXECUTABLE, (root / "kaptracs").read_text()) EXECUTABLE.chmod(0o755) install_host_services(root) def restic(*arguments, repository=None, password_file=None): environment = dict( os.environ, RESTIC_REPOSITORY=str(repository or STATE / "backups"), RESTIC_PASSWORD_FILE=str( password_file or CONFIG / "secrets" / "backup-password" ), ) return run(["restic", *arguments], env=environment) def _kit_crypt(payload, passphrase, decrypt=False): """Authenticated encrypt/decrypt using scrypt, AES-CTR, and encrypt-then-MAC.""" if decrypt: envelope = json.loads(payload) if ( envelope.get("format") != "kaptracs-recovery-kit-v1" or envelope.get("kdf") != "pbkdf2-sha256-600000" ): raise ValueError("Unsupported recovery kit") salt = base64.b64decode(envelope["salt"]) nonce = base64.b64decode(envelope["nonce"]) ciphertext = base64.b64decode(envelope["ciphertext"]) else: salt, nonce, ciphertext = secrets.token_bytes(16), secrets.token_bytes(16), None # PBKDF2 is available in the Ubuntu system Python and bootstrap environments; # the high work factor protects independently chosen recovery passphrases. material = hashlib.pbkdf2_hmac( "sha256", passphrase.encode(), salt, 600_000, dklen=64 ) encryption_key, mac_key = material[:32], material[32:] if decrypt: expected = hmac.new( mac_key, salt + nonce + ciphertext, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(expected, envelope["mac"]): raise ValueError("Recovery kit authentication failed") result = run( [ "openssl", "enc", "-aes-256-ctr", "-K", encryption_key.hex(), "-iv", nonce.hex(), "-d" if decrypt else "-e", ], input=ciphertext if decrypt else payload, ).stdout if decrypt: return result ciphertext = result return json.dumps( { "format": "kaptracs-recovery-kit-v1", "kdf": "pbkdf2-sha256-600000", "salt": base64.b64encode(salt).decode(), "nonce": base64.b64encode(nonce).decode(), "ciphertext": base64.b64encode(ciphertext).decode(), "mac": hmac.new( mac_key, salt + nonce + ciphertext, hashlib.sha256 ).hexdigest(), }, sort_keys=True, ).encode() def export_recovery_kit(output, passphrase=None): values = env_values() if not values: raise ValueError("No deployment configuration exists") supplied = passphrase is not None passphrase = passphrase or getpass.getpass("Recovery kit passphrase: ") confirmation = ( passphrase if supplied else getpass.getpass("Confirm recovery kit passphrase: ") ) if not passphrase or (confirmation and confirmation != passphrase): raise ValueError("Recovery kit passphrases do not match") secret_files = {} secret_root = CONFIG / "secrets" if secret_root.is_dir(): for path in secret_root.iterdir(): if ( path.name != "backup-password" and path.is_file() and not path.is_symlink() and path.stat().st_size <= 64 * 1024 ): secret_files[path.name] = base64.b64encode(path.read_bytes()).decode() payload = json.dumps( { "schema_version": 1, "created_at": datetime.now(timezone.utc).isoformat(), "installation_id": values["KAPTRACS_INSTALLATION_ID"], "configuration": values, "secret_files": secret_files, }, sort_keys=True, ).encode() target = Path(output) atomic(target, _kit_crypt(payload, passphrase).decode()) readme = target.with_suffix(target.suffix + ".README.txt") atomic( readme, "KAPTRACS recovery kit. Store this file separately from backups. The passphrase cannot be recovered. Export a fresh kit after any application-key or deployment-configuration change.\n", ) atomic( STATE / "recovery-kit-status.json", { "exported_at": datetime.now(timezone.utc).isoformat(), "configuration_sha256": hashlib.sha256( json.dumps(values, sort_keys=True, separators=(",", ":")).encode() ).hexdigest(), }, ) return target def read_recovery_kit(path, passphrase=None): passphrase = passphrase or getpass.getpass("Recovery kit passphrase: ") kit = json.loads(_kit_crypt(Path(path).read_bytes(), passphrase, decrypt=True)) if kit.get("schema_version") != 1 or not isinstance(kit.get("configuration"), dict): raise ValueError("Recovery kit payload is invalid") if not isinstance(kit.get("secret_files", {}), dict): raise ValueError("Recovery kit secret files are invalid") for name, encoded in kit.get("secret_files", {}).items(): if not re.fullmatch(r"[a-zA-Z0-9._-]{1,128}", name): raise ValueError("Recovery kit contains an invalid secret filename") base64.b64decode(encoded, validate=True) return kit def normalize_recovery_layout(destination, release_root=None): """Recognize exactly one supported recovery layout and normalize it for restore.""" recovery_files = list(destination.rglob("recovery.json")) admin_manifests = [ path for path in destination.rglob("manifest.json") if path.parent.joinpath("database.dump").is_file() ] if recovery_files and admin_manifests: raise ValueError("Recovery snapshot layout is ambiguous") if len(recovery_files) == 1: recovered = recovery_files[0].parent identity = json.loads(recovery_files[0].read_text()) required = {"database.dump", "artifacts.tar"} if not all((recovered / name).is_file() for name in required): raise ValueError("Standalone recovery snapshot is incomplete") format_name = identity.get("format", "standalone-cli-v1") if format_name not in {"standalone-cli-v1", BACKUP_FORMAT}: raise ValueError("Unknown standalone recovery format") return recovered, identity if len(admin_manifests) == 1: manifest = json.loads(admin_manifests[0].read_text()) format_name = manifest.get("format", "admin-center-v1") if format_name not in {"admin-center-v1", BACKUP_FORMAT}: raise ValueError("Unknown Admin Center recovery format") app_data = destination / "app" / "data" if not app_data.is_dir(): matches = [ path for path in destination.rglob("data") if path.parent.name == "app" ] if len(matches) != 1: raise ValueError( "Admin Center recovery snapshot has no unique app data" ) app_data = matches[0] recovered = admin_manifests[0].parent artifacts = recovered / "artifacts.tar" with tarfile.open(artifacts, "w") as archive: archive.add(app_data, arcname="app/data", recursive=True) release_version = manifest.get("release_version") if not release_version and release_root: release_version = json.loads( (Path(release_root) / "release" / "manifest.json").read_text() )["version"] if not release_version or release_version == "unknown": raise ValueError( "Legacy Admin Center recovery requires an exact signed --release-bundle" ) expected = manifest.get("database_sha256") if expected and sha256_file(recovered / "database.dump") != expected: raise ValueError("Database backup integrity check failed") return recovered, { "format": format_name, "version": release_version, "installation_id": manifest.get("installation_id"), "database_sha256": expected, } raise ValueError("Unknown or incomplete recovery snapshot layout") def backup(repository=None, password_file=None): progress("backup") values = env_values() cmd = compose() repository = repository or values.get("BACKUP_REPOSITORY") or STATE / "backups" password_file = ( password_file or values.get("BACKUP_PASSWORD_FILE") or CONFIG / "secrets" / "backup-password" ) if str(repository).startswith("rclone:") and not shutil.which("rclone"): raise ValueError("rclone is required for the configured SMB backup repository") configured_services = set( run(cmd + ["config", "--services"]).stdout.decode().splitlines() ) lifecycle_services = [ service for service in SERVICES if service in configured_services ] if not lifecycle_services: raise ValueError("No application services are configured for backup") try: restic("cat", "config", repository=repository, password_file=password_file) except subprocess.CalledProcessError: restic("init", repository=repository, password_file=password_file) with tempfile.TemporaryDirectory(dir=STATE) as directory: stage_root = Path(directory) run(cmd + ["stop", *lifecycle_services]) try: stream_file( cmd + ["exec", "-T", "db", "pg_dump", "-U", "kaptracs", "-Fc", "kaptracs"], stage_root / "database.dump", ) stream_file( cmd + [ "run", "--rm", "--no-deps", "-T", "--user", "10001", "--entrypoint", "tar", "app", "-C", "/", "-cf", "-", "app/data", "app/config", ], stage_root / "artifacts.tar", ) stream_file( cmd + ["exec", "-T", "db", "pg_restore", "--list"], stage_root / "database.dump", restore_input=True, ) schema_revision = ( run( cmd + ["run", "--rm", "migrate", "python", "-m", "alembic", "current"] ) .stdout.decode() .strip() ) atomic( stage_root / "recovery.json", { "format": BACKUP_FORMAT, "schema_version": 2, "version": values["RELEASE_VERSION"], "installation_id": values["KAPTRACS_INSTALLATION_ID"], "schema_revision": schema_revision, "artifacts_sha256": sha256_file(stage_root / "artifacts.tar"), "database_sha256": sha256_file(stage_root / "database.dump"), }, ) result = restic( "backup", "--json", "--tag", values["KAPTRACS_INSTALLATION_ID"], str(stage_root), repository=repository, password_file=password_file, ) summaries = [json.loads(line) for line in result.stdout.splitlines()] snapshot = next( item["snapshot_id"] for item in summaries if item.get("message_type") == "summary" ) restic("check", repository=repository, password_file=password_file) atomic( STATE / "last-backup.json", { "snapshot_id": snapshot, "version": values["RELEASE_VERSION"], "at": datetime.now(timezone.utc).isoformat(), }, ) return snapshot finally: run(cmd + ["up", "-d", "--no-deps", *lifecycle_services]) def restore( snapshot, confirm, repository=None, password_file=None, recovery_kit=None, release_bundle=None, offline_root=None, offline=False, ): progress("restore") if not confirm or not re.fullmatch("[a-f0-9]{8,64}", snapshot or ""): raise ValueError("Restore requires a snapshot ID and --confirm-restore") with tempfile.TemporaryDirectory(dir=STATE) as directory: destination = Path(directory) restic( "restore", snapshot, "--target", str(destination), repository=repository, password_file=password_file, ) recovered, identity = normalize_recovery_layout(destination, offline_root) if sha256_file(recovered / "database.dump") != identity.get( "database_sha256", sha256_file(recovered / "database.dump") ): raise ValueError("Database backup integrity check failed") kit = None if recovery_kit: kit = read_recovery_kit(recovery_kit) values = kit["configuration"] elif (recovered / "kaptracs.env").exists(): values = dict( line.split("=", 1) for line in (recovered / "kaptracs.env").read_text().splitlines() if "=" in line ) else: raise ValueError("This recovery point requires --recovery-kit") if ( identity.get("installation_id") and identity["installation_id"] != values["KAPTRACS_INSTALLATION_ID"] ): raise ValueError("Recovery kit and recovery point identities differ") manifest, root = stage(identity["version"], offline_root) if values.get("KAPTRACS_IMAGE") != manifest["images"]["application"]: raise ValueError("Recovery image differs from the signed release") values = dict(values) values["KAPTRACS_INSTALLATION_ID"] += "-recovery-" + secrets.token_hex(3) candidate_env = destination / "kaptracs.env" candidate_env.write_text( "".join(f"{key}={value}\n" for key, value in sorted(values.items())) ) candidate_env.chmod(0o600) acceptance_override = destination / "acceptance.override.yml" acceptance_override.write_text( 'services:\n app:\n ports:\n - "127.0.0.1:8080:80"\n' ) cmd = compose(root, values, candidate_env, acceptance_override) run(cmd + ["up", "-d", "--wait", "db"]) stream_file( cmd + [ "exec", "-T", "db", "pg_restore", "-U", "kaptracs", "--exit-on-error", "--no-owner", "-d", "kaptracs", ], recovered / "database.dump", restore_input=True, ) # Replace only known application directories inside this installation's volumes. run( cmd + [ "run", "--rm", "--no-deps", "-T", "--user", "10001", "--entrypoint", "python", "app", "-c", "import shutil; [shutil.rmtree(p,ignore_errors=True) for p in ['/app/data','/app/config']]", ] ) stream_file( cmd + [ "run", "--rm", "--no-deps", "-T", "--user", "10001", "--entrypoint", "tar", "app", "-C", "/", "-xf", "-", ], recovered / "artifacts.tar", restore_input=True, ) activate( root, manifest, values, offline=offline, acceptance=True, env_file=candidate_env, extra_override=acceptance_override, ) save_env(values) if kit: for name, encoded in kit.get("secret_files", {}).items(): if not re.fullmatch(r"[a-zA-Z0-9._-]{1,128}", name): raise ValueError("Recovery kit contains an invalid secret filename") atomic(CONFIG / "secrets" / name, base64.b64decode(encoded).decode()) atomic( STATE / "acceptance.json", { "state": "awaiting-activation", "snapshot_id": snapshot, "version": identity["version"], }, ) @contextlib.contextmanager def locked(): import fcntl STATE.mkdir(parents=True, exist_ok=True, mode=0o700) with (STATE / "lifecycle.lock").open("a+") as handle: os.chown(handle.name, 0, 10001) os.chmod(handle.name, 0o660) fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) yield def execute(args): values = env_values() offline_root = None if getattr(args, "offline_bundle", None): offline_metadata, offline_root = prepare_offline(args.offline_bundle) elif args.command == "update-staged": if not args.version: raise ValueError("update-staged requires --version") offline_root = OFFLINE_ROOT / args.version if not (offline_root / "offline.json").is_file(): raise ValueError("The requested verified offline bundle is not staged") offline_metadata = json.loads((offline_root / "offline.json").read_text()) elif args.command in {"install", "restore"}: ensure_host_dependencies() if args.command == "stage-bundle": if not offline_root: raise ValueError("stage-bundle requires --offline-bundle") atomic(CONFIG / "host-agent.env", "KAPTRACS_OFFLINE=true\n") if CURRENT.exists(): run(["systemctl", "restart", "kaptracs-host-agent.service"]) print(f"Verified offline release staged: {offline_metadata['version']}") elif args.command == "doctor": preflight() if CURRENT.exists(): validate_compose(CURRENT.resolve(), values) print("Host prerequisites and deployment configuration verified") elif args.command in {"status", "logs", "start", "stop", "restart"}: operation = { "status": ["ps"], "logs": ["logs", "--tail", "100"], "start": ["start"], "stop": ["stop"], "restart": ["restart"], }[args.command] print( run( compose() + operation + ([args.service] if args.service else []) ).stdout.decode(), end="", ) elif args.command == "backup": print(backup(args.repository, args.password_file)) elif args.command == "recovery-kit": if args.service != "export" or not args.output: raise ValueError("Specify recovery-kit export") print(export_recovery_kit(args.output)) elif args.command == "restore": if args.release_bundle and offline_root: raise ValueError("Use either --release-bundle or --offline-bundle") release_root = offline_root if args.release_bundle: _, release_root = prepare_release_bundle(args.release_bundle) restore( args.snapshot, args.confirm_restore, args.repository, args.password_file, args.recovery_kit, args.release_bundle, release_root, bool(offline_root), ) elif args.command == "activate": acceptance = STATE / "acceptance.json" if not acceptance.exists(): raise ValueError("No restored deployment is awaiting activation") manifest = validate_manifest( verify( (CURRENT / "manifest.json").read_bytes(), (CURRENT / "manifest.json.sig").read_bytes(), ) ) activate(CURRENT.resolve(), manifest, values, offline=bool(offline_root)) acceptance.unlink() elif args.command == "check-update": print(channel()["version"]) elif args.command in { "install", "configure", "update", "update-staged", "rollback", }: preflight() if args.command == "configure": if not CURRENT.exists(): raise ValueError("Install a release first") root = CURRENT.resolve() manifest = validate_manifest( verify( (root / "manifest.json").read_bytes(), (root / "manifest.json.sig").read_bytes(), ) ) profile = args.profile or values["KAPTRACS_PROFILE"] hostname = args.hostname or values["DEPLOY_HOSTNAME"] if profile not in manifest["supported_profiles"]: raise ValueError("Release does not support this profile") values.update( network_values( profile, hostname, args.bind_address or values.get("KAPTRACS_BIND_ADDRESS"), args.trusted_proxy or values.get("TRUSTED_PROXY_ADDRESSES"), args.accept_trusted_lan_http, ) ) save_network_override(values) save_env(values) activate(root, manifest, values, offline=bool(offline_root)) print("Configuration applied; existing secrets preserved") return if args.command == "rollback": info = json.loads((CONFIG / "installation.json").read_text()) previous = Path(info["previous"]) if info.get("previous") else None if not previous or previous.parent != RELEASES or not previous.is_dir(): raise ValueError("No previous release retained") manifest = validate_manifest( verify( (previous / "manifest.json").read_bytes(), (previous / "manifest.json.sig").read_bytes(), ) ) current_manifest = json.loads((CURRENT / "manifest.json").read_text()) if not current_manifest["database"].get("rollback_schema_compatible"): raise ValueError( "Restore the matching recovery snapshot before rollback" ) backup() values.update( RELEASE_VERSION=manifest["version"], KAPTRACS_IMAGE=manifest["images"]["application"], ) save_env(values) activate(previous, manifest, values) return target = args.version or ( offline_metadata["version"] if offline_root else channel()["version"] ) manifest, root = stage(target, offline_root) if offline_root else stage(target) if values and values["KAPTRACS_PROFILE"] not in manifest["supported_profiles"]: raise ValueError("Release does not support the installed profile") if offline_root and values.get("KAPTRACS_PROFILE") in {"public", "cloudflare"}: raise ValueError( "Public and Cloudflare activation require connectivity and cannot run in explicit offline mode" ) if args.command in {"update", "update-staged"}: if not CURRENT.exists(): raise ValueError("Install a release first") current = version(values["RELEASE_VERSION"]) proposed = version(target) if ( proposed <= current or proposed[0] != current[0] or values["RELEASE_VERSION"] not in manifest["supported_source_versions"] ): raise ValueError( "Unsupported upgrade path; install a supported intermediate release" ) backup() elif values: if values.get("RELEASE_VERSION") != target: raise ValueError("Use update to change an existing release") else: profile = args.profile or ( input("Profile [local/lan/public/reverse-proxy/cloudflare]: ").strip() if sys.stdin.isatty() else "" ) if profile not in manifest["supported_profiles"]: raise ValueError( "Specify a supported --profile for unattended installation" ) hostname, bind_address, trusted_proxy, accept_lan = prompt_install_network( profile, args ) hostname = hostname or ("localhost" if profile == "local" else "") values = { "KAPTRACS_INSTALLATION_ID": "kaptracs-" + secrets.token_hex(6), **network_values( profile, hostname, bind_address, trusted_proxy, accept_lan, ), } for key in [ "POSTGRES_PASSWORD", "SECRET_KEY", "MAIL_TOKEN_SECRET", ]: values[key] = secrets.token_hex(32) (CONFIG / "secrets").mkdir(parents=True, exist_ok=True, mode=0o700) atomic(CONFIG / "secrets" / "backup-password", secrets.token_hex(32)) if profile == "cloudflare": if args.cloudflare_token_file: source = args.cloudflare_token_file details = source.stat() if details.st_mode & 0o077: raise ValueError( "Cloudflare token file must not be group/world accessible" ) token = source.read_text().strip() else: token = getpass.getpass("Cloudflare Tunnel token: ") if not token: raise ValueError("Cloudflare Tunnel token is required") atomic(CONFIG / "secrets" / "cloudflared-token", token) values.update( RELEASE_VERSION=target, KAPTRACS_IMAGE=manifest["images"]["application"] ) save_network_override(values) save_env(values) atomic( CONFIG / "host-agent.env", f"KAPTRACS_OFFLINE={'true' if offline_root else 'false'}\n", ) activate(root, manifest, values, offline=bool(offline_root)) print( "Release ready. Open /setup before exposing the instance to an untrusted network. Back up /etc/kaptracs securely." ) elif args.command == "uninstall": run(compose() + ["down"]) print("Services removed. Persistent volumes, secrets and backups retained.") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "command", choices=[ "install", "configure", "start", "stop", "restart", "status", "logs", "doctor", "backup", "restore", "check-update", "update", "update-staged", "stage-bundle", "rollback", "activate", "recovery-kit", "uninstall", ], ) parser.add_argument("service", nargs="?") parser.add_argument("--version") parser.add_argument("--profile", choices=sorted(PROFILES)) parser.add_argument("--hostname") parser.add_argument("--bind-address") parser.add_argument("--trusted-proxy") parser.add_argument("--accept-trusted-lan-http", action="store_true") parser.add_argument("--offline-bundle", type=Path) parser.add_argument("--cloudflare-token-file", type=Path) parser.add_argument("--snapshot") parser.add_argument("--repository") parser.add_argument("--password-file", type=Path) parser.add_argument("--recovery-kit", type=Path) parser.add_argument("--release-bundle", type=Path) parser.add_argument("--output", type=Path) parser.add_argument("--confirm-restore", action="store_true") args = parser.parse_args() try: os.umask(0o077) with locked(): operation = STATE / "operation.json" if operation.exists(): prior = json.loads(operation.read_text()) if prior.get("state") in {"running", "failed"}: atomic( STATE / "previous-operation.json", {**prior, "interrupted": prior.get("state") == "running"}, ) atomic( STATE / "operation.json", {"command": args.command, "state": "running"} ) try: execute(args) except Exception as exc: atomic( STATE / "operation.json", { "command": args.command, "state": "failed", "stage": json.loads((STATE / "operation.json").read_text()).get( "stage", "preflight" ), "error_type": type(exc).__name__, "recovery": "Inspect status and last-backup.json; restore its snapshot before retrying incompatible rollback.", }, ) raise atomic( STATE / "operation.json", {"command": args.command, "state": "succeeded"}, ) except Exception as exc: # Do not echo captured subprocess commands/output or secret-bearing URLs. print( f"Operation failed ({type(exc).__name__}). See /var/lib/kaptracs/operation.json and the recovery runbook.", file=sys.stderr, ) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())