Initial commit

template/scripts/plugin.py

@@ -0,0 +1,150 @@
+from pathlib import Path
+from typing import Any
+
+import base64
+import json
+import makejinja
+import re
+import validate
+
+
+# Return the stripped contents of file_path, rejecting a missing or empty file
+def _read_stripped(file_path: str) -> str:
+    try:
+        content = Path(file_path).read_text().strip()
+    except FileNotFoundError:
+        raise FileNotFoundError(f"File not found: {file_path}") from None
+    if not content:
+        raise ValueError(f"{file_path} is empty")
+    return content
+
+
+# Return the parsed contents of a JSON file
+def _read_json(file_path: str) -> dict[str, Any]:
+    try:
+        return json.loads(_read_stripped(file_path))
+    except json.JSONDecodeError:
+        raise ValueError(f"Could not decode JSON file: {file_path}") from None
+
+
+# Return the age public or private key from age.key
+def age_key(key_type: str, file_path: str = 'age.key') -> str:
+    file_content = _read_stripped(file_path)
+    if key_type == 'public':
+        # Matches both classic (age1...) and post-quantum (age1pq1...) recipients
+        key_match = re.search(r"# public key: (age1[\w]+)", file_content)
+        if not key_match:
+            raise ValueError("Could not find public key in the age key file.")
+        return key_match.group(1)
+    elif key_type == 'private':
+        # (?:PQ-)? matches post-quantum identities (AGE-SECRET-KEY-PQ-1...) as well as classic ones
+        key_match = re.search(r"(AGE-SECRET-KEY-(?:PQ-)?1[\w]+)", file_content)
+        if not key_match:
+            raise ValueError("Could not find private key in the age key file.")
+        return key_match.group(1)
+    else:
+        raise ValueError("Invalid key type. Use 'public' or 'private'.")
+
+
+# Return cloudflare tunnel fields from cloudflare-tunnel.json
+def cloudflare_tunnel_id(file_path: str = 'cloudflare-tunnel.json') -> str:
+    data = _read_json(file_path)
+    tunnel_id = data.get("TunnelID")
+    if tunnel_id is None:
+        raise KeyError(f"Missing 'TunnelID' key in {file_path}")
+    if not tunnel_id:
+        raise ValueError(f"'TunnelID' is empty in {file_path}")
+    return tunnel_id
+
+
+# Return cloudflare tunnel fields from cloudflare-tunnel.json in TUNNEL_TOKEN format
+def cloudflare_tunnel_secret(file_path: str = 'cloudflare-tunnel.json') -> str:
+    data = _read_json(file_path)
+    for field in ("AccountTag", "TunnelID", "TunnelSecret"):
+        if field not in data:
+            raise KeyError(f"Missing '{field}' key in {file_path}")
+        if not data[field]:
+            raise ValueError(f"'{field}' is empty in {file_path}")
+    transformed_data = {
+        "a": data["AccountTag"],
+        "t": data["TunnelID"],
+        "s": data["TunnelSecret"]
+    }
+    json_string = json.dumps(transformed_data, separators=(',', ':'))
+    return base64.b64encode(json_string.encode('utf-8')).decode('utf-8')
+
+
+# Return the Flux deploy key from deploy.key
+def deploy_key(file_path: str = 'deploy.key') -> str:
+    return _read_stripped(file_path)
+
+
+# Return the Flux webhook token from flux-webhook-token.txt
+def webhook_token(file_path: str = 'flux-webhook-token.txt') -> str:
+    return _read_stripped(file_path)
+
+
+CONFIG_FILE = 'cluster.toml'
+
+# SSH host keys as published by each provider. Must cover every host in
+# KNOWN_SSH_HOSTS in validate.py; any other host requires the user to set
+# repository.known_hosts in cluster.toml.
+KNOWN_HOSTS = {
+    'github.com': (
+        'github.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl\n'
+        'github.com ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBEmKSENjQEezOmxkZMy7opKgwFB9nkt5YRrYMjNuG5N87uRgg6CLrbo5wAdT/y6v0mKV0U2w0WZ2YB/++Tpockg=\n'
+        'github.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQCj7ndNxQowgcQnjshcLrqPEiiphnt+VTTvDP6mHBL9j1aNUkY4Ue1gvwnGLVlOhGeYrnZaMgRK6+PKCUXaDbC7qtbW8gIkhL7aGCsOr/C56SJMy/BCZfxd1nWzAOxSDPgVsmerOBYfNqltV9/hWCqBywINIR+5dIg6JTJ72pcEpEjcYgXkE2YEFXV1JHnsKgbLWNlhScqb2UmyRkQyytRLtL+38TGxkxCflmO+5Z8CSSNY7GidjMIZ7Q4zMjA2n1nGrlTDkzwDCsw+wqFPGQA179cnfGWOWRVruj16z6XyvxvjJwbz0wQZ75XK5tKSb7FNyeIEs4TT4jk+S4dhPeAUC5y+bDYirYgM4GC7uEnztnZyaVWQ7B381AK4Qdrwt51ZqExKbQpTUNn+EjqoTwvqNj4kqx5QUCI0ThS/YkOxJCXmPUWZbhjpCg56i+2aB6CmK2JGhn57K5mj0MNdBXA4/WnwH6XoPWJzK5Nyu2zB3nAZp+S5hpQs+p1vN1/wsjk='
+    ),
+    'gitlab.com': (
+        'gitlab.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIAfuCHKVTjquxvt6CM6tdG4SLp1Btn/nOeHHE5UOzRdf\n'
+        'gitlab.com ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBFSMqzJeV9rUzU4kWitGjeR4PWSa29SPqJ1fVkhtj3Hw9xjLVXVYrU9QlYWrOLXBpQ6KWjbjTDTdDkoohFzgbEY=\n'
+        'gitlab.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsj2bNKTBSpIYDEGk9KxsGh3mySTRgMtXL583qmBpzeQ+jqCMRgBqB98u3z++J1sKlXHWfM9dyhSevkMwSbhoR8XIq/U0tCNyokEi/ueaBMCvbcTHhO7FcwzY92WK4Yt0aGROY5qX2UKSeOvuP4D6TPqKF1onrSzH9bx9XUf2lEdWT/ia1NEKjunUqu1xOB/StKDHMoX4/OKyIzuS0q/T1zOATthvasJFoPrAjkohTyaDUz2LN5JoH839hViyEG82yB+MjcFV5MU3N1l1QL3cVUCh93xSaua1N85qivl+siMkPGbO5xR/En4iEY6K2XPASUEMaieWVNTRCtJ4S8H+9'
+    ),
+    'codeberg.org': (
+        'codeberg.org ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIIVIC02vnjFyL+I4RHfvIGNtOgJMe769VTF1VR4EB3ZB\n'
+        'codeberg.org ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBL2pDxWr18SoiDJCGZ5LmxPygTlPu+cCKSkpqkvCyQzl5xmIMeKNdfdBpfbCGDPoZQghePzFZkKJNR/v9Win3Sc=\n'
+        'codeberg.org ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC8hZi7K1/2E2uBX8gwPRJAHvRAob+3Sn+y2hxiEhN0buv1igjYFTgFO2qQD8vLfU/HT/P/rqvEeTvaDfY1y/vcvQ8+YuUYyTwE2UaVU5aJv89y6PEZBYycaJCPdGIfZlLMmjilh/Sk8IWSEK6dQr+g686lu5cSWrFW60ixWpHpEVB26eRWin3lKYWSQGMwwKv4LwmW3ouqqs4Z4vsqRFqXJ/eCi3yhpT+nOjljXvZKiYTpYajqUC48IHAxTWugrKe1vXWOPxVXXMQEPsaIRc2hpK+v1LmfB7GnEGvF1UAKnEZbUuiD9PBEeD5a1MZQIzcoPWCrTxipEpuXQ5Tni4mN'
+    ),
+}
+
+
+# makejinja adds import_paths (this directory) to sys.path, and both
+# makejinja and pydantic come from the uv project environment, so the
+# validator runs in-process.
+def validate_config() -> dict[str, Any]:
+    try:
+        return validate.load(CONFIG_FILE)
+    except validate.ConfigError as e:
+        raise RuntimeError(f"config validation failed:\n{e}") from None
+
+
+class Plugin(makejinja.plugin.Plugin):
+    def __init__(self, data: dict[str, Any]):
+        self._data = data
+
+
+    def data(self) -> makejinja.plugin.Data:
+        data = validate_config()
+        if (
+            data['ingress']['mode'] == 'cloudflare-tunnel'
+            and not Path('cloudflare-tunnel.json').is_file()
+        ):
+            raise RuntimeError('cloudflare-tunnel.json not found — see README')
+        # The deploy key secret always pins every bundled provider host key
+        # (entries are matched per-hostname, so unused ones are inert);
+        # user-supplied entries for self-hosted git servers are appended.
+        repository = data['repository']
+        repository['known_hosts'] = '\n'.join(
+            [*KNOWN_HOSTS.values(), repository['known_hosts']]
+        ).strip()
+        return data
+
+
+    def functions(self) -> makejinja.plugin.Functions:
+        return [
+            age_key,
+            cloudflare_tunnel_id,
+            cloudflare_tunnel_secret,
+            deploy_key,
+            webhook_token
+        ]

