"""Fail-closed production release preflight for a Hayva deployment host. The command prints only control names and safe reasons. It never prints secret values. """ from __future__ import annotations import argparse import ipaddress import json import os import re import shutil import stat import subprocess import sys from datetime import UTC, datetime from pathlib import Path from urllib.parse import unquote, urlparse PLACEHOLDER = re.compile(r"(?:replace|change[-_ ]?me|example|unconfigured)", re.IGNORECASE) REQUIRED_ENVIRONMENT = ( "APP_ENV", "APP_BASE_URL", "HAYVA_TLS_DIR", "COMPUTE_PROVIDER", "COMPUTE_PROVIDER_URL", "DATABASE_URL", "DATABASE_MIGRATION_URL", "POSTGRES_DB", "POSTGRES_USER", "POSTGRES_PASSWORD", "POSTGRES_RUNTIME_USER", "POSTGRES_RUNTIME_PASSWORD", "REDIS_URL", "OPENAI_MODEL", ) REQUIRED_SECRET_FILES = ( "secrets/internal/app-secret", "secrets/internal/ai-service-token", "secrets/internal/computer-service-token", "secrets/internal/computer-capability-private-key", "secrets/internal/computer-capability-public-key", "secrets/internal/compute-provider-token", "secrets/encryption/data-encryption-key", "secrets/providers/openai-api-key", "secrets/redis/users.acl", ) REQUIRED_PROTECTED_FILES = ("secrets/bootstrap/owner_token",) def parse_env_file(path: Path) -> tuple[dict[str, str], list[str]]: values: dict[str, str] = {} errors: list[str] = [] if not path.is_file(): return values, ["The production env file does not exist"] for line_number, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].strip() if "=" not in line: errors.append(f"Environment line {line_number} is malformed") continue key, value = line.split("=", 1) key = key.strip() if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key): errors.append(f"Environment line {line_number} has an invalid key") continue if key in values: errors.append(f"Environment key {key} is duplicated") continue value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: value = value[1:-1] values[key] = value return values, errors def placeholder(value: str | None) -> bool: return not value or bool(PLACEHOLDER.search(value)) def authenticated_url( value: str, *, schemes: set[str], expected_username: str | None = None ) -> bool: parsed = urlparse(value) if parsed.scheme not in schemes or not parsed.hostname or not parsed.username or not parsed.password: return False if placeholder(unquote(parsed.username)) or placeholder(unquote(parsed.password)): return False return expected_username is None or unquote(parsed.username) == expected_username def public_https_url(value: str) -> bool: parsed = urlparse(value) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: return False hostname = parsed.hostname.lower().rstrip(".") if hostname in {"localhost", "localhost.localdomain"} or hostname.endswith(".local"): return False try: address = ipaddress.ip_address(hostname) except ValueError: return "." in hostname return address.is_global def check_environment(values: dict[str, str], root: Path) -> list[str]: errors: list[str] = [] for key in REQUIRED_ENVIRONMENT: if placeholder(values.get(key)): errors.append(f"{key} is missing or contains a placeholder") if values.get("APP_ENV") != "production": errors.append("APP_ENV must be production") if not public_https_url(values.get("APP_BASE_URL", "")): errors.append("APP_BASE_URL must be a public HTTPS origin") if values.get("COMPUTE_PROVIDER") not in { "libvirt_gateway", "proxmox_gateway", "cloud_gateway" }: errors.append("COMPUTE_PROVIDER must select a real VM gateway") provider = urlparse(values.get("COMPUTE_PROVIDER_URL", "")) if provider.scheme != "https" or not provider.hostname or provider.username or provider.password: errors.append("COMPUTE_PROVIDER_URL must be HTTPS and contain no userinfo") runtime_user = values.get("POSTGRES_RUNTIME_USER") migration_user = values.get("POSTGRES_USER") if runtime_user == migration_user: errors.append("Migration and runtime PostgreSQL identities must differ") runtime_database_url = values.get("DATABASE_URL", "") migration_database_url = values.get("DATABASE_MIGRATION_URL", "") if not authenticated_url( runtime_database_url, schemes={"postgresql+asyncpg"}, expected_username=runtime_user, ): errors.append("DATABASE_URL must use the authenticated runtime PostgreSQL identity") if not authenticated_url( migration_database_url, schemes={"postgresql+asyncpg"}, expected_username=migration_user, ): errors.append("DATABASE_MIGRATION_URL must use the migration PostgreSQL identity") if not authenticated_url(values.get("REDIS_URL", ""), schemes={"redis", "rediss"}): errors.append("REDIS_URL must use an authenticated Redis identity") runtime_database = urlparse(runtime_database_url) migration_database = urlparse(migration_database_url) if unquote(runtime_database.password or "") != values.get("POSTGRES_RUNTIME_PASSWORD"): errors.append("DATABASE_URL does not match POSTGRES_RUNTIME_PASSWORD") if unquote(migration_database.password or "") != values.get("POSTGRES_PASSWORD"): errors.append("DATABASE_MIGRATION_URL does not match POSTGRES_PASSWORD") tls_value = values.get("HAYVA_TLS_DIR", "") tls_dir = Path(tls_value) if not tls_dir.is_absolute(): errors.append("HAYVA_TLS_DIR must be an absolute protected directory") else: for filename in ("fullchain.pem", "privkey.pem"): if not (tls_dir / filename).is_file(): errors.append(f"TLS material {filename} is missing") for relative in REQUIRED_SECRET_FILES: path = root / relative if not path.is_file(): errors.append(f"Required secret file is missing: {relative}") continue try: value = path.read_text(encoding="utf-8").strip() except (OSError, UnicodeError): errors.append(f"Required secret file cannot be read safely: {relative}") continue if placeholder(value) or len(value) < 16: errors.append(f"Required secret file is empty or a placeholder: {relative}") if os.name != "nt" and stat.S_IMODE(path.stat().st_mode) & 0o077: errors.append(f"Secret file permissions are too broad: {relative}") for relative in REQUIRED_PROTECTED_FILES: path = root / relative if not path.is_file(): errors.append(f"Required protected file is missing: {relative}") elif os.name != "nt" and stat.S_IMODE(path.stat().st_mode) & 0o077: errors.append(f"Protected file permissions are too broad: {relative}") redis_url = urlparse(values.get("REDIS_URL", "")) redis_acl = root / "secrets/redis/users.acl" if redis_acl.is_file() and redis_url.username and redis_url.password: expected_acl = f"user {unquote(redis_url.username)} on >{unquote(redis_url.password)}" try: acl_text = redis_acl.read_text(encoding="utf-8") except (OSError, UnicodeError): acl_text = "" if expected_acl not in acl_text: errors.append("REDIS_URL credentials do not match the protected Redis ACL") return errors def run_checked(command: list[str], *, root: Path) -> str | None: try: result = subprocess.run( command, cwd=root, check=False, capture_output=True, text=True, timeout=120 ) except (OSError, subprocess.TimeoutExpired): return f"Command failed or timed out: {command[0]}" if result.returncode != 0: return f"Command failed: {' '.join(command[:3])}" return None def check_repository(root: Path) -> tuple[list[str], str | None]: errors: list[str] = [] if not (root / ".git").exists(): return ["A Git checkout is required to bind evidence to an exact revision"], None revision_result = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=root, check=False, capture_output=True, text=True ) revision = revision_result.stdout.strip() if revision_result.returncode == 0 else None if not revision: errors.append("The release has no resolvable Git revision") status = subprocess.run( ["git", "status", "--porcelain"], cwd=root, check=False, capture_output=True, text=True ) if status.returncode != 0 or status.stdout.strip(): errors.append("The release worktree must be clean") remotes = subprocess.run( ["git", "remote"], cwd=root, check=False, capture_output=True, text=True ) if remotes.returncode != 0 or not remotes.stdout.strip(): errors.append("The release revision has no configured source remote") return errors, revision def check_runtime(root: Path, env_file: Path, tls_dir: Path) -> list[str]: errors: list[str] = [] for executable in ("docker", "age", "openssl"): if not shutil.which(executable): errors.append(f"Required release executable is unavailable: {executable}") if errors: return errors for command in ( ["docker", "info"], ["docker", "compose", "version"], [ "docker", "compose", "--env-file", str(env_file), "-f", "docker-compose.yml", "-f", "docker-compose.production.yml", "config", "--quiet", ], [ "openssl", "x509", "-checkend", "2592000", "-noout", "-in", str(tls_dir / "fullchain.pem"), ], ): error = run_checked(command, root=root) if error: errors.append(error) return errors def write_evidence(path: Path, *, revision: str) -> None: payload = { "schema": "hayva-production-preflight-v1", "status": "passed", "revision": revision, "checked_at": datetime.now(UTC).isoformat(), "checks": [ "configuration", "secret-files", "tls-expiry", "git-revision", "docker-engine", "compose-render", ], } path.parent.mkdir(parents=True, exist_ok=True) descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(payload, handle, indent=2) handle.write("\n") def main() -> int: parser = argparse.ArgumentParser(description="Verify fail-closed Hayva production prerequisites") parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--env-file", type=Path, default=Path(".env")) parser.add_argument("--evidence", type=Path) args = parser.parse_args() root = args.root.resolve() env_file = args.env_file if args.env_file.is_absolute() else root / args.env_file values, errors = parse_env_file(env_file) errors.extend(check_environment(values, root)) repository_errors, revision = check_repository(root) errors.extend(repository_errors) tls_dir = Path(values.get("HAYVA_TLS_DIR", "")) errors.extend(check_runtime(root, env_file, tls_dir)) if errors: print("Hayva production preflight: NOT READY", file=sys.stderr) for error in sorted(set(errors)): print(f"- {error}", file=sys.stderr) return 1 if args.evidence: evidence = args.evidence if args.evidence.is_absolute() else root / args.evidence write_evidence(evidence, revision=revision or "unavailable") print(f"Hayva production preflight: PASSED ({revision})") return 0 if __name__ == "__main__": raise SystemExit(main())