"""Command-line interface for GovOPlaN deployment reconciliation.""" from __future__ import annotations import argparse from contextlib import contextmanager from dataclasses import replace from datetime import UTC, datetime import fcntl import getpass import json import os from pathlib import Path import shutil import subprocess import sys import time from typing import Iterator, Mapping, Sequence from urllib.error import URLError from urllib.request import urlopen from .bundle import ( atomic_write, bundle_paths, canonical_json, digest_json, environment_fingerprint, ensure_private_directory, initial_secrets, read_env, reconcile_runtime_environment, render_compose, render_garage_config, render_load_balancer_config, service_names, write_env, ) from .model import ( ComponentConfig, DEFAULT_GARAGE_IMAGE, DEFAULT_LOAD_BALANCER_IMAGE, InstallationSpec, ListenConfig, ReplicaConfig, SpecError, default_spec, load_spec, parse_spec, ) from .planning import DeploymentPlan, build_plan def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="govoplan-deploy", description=( "Create, validate, and reconcile a declarative GovOPlaN Compose deployment." ), ) subparsers = parser.add_subparsers(dest="command", required=True) init = subparsers.add_parser( "init", help="Create a new private installation bundle." ) _directory_argument(init) _configuration_arguments(init) init.add_argument( "--non-interactive", action="store_true", help="Use flags/defaults without prompting.", ) configure = subparsers.add_parser( "configure", help="Change installation choices while preserving generated secrets.", ) _directory_argument(configure) _configuration_arguments(configure, defaults=False) render = subparsers.add_parser( "render", help="Regenerate Compose and the deployment plan without applying." ) _directory_argument(render) render.add_argument("--json", action="store_true", help="Print the plan as JSON.") doctor = subparsers.add_parser( "doctor", help="Run specification, secret, host, and provenance checks." ) _directory_argument(doctor) doctor.add_argument("--json", action="store_true", help="Print the plan as JSON.") apply_parser = subparsers.add_parser( "apply", help="Apply an allowed plan with Docker Compose." ) _directory_argument(apply_parser) apply_parser.add_argument( "--allow-unverified-images", action="store_true", help="Allow mutable/unverified images for an evaluation profile only.", ) apply_parser.add_argument( "--skip-pull", action="store_true", help="Do not pull images before reconciliation.", ) apply_parser.add_argument( "--health-timeout-seconds", type=float, default=120.0, help="Maximum time to wait for the public health endpoint.", ) status = subparsers.add_parser( "status", help="Show desired state and current Compose process state." ) _directory_argument(status) status.add_argument("--json", action="store_true", help="Print JSON.") return parser def _directory_argument(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--directory", type=Path, default=Path.home() / ".local" / "share" / "govoplan" / "installations" / "default", help="Private installation state directory.", ) def _configuration_arguments( parser: argparse.ArgumentParser, *, defaults: bool = True ) -> None: default = (lambda value: value) if defaults else (lambda _value: None) parser.add_argument("--installation-id", default=default("govoplan-local")) parser.add_argument( "--profile", choices=("evaluation", "self-hosted"), default=default("evaluation"), ) parser.add_argument("--public-url", default=default("http://127.0.0.1:8080")) parser.add_argument("--listen-address", default=default("127.0.0.1")) parser.add_argument("--listen-port", type=int, default=default(8080)) parser.add_argument( "--postgres", choices=("managed", "external"), default=default("managed"), ) parser.add_argument( "--database-url", help="External PostgreSQL URL; stored privately." ) parser.add_argument( "--redis", choices=("managed", "external", "disabled"), default=default("managed"), ) parser.add_argument("--redis-url", help="External Redis URL; stored privately.") parser.add_argument( "--mail", choices=("disabled", "external-relay", "test-mail"), default=default("disabled"), ) parser.add_argument( "--storage", choices=("local", "garage", "s3"), default=default("local"), ) parser.add_argument( "--garage-image", default=default(DEFAULT_GARAGE_IMAGE), help="Garage image used by managed object storage.", ) parser.add_argument("--s3-endpoint-url", help="S3-compatible endpoint URL.") parser.add_argument("--s3-region", help="S3 region.") parser.add_argument( "--s3-access-key-id", help="S3 access key id; stored privately." ) parser.add_argument( "--s3-secret-access-key", help="S3 secret access key; stored privately.", ) parser.add_argument("--s3-bucket", help="S3 bucket for managed files.") parser.add_argument( "--load-balancer-image", default=default(DEFAULT_LOAD_BALANCER_IMAGE), help="HAProxy image used by the managed local load balancer.", ) parser.add_argument( "--api-replicas", type=int, default=default(1), help="API containers on this Compose host (1-64).", ) parser.add_argument( "--web-replicas", type=int, default=default(1), help="WebUI containers on this Compose host (1-64).", ) parser.add_argument( "--worker-replicas", type=int, default=None, help="Worker containers on this Compose host (1-128, or 0 without Redis).", ) parser.add_argument( "--module-set", choices=("core", "base", "full"), default=default("base"), ) parser.add_argument("--api-image", default=default("govoplan-api:unpublished")) parser.add_argument("--web-image", default=default("govoplan-web:unpublished")) parser.add_argument("--release-version", default=default("unpublished")) parser.add_argument("--release-channel", default=default("stable")) parser.add_argument("--manifest-url", default=default("")) parser.add_argument("--manifest-sha256", default=default("")) def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "init": return _init(args) if args.command == "configure": return _configure(args) if args.command in {"render", "doctor"}: return _render_or_doctor(args) if args.command == "apply": return _apply(args) if args.command == "status": return _status(args) except (SpecError, ValueError, OSError, subprocess.SubprocessError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 raise RuntimeError(f"unsupported command: {args.command}") def _init(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) ensure_private_directory(paths.root) if paths.spec.exists(): raise ValueError( f"{paths.spec} already exists; use configure for an existing installation" ) if not args.non_interactive and sys.stdin.isatty(): _prompt_configuration(args) spec = default_spec( installation_id=args.installation_id, profile=args.profile, public_url=args.public_url, listen_address=args.listen_address, listen_port=args.listen_port, postgres_mode=args.postgres, redis_mode=args.redis, mail_mode=args.mail, storage_mode=args.storage, garage_image=args.garage_image, load_balancer_image=args.load_balancer_image, api_replicas=args.api_replicas, web_replicas=args.web_replicas, worker_replicas=args.worker_replicas, module_set=args.module_set, api_image=args.api_image, web_image=args.web_image, manifest_url=args.manifest_url, manifest_sha256=args.manifest_sha256, version=args.release_version, channel=args.release_channel, ) supplied = _supplied_secret_values(args) secrets = initial_secrets(spec, supplied=supplied) _write_bundle(spec, paths, secrets) plan = build_plan(spec, paths, include_host_checks=False) _write_plan(paths.plan, plan) print(f"Created GovOPlaN installation bundle at {paths.root}") print(f"Edit choices with: govoplan-deploy configure --directory {paths.root}") print(f"Check readiness with: govoplan-deploy doctor --directory {paths.root}") if any(check.level == "error" for check in plan.checks): print("The bundle is not apply-ready yet; run doctor for the blocking checks.") return 0 def _configure(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) ensure_private_directory(paths.root) current = load_spec(paths.spec) if args.installation_id and args.installation_id != current.installation_id: raise ValueError( "installation_id is immutable; create a separate installation " "instead of renaming a Compose project" ) spec = _updated_spec(current, args) if ( current.components.postgres.mode == "managed" and spec.components.postgres.mode == "external" and not args.database_url ): raise ValueError( "switching PostgreSQL from managed to external requires --database-url" ) if ( current.components.redis.mode != "external" and spec.components.redis.mode == "external" and not args.redis_url ): raise ValueError("switching Redis to external requires --redis-url") if current.components.storage.mode != "s3" and spec.components.storage.mode == "s3": missing_s3_options = [ flag for flag, value in ( ("--s3-endpoint-url", args.s3_endpoint_url), ("--s3-region", args.s3_region), ("--s3-access-key-id", args.s3_access_key_id), ("--s3-secret-access-key", args.s3_secret_access_key), ("--s3-bucket", args.s3_bucket), ) if not value ] if missing_s3_options: raise ValueError( "switching to external S3 requires: " + ", ".join(missing_s3_options) ) secrets = read_env(paths.env) secrets.update(_supplied_secret_values(args)) secrets = reconcile_runtime_environment(spec, secrets) _write_bundle(spec, paths, secrets) plan = build_plan(spec, paths, include_host_checks=False) _write_plan(paths.plan, plan) print(f"Updated desired state at {paths.root}") _print_plan(plan) return 0 def _render_or_doctor(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) spec = load_spec(paths.spec) secrets = reconcile_runtime_environment(spec, read_env(paths.env)) _write_bundle(spec, paths, secrets) plan = build_plan( spec, paths, include_host_checks=args.command == "doctor", ) _write_plan(paths.plan, plan) if args.json: print(json.dumps(plan.to_dict(), indent=2, sort_keys=True)) else: _print_plan(plan) return 1 if plan.blocked else 0 def _apply(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) spec = load_spec(paths.spec) if args.allow_unverified_images and spec.profile != "evaluation": raise ValueError( "--allow-unverified-images is restricted to evaluation installations" ) ensure_private_directory(paths.root) with _deployment_lock(paths.lock): secrets = reconcile_runtime_environment(spec, read_env(paths.env)) _write_bundle(spec, paths, secrets) plan = build_plan(spec, paths, include_host_checks=True) _write_plan(paths.plan, plan) effective_errors = [ check for check in plan.checks if check.level == "error" and not ( args.allow_unverified_images and check.id.startswith( ( "release.api_image.", "release.web_image.", "components.postgres.image.", "components.redis.image.", "components.mail.image.", "components.storage.image.", "components.load_balancer.image.", "release.manifest", "modules.image_composition", ) ) ) ] if effective_errors: _print_plan(plan) raise ValueError("deployment plan is blocked; resolve doctor errors first") docker = shutil.which("docker") if docker is None: raise ValueError("Docker CLI is required for apply") compose = [ docker, "compose", "--env-file", str(paths.env), "--project-name", spec.installation_id, "--file", str(paths.compose), ] if not args.skip_pull: _run([*compose, "pull"], cwd=paths.root) dependencies = [ name for name in ("postgres", "redis", "garage", "test-mail") if name in service_names(spec) ] if dependencies: _run([*compose, "up", "--detach", *dependencies], cwd=paths.root) _run([*compose, "run", "--rm", "migrate"], cwd=paths.root) if _receipt_uses_direct_web_port(paths.receipt): _run([*compose, "stop", "web"], cwd=paths.root) runtime_services = [ name for name in ("api", "web", "load-balancer", "worker", "scheduler") if name in service_names(spec) ] _run( [*compose, "up", "--detach", "--remove-orphans", *runtime_services], cwd=paths.root, ) _wait_for_health( f"{spec.public_url}/health", timeout_seconds=args.health_timeout_seconds, ) receipt = { "schema_version": 1, "installation_id": spec.installation_id, "applied_at": _now(), "spec_sha256": digest_json(spec.to_dict()), "compose_sha256": digest_json(render_compose(spec)), "environment_fingerprint": environment_fingerprint(secrets), "release": { "channel": spec.release.channel, "version": spec.release.version, "manifest_sha256": spec.release.manifest_sha256, "api_image": spec.release.api_image, "web_image": spec.release.web_image, }, "services": list(service_names(spec)), "replicas": { "api": spec.replicas.api, "web": spec.replicas.web, "worker": spec.replicas.worker, }, "listen": { "address": spec.listen.address, "port": spec.listen.port, }, "management": { "mode": "govoplan-deploy", "agent": "cli", "web_updates": False, }, } atomic_write(paths.receipt, canonical_json(receipt), mode=0o600) print(f"GovOPlaN is ready at {spec.public_url}") return 0 def _status(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) spec = load_spec(paths.spec) plan = build_plan(spec, paths, include_host_checks=False) docker = shutil.which("docker") processes: object = [] process_error = "" if docker: result = subprocess.run( [ docker, "compose", "--env-file", str(paths.env), "--project-name", spec.installation_id, "--file", str(paths.compose), "ps", "--format", "json", ], cwd=paths.root, check=False, capture_output=True, text=True, timeout=15, ) if result.returncode == 0: processes = _parse_compose_ps(result.stdout) else: process_error = result.stderr.strip() or result.stdout.strip() else: process_error = "Docker CLI is unavailable." payload = { "installation_id": spec.installation_id, "public_url": spec.public_url, "plan": plan.to_dict(), "processes": processes, "process_error": process_error, } if args.json: print(json.dumps(payload, indent=2, sort_keys=True)) else: _print_plan(plan) if process_error: print(f"Runtime: {process_error}") elif isinstance(processes, list): print(f"Runtime: {len(processes)} Compose process(es) reported.") return 0 def _updated_spec( current: InstallationSpec, args: argparse.Namespace ) -> InstallationSpec: module_set = getattr(args, "module_set", None) if module_set: modules = default_spec(module_set=module_set).enabled_modules else: modules = current.enabled_modules release = replace( current.release, channel=args.release_channel or current.release.channel, version=args.release_version or current.release.version, manifest_url=( args.manifest_url if args.manifest_url is not None else current.release.manifest_url ), manifest_sha256=( args.manifest_sha256 if args.manifest_sha256 is not None else current.release.manifest_sha256 ), api_image=args.api_image or current.release.api_image, web_image=args.web_image or current.release.web_image, ) storage_mode = args.storage or current.components.storage.mode components = ComponentConfig( postgres=replace( current.components.postgres, mode=args.postgres or current.components.postgres.mode, ), redis=replace( current.components.redis, mode=args.redis or current.components.redis.mode, ), mail=replace( current.components.mail, mode=args.mail or current.components.mail.mode, ), storage=replace( current.components.storage, mode=storage_mode, image=( args.garage_image or current.components.storage.image or DEFAULT_GARAGE_IMAGE ) if storage_mode == "garage" else "", ), load_balancer=replace( current.components.load_balancer, image=(args.load_balancer_image or current.components.load_balancer.image), ), ) replicas = ReplicaConfig( api=( args.api_replicas if args.api_replicas is not None else current.replicas.api ), web=( args.web_replicas if args.web_replicas is not None else current.replicas.web ), worker=( args.worker_replicas if args.worker_replicas is not None else ( 0 if components.redis.mode == "disabled" else max(current.replicas.worker, 1) ) ), ) value = replace( current, installation_id=args.installation_id or current.installation_id, profile=args.profile or current.profile, public_url=args.public_url or current.public_url, listen=ListenConfig( address=args.listen_address or current.listen.address, port=args.listen_port or current.listen.port, ), release=release, components=components, replicas=replicas, enabled_modules=modules, ) return parse_spec(value.to_dict()) def _supplied_secret_values(args: argparse.Namespace) -> dict[str, str]: values: dict[str, str] = {} if getattr(args, "database_url", None): values["DATABASE_URL"] = args.database_url values["GOVOPLAN_DATABASE_URL_PGTOOLS"] = _pgtools_url(args.database_url) if getattr(args, "redis_url", None): values["REDIS_URL"] = args.redis_url s3_values = { "FILE_STORAGE_S3_ENDPOINT_URL": getattr(args, "s3_endpoint_url", None), "FILE_STORAGE_S3_REGION": getattr(args, "s3_region", None), "FILE_STORAGE_S3_ACCESS_KEY_ID": getattr(args, "s3_access_key_id", None), "FILE_STORAGE_S3_SECRET_ACCESS_KEY": getattr( args, "s3_secret_access_key", None ), "FILE_STORAGE_S3_BUCKET": getattr(args, "s3_bucket", None), } values.update({key: value for key, value in s3_values.items() if value}) return values def _pgtools_url(database_url: str) -> str: if database_url.startswith("postgresql+psycopg://"): return "postgresql://" + database_url.removeprefix("postgresql+psycopg://") return database_url def _write_bundle( spec: InstallationSpec, paths, secrets: Mapping[str, str], ) -> None: ensure_private_directory(paths.root) atomic_write(paths.spec, canonical_json(spec.to_dict()), mode=0o600) write_env(paths.env, secrets) atomic_write(paths.compose, canonical_json(render_compose(spec)), mode=0o600) atomic_write( paths.load_balancer_config, render_load_balancer_config(spec).encode("utf-8"), mode=0o644, ) atomic_write( paths.garage_config, render_garage_config().encode("utf-8"), mode=0o644, ) def _write_plan(path: Path, plan: DeploymentPlan) -> None: atomic_write(path, canonical_json(plan.to_dict()), mode=0o600) def _print_plan(plan: DeploymentPlan) -> None: state = "BLOCKED" if plan.blocked else "READY" print(f"GovOPlaN deployment plan: {state}") for action in plan.actions: print(f" {action.action:12} {action.target}: {action.detail}") for check in plan.checks: prefix = {"ok": "OK", "warning": "WARN", "error": "ERROR"}.get( check.level, check.level.upper() ) print(f" [{prefix}] {check.message}") if check.action and check.level != "ok": print(f" {check.action}") def _prompt_configuration(args: argparse.Namespace) -> None: args.profile = _prompt_choice( "Installation profile", args.profile, ("evaluation", "self-hosted") ) if args.profile == "self-hosted" and args.public_url.startswith("http://"): args.public_url = "https://govoplan.example.org" args.public_url = _prompt("Public URL", args.public_url) args.postgres = _prompt_choice("PostgreSQL", args.postgres, ("managed", "external")) if args.postgres == "external" and not args.database_url: args.database_url = getpass.getpass( "External PostgreSQL URL (input hidden): " ).strip() redis_choices = ( ("managed", "external") if args.profile == "self-hosted" else ("managed", "external", "disabled") ) args.redis = _prompt_choice("Redis", args.redis, redis_choices) if args.redis == "external" and not args.redis_url: args.redis_url = getpass.getpass("External Redis URL (input hidden): ").strip() mail_choices = ( ("disabled", "external-relay") if args.profile == "self-hosted" else ("disabled", "external-relay", "test-mail") ) args.mail = _prompt_choice( "Mail integration", args.mail, mail_choices, ) args.storage = _prompt_choice( "File storage", args.storage, ("local", "garage", "s3"), ) if args.storage == "s3": args.s3_endpoint_url = args.s3_endpoint_url or _prompt( "S3 endpoint URL", "https://s3.example.org" ) args.s3_region = args.s3_region or _prompt("S3 region", "eu-central-1") args.s3_access_key_id = args.s3_access_key_id or _prompt("S3 access key id", "") args.s3_secret_access_key = ( args.s3_secret_access_key or getpass.getpass("S3 secret access key (input hidden): ").strip() ) args.s3_bucket = args.s3_bucket or _prompt("S3 bucket", "govoplan-files") args.api_replicas = _prompt_integer( "API replicas on this host", args.api_replicas, minimum=1, maximum=64, ) args.web_replicas = _prompt_integer( "WebUI replicas on this host", args.web_replicas, minimum=1, maximum=64, ) if args.redis != "disabled": args.worker_replicas = _prompt_integer( "Worker replicas on this host", args.worker_replicas or 1, minimum=1, maximum=128, ) else: args.worker_replicas = 0 args.module_set = _prompt_choice( "Initial module set", args.module_set, ("core", "base", "full") ) def _prompt(label: str, default: str) -> str: value = input(f"{label} [{default}]: ").strip() return value or default def _prompt_choice(label: str, default: str, choices: tuple[str, ...]) -> str: while True: value = _prompt(f"{label} ({'/'.join(choices)})", default) if value in choices: return value print(f"Choose one of: {', '.join(choices)}") def _prompt_integer( label: str, default: int, *, minimum: int, maximum: int, ) -> int: while True: raw = _prompt(label, str(default)) try: value = int(raw) except ValueError: value = minimum - 1 if minimum <= value <= maximum: return value print(f"Choose a number between {minimum} and {maximum}.") @contextmanager def _deployment_lock(path: Path) -> Iterator[None]: descriptor = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) try: try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: raise ValueError( "another deployment operation owns the installation lock" ) from exc os.ftruncate(descriptor, 0) os.write(descriptor, f"{os.getpid()}\n".encode("ascii")) os.fsync(descriptor) yield finally: fcntl.flock(descriptor, fcntl.LOCK_UN) os.close(descriptor) def _run(argv: Sequence[str], *, cwd: Path) -> None: result = subprocess.run(list(argv), cwd=cwd, check=False) if result.returncode != 0: raise subprocess.CalledProcessError(result.returncode, list(argv)) def _wait_for_health(url: str, *, timeout_seconds: float) -> None: deadline = time.monotonic() + max(timeout_seconds, 1.0) last_error = "health endpoint did not answer" while time.monotonic() < deadline: try: with urlopen(url, timeout=3) as response: # noqa: S310 - URL originates in the validated installation specification. if 200 <= response.status < 300: return last_error = f"health endpoint returned HTTP {response.status}" except (OSError, URLError) as exc: last_error = str(exc) time.sleep(2) raise ValueError(f"GovOPlaN did not become healthy: {last_error}") def _parse_compose_ps(output: str) -> list[object]: stripped = output.strip() if not stripped: return [] try: value = json.loads(stripped) except json.JSONDecodeError: rows = [] for line in stripped.splitlines(): try: rows.append(json.loads(line)) except json.JSONDecodeError: return [{"raw": stripped}] return rows return value if isinstance(value, list) else [value] def _receipt_uses_direct_web_port(path: Path) -> bool: if not path.exists(): return False try: receipt = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False services = receipt.get("services") if isinstance(receipt, dict) else None return ( isinstance(services, list) and "web" in services and "load-balancer" not in services ) def _now() -> str: return ( datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") )