template/scripts/test_validate.py

@@ -0,0 +1,173 @@
+"""Unit tests for the cluster.toml validator.
+
+Run from the repo root:
+    uv run --locked pytest template/scripts/test_validate.py -q
+"""
+
+from pathlib import Path
+
+import json
+import sys
+import tomllib
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from pydantic import ValidationError  # noqa: E402
+from validate import Config, ConfigError, format_errors, load, schema  # noqa: E402
+
+REPO_ROOT = Path(__file__).parents[2]
+VALID = sorted((REPO_ROOT / ".github/template-tests/valid").glob("*.toml"))
+INVALID = sorted((REPO_ROOT / ".github/template-tests/invalid").glob("*.toml"))
+
+
+def config_from(fixture: str, **overrides) -> dict:
+    raw = tomllib.loads((REPO_ROOT / ".github/template-tests/valid" / fixture).read_text())
+    for dotted, value in overrides.items():
+        target = raw
+        *parents, leaf = dotted.split(".")
+        for key in parents:
+            target = target.setdefault(key, {})
+        if value is None:
+            target.pop(leaf, None)
+        else:
+            target[leaf] = value
+    return raw
+
+
+@pytest.mark.parametrize("fixture", VALID, ids=lambda p: p.stem)
+def test_valid_fixture_accepted(fixture: Path):
+    load(str(fixture))
+
+
+@pytest.mark.parametrize("fixture", INVALID, ids=lambda p: p.stem)
+def test_invalid_fixture_rejected(fixture: Path):
+    with pytest.raises(ConfigError):
+        load(str(fixture))
+
+
+def _load_raw(raw: dict) -> Config:
+    try:
+        return Config.model_validate(raw)
+    except ValidationError as e:
+        raise ConfigError(format_errors(e)) from None
+
+
+def test_host_bits_error_suggests_network_address():
+    raw = config_from("public.toml", **{"network.node_cidr": "10.10.10.5/24"})
+    with pytest.raises(ConfigError, match=r"did you mean 10\.10\.10\.0/24"):
+        _load_raw(raw)
+
+
+def test_duplicate_address_names_both_owners():
+    raw = config_from("public.toml")
+    raw["nodes"][0]["address"] = raw["gateways"]["internal"]
+    with pytest.raises(ConfigError, match=r"gateways\.internal and nodes\[0\]\.address"):
+        _load_raw(raw)
+
+
+def test_default_gateway_derived_from_node_cidr():
+    raw = config_from("private.toml")
+    assert "default_gateway" not in raw["network"]
+    data = _load_raw(raw).model_dump(mode="json")
+    assert data["network"]["default_gateway"] == "10.10.10.1"
+
+
+def test_coredns_addr_default_and_override():
+    raw = config_from("private.toml")
+    assert _load_raw(raw).model_dump(mode="json")["kubernetes"]["coredns_addr"] == "10.43.0.10"
+    raw = config_from("private.toml", **{"kubernetes.coredns_addr": "10.43.0.53"})
+    assert _load_raw(raw).model_dump(mode="json")["kubernetes"]["coredns_addr"] == "10.43.0.53"
+    raw = config_from("private.toml", **{"kubernetes.coredns_addr": "192.168.9.9"})
+    with pytest.raises(ConfigError, match="not inside svc_cidr"):
+        _load_raw(raw)
+
+
+def test_spegel_enabled_follows_node_count():
+    two_nodes = config_from("private.toml")
+    assert _load_raw(two_nodes).spegel.enabled is True
+    one_node = config_from("private.toml")
+    one_node["nodes"] = one_node["nodes"][:1]
+    assert _load_raw(one_node).spegel.enabled is False
+    empty_section = config_from("private.toml", spegel={})
+    assert _load_raw(empty_section).spegel.enabled is True
+    explicit = config_from("private.toml", **{"spegel.enabled": False})
+    assert _load_raw(explicit).spegel.enabled is False
+
+
+def test_controller_count_ignores_workers():
+    raw = config_from("private.toml")
+    assert [n["controller"] for n in raw["nodes"]] == [True, False]
+    assert _load_raw(raw).controller_count == 1
+    raw["nodes"][1]["controller"] = True
+    assert _load_raw(raw).controller_count == 2
+
+
+def test_derived_fields_are_not_settable():
+    raw = config_from("public.toml", cilium_bgp_enabled=True)
+    with pytest.raises(ConfigError, match="cilium_bgp_enabled"):
+        _load_raw(raw)
+    raw = config_from("public.toml", controller_count=3)
+    with pytest.raises(ConfigError, match="controller_count"):
+        _load_raw(raw)
+
+
+def test_cert_sans_single_source():
+    raw = config_from("public.toml")
+    assert _load_raw(raw).cert_sans == ["127.0.0.1", "10.10.10.254", "example.com"]
+    raw = config_from("private.toml")
+    assert _load_raw(raw).cert_sans == ["127.0.0.1", "10.10.10.254"]
+
+
+def test_ingress_mode_follows_dns_provider():
+    cloudflare = config_from("public.toml", ingress=None)
+    assert _load_raw(cloudflare).ingress.mode == "cloudflare-tunnel"
+    internal = config_from("internal.toml")
+    assert (REPO_ROOT / ".github/template-tests/valid/internal.toml").exists()
+    assert "ingress" not in internal
+    assert _load_raw(internal).ingress.mode == "none"
+
+
+def test_direct_mode_requires_cloudflare_dns():
+    raw = config_from("internal.toml", **{"ingress.mode": "direct"})
+    with pytest.raises(ConfigError, match="requires dns.provider 'cloudflare'"):
+        _load_raw(raw)
+
+
+def test_schematic_id_inherits_from_talos_section():
+    raw = config_from("public.toml")
+    cfg = _load_raw(raw)
+    assert cfg.nodes[0].schematic_id == cfg.talos.schematic_id
+    assert cfg.nodes[1].schematic_id is not None
+
+
+def test_partial_bgp_rejected():
+    raw = config_from("private.toml", **{"cilium.bgp.router_addr": "10.10.1.1", "cilium.bgp.router_asn": "64513"})
+    with pytest.raises(ConfigError, match="partially configured"):
+        _load_raw(raw)
+
+
+def test_node_defaults_exported():
+    data = _load_raw(config_from("private.toml")).model_dump(mode="json")
+    node = data["nodes"][0]
+    assert node["mtu"] == 1500
+    assert node["secureboot"] is False
+    assert node["kernel_modules"] == []
+
+
+def test_gateways_may_leave_node_cidr_only_with_bgp():
+    with_bgp = config_from("public.toml", **{"gateways.external": "192.168.50.1"})
+    _load_raw(with_bgp)  # public.toml enables BGP
+    without_bgp = config_from("private.toml", **{"gateways.external": "192.168.50.1"})
+    with pytest.raises(ConfigError, match="required unless BGP is enabled"):
+        _load_raw(without_bgp)
+
+
+def test_schema_file_matches_model():
+    committed = json.loads((REPO_ROOT / "cluster.schema.json").read_text())
+    assert committed == schema(), "cluster.schema.json is stale: run `just template schema`"
+
+
+def test_schema_omits_computed_fields():
+    assert "cluster_issuer" not in schema()["properties"]

