From aadb47a9550f43ee6089ca4f78f29530d02e284d Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Sun, 4 Oct 2026 08:23:46 +0200 Subject: [PATCH] feat: Add script to generate Caddy config from Docker container labels --- docker_to_caddy.py | 245 ++++++++++++++++++++++++++++++++++++++++ docker_to_caddy_test.py | 168 +++++++++++++++++++++++++++ 2 files changed, 413 insertions(+) create mode 100755 docker_to_caddy.py create mode 100644 docker_to_caddy_test.py diff --git a/docker_to_caddy.py b/docker_to_caddy.py new file mode 100755 index 0000000..cd7e478 --- /dev/null +++ b/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()) diff --git a/docker_to_caddy_test.py b/docker_to_caddy_test.py new file mode 100644 index 0000000..97a7a7e --- /dev/null +++ b/docker_to_caddy_test.py @@ -0,0 +1,168 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["pytest"] +# /// + +import copy +from pathlib import Path +import subprocess +import sys +from unittest.mock import patch + +import pytest + +import docker_to_caddy as script + + +def container(name: str = "app", domain: str = "ascii.app.abdus.dev") -> dict: + return { + "Name": f"/{name}", + "Config": {"Labels": {"caddy.domain": domain}}, + "State": {"Running": True, "Status": "running"}, + "HostConfig": {"PortBindings": { + "3000/tcp": [{"HostIp": "127.0.0.1", "HostPort": "4510"}], + }}, + "NetworkSettings": {"Ports": { + "3000/tcp": [{"HostIp": "127.0.0.1", "HostPort": "4510"}], + }}, + } + + +@pytest.mark.parametrize("address, expected", [ + ("127.0.0.1", "127.0.0.1:4510"), + ("0.0.0.0", "127.0.0.1:4510"), + ("::", "[::1]:4510"), + ("::1", "[::1]:4510"), + ("192.168.1.2", "192.168.1.2:4510"), +]) +def test_bindings(address: str, expected: str) -> None: + app = container() + app["NetworkSettings"]["Ports"]["3000/tcp"][0]["HostIp"] = address + assert f"reverse_proxy {expected}" in script.render_config(containers=[app]) + + +@pytest.mark.parametrize("selector, succeeds", [(None, False), ("3000", True), ("9000", False)]) +def test_port_selection(selector: str | None, succeeds: bool) -> None: + app = container() + app["NetworkSettings"]["Ports"]["4000/tcp"] = [ + {"HostIp": "::1", "HostPort": "4600"}, + ] + if selector is not None: + app["Config"]["Labels"]["caddy.port"] = selector + if succeeds: + assert "127.0.0.1:4510" in script.render_config(containers=[app]) + else: + with pytest.raises(ValueError): + script.render_config(containers=[app]) + + +@pytest.mark.parametrize("domain", ["", "bad host.dev", "example.dev {", "*.abdus.dev", "https://abdus.dev"]) +def test_invalid_domains(domain: str) -> None: + with pytest.raises(ValueError): + script.render_config(containers=[container(domain=domain)]) + + +def test_duplicate_domains() -> None: + with pytest.raises(ValueError): + script.render_config(containers=[container(), container(name="other")]) + + +def test_missing_published_ports() -> None: + app = container() + app["NetworkSettings"]["Ports"] = {} + with pytest.raises(ValueError): + script.render_config(containers=[app]) + + +def test_host_networking() -> None: + app = container() + app["HostConfig"]["NetworkMode"] = "host" + with pytest.raises(ValueError): + script.render_config(containers=[app]) + + +def test_stopped_unlabeled_and_empty() -> None: + stopped = container() + stopped["State"] = {"Running": False, "Status": "exited"} + unlabeled = container() + unlabeled["Config"]["Labels"] = None + assert script.render_config(containers=[stopped, unlabeled]) == script.render_config(containers=[]) + + +def test_deterministic_and_ipv4_preference() -> None: + first = container(name="a", domain="a.abdus.dev") + second = container(name="z", domain="z.abdus.dev") + first["NetworkSettings"]["Ports"]["3000/tcp"].append( + {"HostIp": "::", "HostPort": "4510"}, + ) + forward = script.render_config(containers=[first, second]) + assert forward == script.render_config(containers=[second, first]) + assert "reverse_proxy 127.0.0.1:4510" in forward + assert forward.index("a.abdus.dev") < forward.index("z.abdus.dev") + + +@pytest.mark.parametrize("left, right, protocol, port, expected", [ + ("0.0.0.0", "127.0.0.1", "tcp", "4510", True), + ("127.0.0.1", "127.0.0.2", "tcp", "4510", False), + ("::", "127.0.0.1", "tcp", "4510", True), + ("127.0.0.1", "127.0.0.1", "udp", "4510", False), + ("127.0.0.1", "127.0.0.1", "tcp", "0", False), +]) +def test_clashes(left: str, right: str, protocol: str, port: str, expected: bool) -> None: + first = container() + first["HostConfig"]["PortBindings"]["3000/tcp"][0]["HostIp"] = left + first["NetworkSettings"]["Ports"] = copy.deepcopy(first["HostConfig"]["PortBindings"]) + second = container(name="stopped") + second["Config"]["Labels"] = None + second["State"] = {"Running": False, "Status": "exited"} + second["NetworkSettings"]["Ports"] = {} + second["HostConfig"]["PortBindings"] = { + f"3000/{protocol}": [{"HostIp": right, "HostPort": port}], + } + with patch.object(script.logging, "warning") as warning: + script.warn_port_clashes(containers=[first, second]) + assert warning.call_count == int(expected) + + +def test_output_and_failure_preservation(tmp_path: Path) -> None: + output = tmp_path / "generated.caddy" + output.write_text(data="previous", encoding="utf-8") + output.chmod(mode=0o640) + with patch("sys.argv", ["docker_to_caddy.py", "--output", str(output)]): + with patch.object(script, "inspect_containers", return_value=[container(), container()]): + assert script.main() == 1 + assert output.read_text() == "previous" + with patch.object(script, "inspect_containers", side_effect=subprocess.CalledProcessError( + returncode=1, cmd="docker", stderr="permission denied", + )): + assert script.main() == 1 + assert output.read_text() == "previous" + script.write_config(config="replacement", output=output) + assert output.read_text() == "replacement" + assert output.stat().st_mode & 0o777 == 0o640 + + +def test_stdout(capsys: pytest.CaptureFixture[str]) -> None: + with patch("sys.argv", ["docker_to_caddy.py"]): + with patch.object(script, "inspect_containers", return_value=[container()]): + assert script.main() == 0 + assert capsys.readouterr().out == script.render_config(containers=[container()]) + + +def test_docker_discovery() -> None: + responses = [ + subprocess.CompletedProcess(args=[], returncode=0, stdout="id1\nid2\n"), + subprocess.CompletedProcess(args=[], returncode=0, stdout='[{"Name":"/one"},{"Name":"/two"}]'), + ] + with patch.object(script.subprocess, "run", side_effect=responses) as run: + assert script.inspect_containers() == [{"Name": "/one"}, {"Name": "/two"}] + assert run.call_args_list[1].kwargs["args"] == ["docker", "container", "inspect", "id1", "id2"] + with patch.object(script.subprocess, "run", return_value=subprocess.CompletedProcess( + args=[], returncode=0, stdout="", + )) as run: + assert script.inspect_containers() == [] + assert run.call_count == 1 + + +if __name__ == "__main__": + sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))