#!/usr/bin/env python3 """Read-only preflight checks for Coolify Docker Compose deployments.""" from __future__ import annotations import argparse import json from pathlib import Path import subprocess import sys from typing import Any DATABASE_MARKERS = ( "postgres", "mysql", "mariadb", "mongo", "redis", "dragonfly", "keydb", "clickhouse", ) def labels_as_dict(raw: Any) -> dict[str, str]: if isinstance(raw, dict): return {str(key): "" if value is None else str(value) for key, value in raw.items()} result: dict[str, str] = {} if isinstance(raw, list): for item in raw: text = str(item) key, separator, value = text.partition("=") result[key] = value if separator else "" return result def service_networks(raw: Any) -> list[str]: if isinstance(raw, dict): return [str(key) for key in raw] if isinstance(raw, list): return [str(item) for item in raw] return [] def effective_network_name(alias: str, networks: dict[str, Any]) -> str: config = networks.get(alias) if isinstance(config, dict) and config.get("name"): return str(config["name"]) return alias def is_external(config: Any) -> bool: if not isinstance(config, dict): return False external = config.get("external") if isinstance(external, dict): return bool(external) return bool(external) def is_database(service_name: str, service: dict[str, Any]) -> bool: image = str(service.get("image", "")) haystack = f"{service_name} {image}".lower() return any(marker in haystack for marker in DATABASE_MARKERS) def local_path(value: str, base: Path) -> Path | None: if not value or value.startswith(("http://", "https://", "git://", "ssh://")): return None if value.startswith("docker-image://"): return None candidate = Path(value) return candidate if candidate.is_absolute() else base / candidate def load_compose(args: argparse.Namespace) -> tuple[dict[str, Any], Path]: if args.json_input: if args.json_input == "-": return json.load(sys.stdin), Path.cwd() path = Path(args.json_input).resolve() with path.open("r", encoding="utf-8") as handle: return json.load(handle), path.parent if not args.compose_file: raise RuntimeError("a Compose file is required unless --json-input is used") compose_path = Path(args.compose_file).resolve() command = [ "docker", "compose", "-f", str(compose_path), "config", "--format", "json", ] try: completed = subprocess.run( command, check=False, capture_output=True, text=True, timeout=30, ) except FileNotFoundError as exc: raise RuntimeError("docker is not installed or not on PATH") from exc except subprocess.TimeoutExpired as exc: raise RuntimeError("docker compose config timed out after 30 seconds") from exc if completed.returncode != 0: detail = completed.stderr.strip() or completed.stdout.strip() raise RuntimeError(f"docker compose config failed: {detail}") try: return json.loads(completed.stdout), compose_path.parent except json.JSONDecodeError as exc: raise RuntimeError("docker compose config did not return valid JSON") from exc def analyze( document: dict[str, Any], mode: str, public_network: str, base: Path ) -> list[dict[str, str]]: findings: list[dict[str, str]] = [] services = document.get("services") or {} networks = document.get("networks") or {} if not isinstance(services, dict): return [{"severity": "error", "code": "COMPOSE_SERVICES", "message": "services must be a mapping"}] if not isinstance(networks, dict): networks = {} def add(severity: str, code: str, message: str, service: str | None = None) -> None: finding = {"severity": severity, "code": code, "message": message} if service: finding["service"] = service findings.append(finding) non_default = [alias for alias in networks if alias != "default"] public_aliases = [ alias for alias in networks if effective_network_name(alias, networks) == public_network ] if mode == "managed" and non_default: add( "error", "MANAGED_CUSTOM_NETWORK", "Coolify-managed Compose should not define custom networks in the repository; use the Coolify Destination instead: " + ", ".join(sorted(non_default)), ) if mode == "raw": if not public_aliases: add( "error", "PUBLIC_NETWORK_MISSING", f"Raw Compose must reference the existing public network {public_network!r}", ) for alias in public_aliases: if not is_external(networks.get(alias)): add( "error", "PUBLIC_NETWORK_NOT_EXTERNAL", f"network alias {alias!r} resolves to {public_network!r} but is not external", ) for service_name, raw_service in services.items(): if not isinstance(raw_service, dict): add("error", "SERVICE_SHAPE", "service definition must be a mapping", str(service_name)) continue service_name = str(service_name) labels = labels_as_dict(raw_service.get("labels")) aliases = service_networks(raw_service.get("networks")) effective = [effective_network_name(alias, networks) for alias in aliases] traefik_value = labels.get("traefik.enable", "").lower() has_traefik_config = traefik_value == "true" or any( key.startswith(("traefik.http.", "traefik.tcp.", "traefik.udp.")) for key in labels ) selected_network = labels.get("traefik.docker.network") if raw_service.get("network_mode") == "host" and has_traefik_config: add( "error", "TRAEFIK_HOST_NETWORK", "a Traefik-routed service should not use host network mode", service_name, ) if mode == "managed" and selected_network: add( "warning", "MANAGED_NETWORK_LABEL", "a repository-level traefik.docker.network label can conflict with Coolify's generated destination network", service_name, ) if mode == "raw" and has_traefik_config: if public_network not in effective: add( "error", "ROUTED_SERVICE_OFF_PUBLIC", f"Traefik-routed service is not attached to {public_network!r}", service_name, ) if len(effective) > 1 and not selected_network: add( "error", "MULTI_NETWORK_UNPINNED", f"service has multiple networks; set traefik.docker.network={public_network}", service_name, ) if selected_network and selected_network != public_network: add( "error", "TRAEFIK_NETWORK_MISMATCH", f"traefik.docker.network selects {selected_network!r}, expected {public_network!r}", service_name, ) if not any( key.startswith("traefik.http.services.") and key.endswith(".loadbalancer.server.port") for key in labels ): add( "warning", "TRAEFIK_PORT_IMPLICIT", "no explicit HTTP load-balancer container port is configured", service_name, ) if is_database(service_name, raw_service) and public_network in effective: add( "warning", "DATABASE_ON_PUBLIC", f"database/cache service is attached to public network {public_network!r}", service_name, ) if is_database(service_name, raw_service) and raw_service.get("ports"): add( "warning", "DATABASE_HOST_PORT", "database/cache service publishes a host port", service_name, ) elif has_traefik_config and raw_service.get("ports"): add( "info", "ROUTED_HOST_PORT", "service also publishes a host port that bypasses Traefik", service_name, ) build = raw_service.get("build") if build: if isinstance(build, str): context_value = build dockerfile_value = "Dockerfile" inline = False elif isinstance(build, dict): context_value = str(build.get("context", ".")) dockerfile_value = str(build.get("dockerfile", "Dockerfile")) inline = bool(build.get("dockerfile_inline")) else: context_value = "" dockerfile_value = "Dockerfile" inline = False context_path = local_path(context_value, base) if context_path and not context_path.exists(): add( "error", "BUILD_CONTEXT_MISSING", f"build context does not exist: {context_path}", service_name, ) elif context_path and not inline: dockerfile_path = Path(dockerfile_value) if not dockerfile_path.is_absolute(): dockerfile_path = context_path / dockerfile_path if not dockerfile_path.exists(): add( "error", "DOCKERFILE_MISSING", f"Dockerfile does not exist: {dockerfile_path}", service_name, ) return findings def render_text(findings: list[dict[str, str]], mode: str, public_network: str) -> None: print(f"Compose preflight: mode={mode}, public_network={public_network}") if not findings: print("[OK] No findings") return order = {"error": 0, "warning": 1, "info": 2} for finding in sorted(findings, key=lambda item: order.get(item["severity"], 99)): scope = f" service={finding['service']}" if finding.get("service") else "" print(f"[{finding['severity'].upper()}] {finding['code']}{scope}: {finding['message']}") counts = { severity: sum(1 for item in findings if item["severity"] == severity) for severity in ("error", "warning", "info") } print( "Summary: " + ", ".join(f"{count} {severity}" for severity, count in counts.items()) ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("compose_file", nargs="?", help="Compose source file") parser.add_argument("--json-input", help="resolved Compose JSON path, or - for stdin; skips docker compose") parser.add_argument("--mode", choices=("managed", "raw"), required=True) parser.add_argument("--public-network", default="coolify") parser.add_argument("--output", choices=("text", "json"), default="text") parser.add_argument("--strict", action="store_true", help="return 1 when warnings exist") return parser.parse_args() def main() -> int: args = parse_args() try: document, base = load_compose(args) findings = analyze(document, args.mode, args.public_network, base) except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as exc: print(f"compose_preflight: {exc}", file=sys.stderr) return 2 if args.output == "json": print( json.dumps( { "mode": args.mode, "public_network": args.public_network, "findings": findings, }, indent=2, sort_keys=True, ) ) else: render_text(findings, args.mode, args.public_network) if any(item["severity"] == "error" for item in findings): return 2 if args.strict and any(item["severity"] == "warning" for item in findings): return 1 return 0 if __name__ == "__main__": sys.exit(main())