template/scripts/validate.py

@@ -0,0 +1,402 @@
+"""Validate cluster.toml, apply defaults, and emit the config as JSON.
+
+Standalone usage (doctor, CI): uv run --locked --no-dev template/scripts/validate.py [cluster.toml]
+Schema export (just template schema): uv run --locked --no-dev template/scripts/validate.py --schema
+In-process usage (makejinja plugin): from validate import load
+
+Exits non-zero with one human-readable error per line on stderr when the
+config is invalid.
+"""
+
+from ipaddress import IPv4Address, IPv4Network
+from pathlib import Path
+from typing import Annotated, Any, Literal, Self
+
+import json
+import re
+import sys
+import tomllib
+
+from pydantic import (
+    AfterValidator,
+    BaseModel,
+    BeforeValidator,
+    ConfigDict,
+    Field,
+    ValidationError,
+    computed_field,
+    model_validator,
+)
+
+# Git hosts whose SSH host keys are bundled with the template; ssh:// URLs
+# pointing anywhere else must provide repository.known_hosts.
+KNOWN_SSH_HOSTS = ["github.com", "gitlab.com", "codeberg.org"]
+
+REPO_URL_PATTERN = r"^(https?://|ssh://git@)[^/]+/.+$"
+FQDN_PATTERN = r"^([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}$"
+
+
+def _network(value: Any) -> Any:
+    """Parse a CIDR, requiring network-address form (no host bits set)."""
+    if not isinstance(value, str):
+        return value
+    try:
+        return IPv4Network(value)
+    except ValueError:
+        try:
+            fixed = IPv4Network(value, strict=False)
+        except ValueError:
+            raise ValueError(f"{value!r} is not a valid IPv4 CIDR") from None
+        raise ValueError(
+            f"{value!r} has host bits set; did you mean {fixed}?"
+        ) from None
+
+
+def _asn(value: str) -> str:
+    if value == "":
+        return value
+    if not re.fullmatch(r"[0-9]+", value):
+        raise ValueError(f"{value!r} must be a decimal ASN")
+    if not 1 <= int(value) <= 4294967295:
+        raise ValueError(f"{value!r} must be in the range 1-4294967295")
+    return value
+
+
+type Cidr = Annotated[IPv4Network, BeforeValidator(_network)]
+type Asn = Annotated[str, AfterValidator(_asn)]
+type Fqdn = Annotated[str, Field(pattern=FQDN_PATTERN)]
+
+
+class Model(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class Network(Model):
+    node_cidr: Cidr
+    dns_servers: list[IPv4Address] = [IPv4Address("1.1.1.1"), IPv4Address("1.0.0.1")]
+    ntp_servers: list[IPv4Address] = [IPv4Address("162.159.200.1"), IPv4Address("162.159.200.123")]
+    # The first IP in node_cidr unless set explicitly.
+    default_gateway: IPv4Address = Field(
+        default_factory=lambda data: data["node_cidr"].network_address + 1
+    )
+    vlan_tag: str | None = Field(default=None, pattern=r"^[0-9]+$")
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.default_gateway not in self.node_cidr:
+            raise ValueError(
+                f"default_gateway {self.default_gateway} is not inside node_cidr {self.node_cidr}"
+            )
+        if self.vlan_tag is not None and not 1 <= int(self.vlan_tag) <= 4094:
+            raise ValueError(f"vlan_tag {self.vlan_tag} must be in the range 1-4094")
+        return self
+
+
+class Api(Model):
+    addr: IPv4Address
+    tls_sans: list[Fqdn] | None = None
+
+
+class Kubernetes(Model):
+    pod_cidr: Cidr = IPv4Network("10.42.0.0/16")
+    svc_cidr: Cidr = IPv4Network("10.43.0.0/16")
+    # The 10th IP in svc_cidr unless set explicitly.
+    coredns_addr: IPv4Address = Field(
+        default_factory=lambda data: data["svc_cidr"].network_address + 10
+    )
+    api: Api
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.coredns_addr not in self.svc_cidr:
+            raise ValueError(
+                f"coredns_addr {self.coredns_addr} is not inside svc_cidr {self.svc_cidr}"
+            )
+        return self
+
+
+class Gateways(Model):
+    internal: IPv4Address
+    dns: IPv4Address
+    # Required when ingress.mode is not "none".
+    external: IPv4Address | None = None
+
+
+class Repository(Model):
+    url: str = Field(pattern=REPO_URL_PATTERN)
+    branch: str = Field(default="main", min_length=1)
+    webhook_provider: Literal["github", "gitlab", "generic-hmac", "none"] = "github"
+    known_hosts: str = ""
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.url.startswith("ssh://"):
+            host = self.url.removeprefix("ssh://git@").split("/", 1)[0].split(":", 1)[0]
+            if host not in KNOWN_SSH_HOSTS and not self.known_hosts:
+                raise ValueError(
+                    f"known_hosts is required for ssh:// URLs to {host!r} "
+                    f"(host keys are only bundled for {', '.join(KNOWN_SSH_HOSTS)})"
+                )
+        return self
+
+
+class Domain(Model):
+    name: Fqdn
+
+
+class Dns(Model):
+    provider: Literal["cloudflare", "none"] = "cloudflare"
+    token: str = ""
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.provider == "cloudflare" and not self.token:
+            raise ValueError("token is required when dns.provider is 'cloudflare'")
+        if self.provider == "none" and self.token:
+            raise ValueError("token must be empty when dns.provider is 'none'")
+        return self
+
+
+class Ingress(Model):
+    mode: Literal["cloudflare-tunnel", "direct", "none"] = "cloudflare-tunnel"
+
+
+class Bgp(Model):
+    router_addr: IPv4Address | Literal[""] = ""
+    router_asn: Asn = ""
+    node_asn: Asn = ""
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        unset = [name for name in ("router_addr", "router_asn", "node_asn") if getattr(self, name) == ""]
+        if unset and len(unset) < 3:
+            raise ValueError(
+                "bgp is partially configured: set router_addr, router_asn and "
+                f"node_asn together (missing: {', '.join(unset)})"
+            )
+        return self
+
+
+class Talos(Model):
+    # Default Image Factory schematic for nodes that don't set their own.
+    schematic_id: str | None = Field(default=None, pattern=r"^[a-z0-9]{64}$")
+
+
+class Spegel(Model):
+    # True when the cluster has more than one node, unless set explicitly.
+    enabled: bool | None = None
+
+
+class Cilium(Model):
+    loadbalancer_mode: Literal["dsr", "snat"] = "dsr"
+    bgp: Bgp = Bgp()
+
+
+class Node(Model):
+    name: str = Field(pattern=r"^[a-z0-9][a-z0-9\-]{0,61}[a-z0-9]$|^[a-z0-9]$")
+    address: IPv4Address
+    controller: bool
+    disk: str
+    mac_addr: str = Field(pattern=r"^([0-9a-f]{2}:){5}[0-9a-f]{2}$")
+    # Falls back to talos.schematic_id when unset.
+    schematic_id: str | None = Field(default=None, pattern=r"^[a-z0-9]{64}$")
+    mtu: int = Field(default=1500, ge=1450, le=9000)
+    secureboot: bool = False
+    encrypt_disk: bool = False
+    kernel_modules: list[str] = []
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.name in ("global", "controller", "worker"):
+            raise ValueError(f"node name {self.name!r} is reserved")
+        return self
+
+
+class Config(Model):
+    model_config = ConfigDict(extra="forbid", title="cluster.toml")
+
+    network: Network
+    kubernetes: Kubernetes
+    gateways: Gateways
+    repository: Repository
+    domain: Domain
+    dns: Dns
+    # Defaults to "cloudflare-tunnel" when dns.provider is "cloudflare",
+    # otherwise "none".
+    ingress: Ingress = Field(
+        default_factory=lambda data: Ingress(
+            mode="cloudflare-tunnel" if data["dns"].provider == "cloudflare" else "none"
+        )
+    )
+    cilium: Cilium = Cilium()
+    talos: Talos = Talos()
+    spegel: Spegel = Spegel()
+    nodes: list[Node]
+
+    @computed_field
+    @property
+    def cilium_bgp_enabled(self) -> bool:
+        bgp = self.cilium.bgp
+        return bgp.router_addr != "" and bgp.router_asn != "" and bgp.node_asn != ""
+
+    # Replica counts for control-plane-only workloads key off this rather
+    # than len(nodes); a cluster can have many workers but one controller.
+    @computed_field
+    @property
+    def controller_count(self) -> int:
+        return sum(1 for node in self.nodes if node.controller)
+
+    @computed_field
+    @property
+    def cluster_issuer(self) -> str:
+        if self.dns.provider == "cloudflare":
+            return "letsencrypt-production"
+        return "internal-ca"
+
+    # Single source for the machine and apiServer certificate SAN lists,
+    # which live in separate patch files.
+    @computed_field
+    @property
+    def cert_sans(self) -> list[str]:
+        return ["127.0.0.1", str(self.kubernetes.api.addr), *(self.kubernetes.api.tls_sans or [])]
+
+    @model_validator(mode="after")
+    def check(self) -> Self:
+        if self.spegel.enabled is None:
+            self.spegel.enabled = len(self.nodes) > 1
+        for i, node in enumerate(self.nodes):
+            if node.schematic_id is None:
+                node.schematic_id = self.talos.schematic_id
+            if node.schematic_id is None:
+                raise ValueError(
+                    f"nodes[{i}].schematic_id is required: set it on the node "
+                    "or set a cluster-wide default in [talos]"
+                )
+        if self.ingress.mode != "none" and self.dns.provider != "cloudflare":
+            raise ValueError(
+                f"ingress.mode {self.ingress.mode!r} requires dns.provider 'cloudflare'"
+            )
+        if self.ingress.mode != "none" and self.gateways.external is None:
+            raise ValueError(
+                f"gateways.external is required when ingress.mode is {self.ingress.mode!r}"
+            )
+
+        cidrs = {
+            "network.node_cidr": self.network.node_cidr,
+            "kubernetes.pod_cidr": self.kubernetes.pod_cidr,
+            "kubernetes.svc_cidr": self.kubernetes.svc_cidr,
+        }
+        names = list(cidrs)
+        for i, a in enumerate(names):
+            for b in names[i + 1:]:
+                if cidrs[a].overlaps(cidrs[b]):
+                    raise ValueError(f"{a} {cidrs[a]} overlaps {b} {cidrs[b]}")
+
+        addresses = {
+            "kubernetes.api.addr": self.kubernetes.api.addr,
+            "gateways.internal": self.gateways.internal,
+            "gateways.dns": self.gateways.dns,
+            "network.default_gateway": self.network.default_gateway,
+        } | {f"nodes[{i}].address": n.address for i, n in enumerate(self.nodes)}
+        if self.gateways.external is not None:
+            addresses["gateways.external"] = self.gateways.external
+        seen: dict[IPv4Address, str] = {}
+        for owner, addr in addresses.items():
+            if addr in seen:
+                raise ValueError(f"address {addr} is used by both {seen[addr]} and {owner}")
+            seen[addr] = owner
+
+        node_cidr = self.network.node_cidr
+        for i, node in enumerate(self.nodes):
+            if node.address not in node_cidr:
+                raise ValueError(
+                    f"nodes[{i}].address {node.address} is not inside node_cidr {node_cidr}"
+                )
+        if self.kubernetes.api.addr not in node_cidr:
+            raise ValueError(
+                f"kubernetes.api.addr {self.kubernetes.api.addr} is not inside node_cidr {node_cidr}"
+            )
+        # Without BGP the gateway VIPs are announced over L2 and must live in
+        # the node network.
+        if not self.cilium_bgp_enabled:
+            for name in ("internal", "dns", "external"):
+                addr = getattr(self.gateways, name)
+                if addr is not None and addr not in node_cidr:
+                    raise ValueError(
+                        f"gateways.{name} {addr} is not inside node_cidr {node_cidr} "
+                        "(required unless BGP is enabled)"
+                    )
+
+        for field, label in (("name", "name"), ("mac_addr", "MAC address")):
+            values: dict[str, int] = {}
+            for i, node in enumerate(self.nodes):
+                value = getattr(node, field)
+                if value in values:
+                    raise ValueError(
+                        f"duplicate node {label} {value!r} on nodes[{values[value]}] and nodes[{i}]"
+                    )
+                values[value] = i
+        return self
+
+
+def format_errors(error: ValidationError) -> str:
+    lines = []
+    for err in error.errors():
+        loc = ".".join(str(part) for part in err["loc"])
+        msg = err["msg"].removeprefix("Value error, ")
+        lines.append(f"{loc}: {msg}" if loc else msg)
+    return "\n".join(lines)
+
+
+class ConfigError(Exception):
+    pass
+
+
+# Validate config_file and return the defaulted config as a plain dict.
+# Raises ConfigError with a human-readable message.
+def load(config_file: str = "cluster.toml") -> dict[str, Any]:
+    path = Path(config_file)
+    try:
+        raw = tomllib.loads(path.read_text())
+    except FileNotFoundError:
+        raise ConfigError(f"{path}: file not found") from None
+    except tomllib.TOMLDecodeError as e:
+        raise ConfigError(f"{path}: invalid TOML: {e}") from None
+
+    try:
+        config = Config.model_validate(raw)
+    except ValidationError as e:
+        raise ConfigError(format_errors(e)) from None
+
+    # Unset optionals stay in the dump as None rather than being dropped:
+    # makejinja renders with StrictUndefined, so a template testing
+    # network.vlan_tag needs the key to exist.
+    return config.model_dump(mode="json")
+
+
+# JSON Schema for editor completion and validation of cluster.toml (taplo's
+# #:schema directive). Cross-field rules and data-aware defaults only exist in
+# the model validators, so the schema is an editing aid, not the gate.
+def schema() -> dict[str, Any]:
+    return {
+        "$schema": "https://json-schema.org/draft/2020-12/schema",
+        **Config.model_json_schema(),
+    }
+
+
+def main() -> int:
+    if sys.argv[1:] == ["--schema"]:
+        json.dump(schema(), sys.stdout, indent=2)
+        sys.stdout.write("\n")
+        return 0
+    try:
+        data = load(sys.argv[1] if len(sys.argv) > 1 else "cluster.toml")
+    except ConfigError as e:
+        print(e, file=sys.stderr)
+        return 1
+    json.dump(data, sys.stdout, indent=2)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())