feat: Add script to generate Caddy config from Docker container labels

docker_to_caddy.py

@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = []
+# ///
+
+import argparse
+import ipaddress
+import json
+import logging
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+from typing import Any
+
+
+class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter):
+    pass
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Generate Caddy reverse proxies from Docker container labels.",
+        formatter_class=HelpFormatter,
+        epilog="""Compose service labels:
+  labels:
+    caddy.domain: ascii.app.abdus.dev
+    caddy.port: "3000"
+    caddy.basic_auth: "true"
+
+caddy.port selects the internal TCP port, not the published host port.
+Omit it when exactly one TCP port is published. Only running labeled
+containers generate routes. Potential Docker port clashes warn on stderr.
+caddy.basic_auth accepts true or false (default). When true, the route
+imports the (auth) snippet defined in your handwritten Caddy config.
+
+Local usage:
+  python3 docker_to_caddy.py
+  python3 docker_to_caddy.py --output docker.generated.caddy
+
+Remote Docker over SSH:
+  env -u DOCKER_CONTEXT DOCKER_HOST=ssh://abdus@your-server docker ps
+  env -u DOCKER_CONTEXT DOCKER_HOST=ssh://abdus@your-server \\
+    python3 docker_to_caddy.py --output docker.generated.caddy
+
+The SSH user needs Docker socket access on the server. These commands do
+not change your selected Docker context (for example, OrbStack). Output is
+written locally; loopback upstreams refer to the server where Caddy runs.
+
+Caddy setup:
+  Add a top-level import docker.generated.caddy to your handwritten config.
+  Remove migrated proxy blocks, then validate and reload Caddy manually.
+  The script does not configure DNS or reload Caddy.
+""",
+    )
+    parser.add_argument("--output", type=Path, help="Output file; otherwise write to stdout")
+    return parser.parse_args()
+
+
+def docker_json(arguments: list[str]) -> list[Any]:
+    result = subprocess.run(
+        args=["docker", *arguments], check=True, capture_output=True, text=True,
+        timeout=60,
+    )
+    value = json.loads(result.stdout)
+    if not isinstance(value, list):
+        raise ValueError("Docker returned an unexpected response")
+    return value
+
+
+def inspect_containers() -> list[dict[str, Any]]:
+    result = subprocess.run(
+        args=["docker", "container", "ls", "--all", "--quiet"],
+        check=True, capture_output=True, text=True, timeout=60,
+    )
+    identifiers = result.stdout.split()
+    containers = []
+    for offset in range(0, len(identifiers), 100):
+        containers.extend(docker_json(arguments=[
+            "container", "inspect", *identifiers[offset:offset + 100],
+        ]))
+    return containers
+
+
+def container_name(container: dict[str, Any]) -> str:
+    return str(container.get("Name") or container["Id"]).lstrip("/")
+
+
+def port_number(value: str) -> int:
+    if not re.fullmatch(r"[0-9]+", value) or not 1 <= int(value) <= 65535:
+        raise ValueError(f"Invalid port: {value!r}")
+    return int(value)
+
+
+def normalize_domain(value: str) -> str:
+    domain = value.lower().rstrip(".")
+    labels = domain.split(".")
+    if len(domain) > 253 or len(labels) < 2 or any(
+        not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label)
+        for label in labels
+    ):
+        raise ValueError(f"Invalid domain: {value!r}; use a single DNS hostname")
+    return domain
+
+
+def upstream(binding: dict[str, str]) -> tuple[int, str, int]:
+    address = ipaddress.ip_address(binding.get("HostIp") or "0.0.0.0")
+    if address.is_unspecified:
+        address = ipaddress.ip_address("127.0.0.1" if address.version == 4 else "::1")
+    port = port_number(value=binding["HostPort"])
+    return address.version, str(address), port
+
+
+def render_config(containers: list[dict[str, Any]]) -> str:
+    routes: dict[str, tuple[str, str, bool]] = {}
+    for container in containers:
+        labels = container.get("Config", {}).get("Labels") or {}
+        if "caddy.domain" not in labels or not container.get("State", {}).get("Running"):
+            continue
+        name = container_name(container=container)
+        try:
+            domain = normalize_domain(value=labels["caddy.domain"])
+            auth_label = labels.get("caddy.basic_auth", "false")
+            if not isinstance(auth_label, str) or auth_label.lower() not in ("true", "false"):
+                raise ValueError("caddy.basic_auth must be true or false")
+            basic_auth = auth_label.lower() == "true"
+            if container.get("HostConfig", {}).get("NetworkMode") == "host":
+                raise ValueError("Host networking is unsupported; publish a TCP port instead")
+            ports = {
+                key: [binding for binding in bindings or [] if binding.get("HostPort")]
+                for key, bindings in (container.get("NetworkSettings", {}).get("Ports") or {}).items()
+                if key.endswith("/tcp")
+            }
+            ports = {key: bindings for key, bindings in ports.items() if bindings}
+            if "caddy.port" in labels:
+                key = f"{port_number(value=labels['caddy.port'])}/tcp"
+                if key not in ports:
+                    raise ValueError(f"Selected container port {key} has no published host port")
+            elif len(ports) == 1:
+                key = next(iter(ports))
+            else:
+                raise ValueError("Expected one published TCP port; set caddy.port to select one")
+            version, address, port = min(upstream(binding=binding) for binding in ports[key])
+            target = f"[{address}]:{port}" if version == 6 else f"{address}:{port}"
+            if domain in routes:
+                raise ValueError(f"Domain {domain} is also claimed by {routes[domain][0]}")
+            routes[domain] = (name, target, basic_auth)
+        except (ValueError, KeyError, TypeError) as error:
+            raise ValueError(f"Container {name}: {error}") from error
+    blocks = ["# Generated by docker_to_caddy.py; do not edit.\n"]
+    for domain, (name, target, basic_auth) in sorted(routes.items()):
+        safe_name = re.sub(r"[^a-zA-Z0-9_.-]", "_", name)
+        auth_import = "    import auth\n" if basic_auth else ""
+        blocks.append(
+            f"# Container: {safe_name}\n{domain} {{\n{auth_import}    reverse_proxy {target}\n}}\n"
+        )
+    return "\n".join(blocks)
+
+
+def addresses_overlap(first: str, second: str) -> bool:
+    left = ipaddress.ip_address(first or "0.0.0.0")
+    right = ipaddress.ip_address(second or "0.0.0.0")
+    if left.version != right.version:
+        return (left.version == 6 and left.is_unspecified) or (
+            right.version == 6 and right.is_unspecified
+        )
+    return left == right or left.is_unspecified or right.is_unspecified
+
+
+def warn_port_clashes(containers: list[dict[str, Any]]) -> None:
+    bindings: list[tuple[str, str, str, int, str]] = []
+    for container in containers:
+        name = container_name(container=container)
+        state = container.get("State", {}).get("Status", "unknown")
+        configured = container.get("HostConfig", {}).get("PortBindings") or {}
+        actual = container.get("NetworkSettings", {}).get("Ports") or {}
+        seen: set[tuple[str, int, str]] = set()
+        for source in (configured, actual):
+            for container_port, entries in source.items():
+                protocol = container_port.rsplit("/", 1)[-1]
+                for binding in entries or []:
+                    value = binding.get("HostPort")
+                    if not value or value == "0":
+                        continue
+                    try:
+                        port = port_number(value=value)
+                        address = str(ipaddress.ip_address(binding.get("HostIp") or "0.0.0.0"))
+                    except ValueError as error:
+                        logging.warning("Cannot check binding for %s: %s", name, error)
+                        continue
+                    identity = (address, port, protocol)
+                    if identity not in seen:
+                        seen.add(identity)
+                        bindings.append((name, state, address, port, protocol))
+    warned: set[tuple[str, str, int, str]] = set()
+    for index, first in enumerate(bindings):
+        for second in bindings[index + 1:]:
+            if first[0] == second[0] or first[3:] != second[3:]:
+                continue
+            if not addresses_overlap(first=first[2], second=second[2]):
+                continue
+            identity = (*sorted((first[0], second[0])), first[3], first[4])
+            if identity in warned:
+                continue
+            warned.add(identity)
+            logging.warning(
+                "Potential port clash: %s (%s) [%s]:%s/%s and %s (%s) [%s]:%s/%s",
+                *first, *second,
+            )
+
+
+def write_config(config: str, output: Path) -> None:
+    output = output.absolute()
+    with tempfile.TemporaryDirectory(dir=output.parent, prefix=".docker-to-caddy-") as directory:
+        temporary = Path(directory) / "config"
+        temporary.write_text(data=config, encoding="utf-8")
+        if output.exists():
+            temporary.chmod(mode=output.stat().st_mode & 0o777)
+        os.replace(src=temporary, dst=output)
+
+
+def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+    arguments = parse_args()
+    try:
+        containers = inspect_containers()
+        warn_port_clashes(containers=containers)
+        config = render_config(containers=containers)
+        if arguments.output is None:
+            sys.stdout.write(config)
+        else:
+            write_config(config=config, output=arguments.output)
+            logging.info("Wrote %s", arguments.output)
+        return 0
+    except subprocess.CalledProcessError as error:
+        logging.error("Docker command failed: %s", (error.stderr or str(error)).strip())
+    except (OSError, ValueError, subprocess.TimeoutExpired) as error:
+        logging.error("%s", error)
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())