#!/usr/bin/env python3 """Validate CoreSystemConfig.json — the gate every builder in tools/ runs first. THE POINT OF THIS FILE: before CoreSystemConfig.json existed, an IP lived in six build scripts and a token in three, so "the door panel can't reach identity" could mean a typo in any one of them, discovered only after a 40-minute ISO build and a reboot. Centralising the values removes the duplication; this script removes the rest — it fails the build at second zero for anything it can prove wrong on paper. Deliberately stdlib-only and dependency-free (no jsonschema): the whole point is that a fresh checkout can validate a config before anything is installed, on a machine where `pip install` may not even be available yet. Exit codes: 0 = valid (warnings may still print) 1 = errors found 2 = unusable file """ from __future__ import annotations import ipaddress import json import re import sys from pathlib import Path from typing import Any HOSTNAME_RE = re.compile(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$") KIOSK_TYPES = {"thin-client", "touch-panel", "door-panel", "kitchen-display"} TIERS = {"auto", "gpu", "cpu"} ARCHITECTURES = {"amd64", "arm64"} # Tokens the validator insists on, and the enable-flag that makes each one required. # A token for a disabled service is not an error — you shouldn't have to invent a # transit token to build a door panel. TOKEN_REQUIREMENTS = [ ("identity_token", "identity", "identity, the door panel and the kitchen display"), ("pantry_vision_token", "pantry_vision", "pantry-vision and the kitchen display"), ("transit_token", "transit", "transit"), ] MIN_TOKEN_LEN = 32 # Substrings that mean "nobody actually generated this". Matched anywhere in the value, # case-insensitively, so padding a placeholder out to 32 characters doesn't sneak past. PLACEHOLDER_MARKERS = ( "changeme", "change-me", "your-", "yourtoken", "placeholder", "example", "xxxxx", "secret", "password", "todo", "fixme", "insert", ) def _looks_like_placeholder(value: str) -> bool: lowered = value.lower() if any(marker in lowered for marker in PLACEHOLDER_MARKERS): return True # A "token" made of one repeated character (aaaa..., 0000...) is padding, not entropy. return len(set(value)) <= 2 and len(value) > 4 class Report: def __init__(self) -> None: self.errors: list[str] = [] self.warnings: list[str] = [] def error(self, where: str, msg: str) -> None: self.errors.append(f"{where}: {msg}") def warn(self, where: str, msg: str) -> None: self.warnings.append(f"{where}: {msg}") def _get(cfg: dict, path: str, default: Any = None) -> Any: node = cfg for part in path.split("."): if not isinstance(node, dict) or part not in node: return default node = node[part] return node def _require(cfg: dict, path: str, kind, rep: Report, where: str | None = None) -> Any: """Fetch a required key, reporting a precise path rather than a KeyError.""" value = _get(cfg, path, None) label = where or path if value is None: rep.error(label, "missing (required)") return None if kind is not None and not isinstance(value, kind): rep.error(label, f"must be {getattr(kind, '__name__', kind)}, got {type(value).__name__}") return None return value def validate_household(cfg: dict, rep: Report) -> None: tz = _require(cfg, "household.timezone", str, rep) if tz: # Checked against the zoneinfo database actually present, not a regex — a # plausible-looking but wrong timezone ("Europe/Wien") silently gives every # image the wrong clock, and every timestamped log with it. if not Path(f"/usr/share/zoneinfo/{tz}").exists(): rep.warn("household.timezone", f"'{tz}' isn't in this machine's /usr/share/zoneinfo — verify it exists on the target") _require(cfg, "household.locale", str, rep) layout = _require(cfg, "household.keyboard_layout", str, rep) if layout and not re.match(r"^[a-z]{2,6}$", layout): rep.error("household.keyboard_layout", f"'{layout}' doesn't look like an xkb layout (e.g. 'de', 'us')") _require(cfg, "household.debian_release", str, rep) def validate_network(cfg: dict, rep: Report) -> None: prefix = _require(cfg, "network.subnet_prefix", str, rep) if not prefix: return if prefix.endswith("."): rep.error("network.subnet_prefix", f"'{prefix}' must not end with a dot") return try: network = ipaddress.ip_network(f"{prefix}.0/24", strict=True) except ValueError as exc: rep.error("network.subnet_prefix", f"'{prefix}' is not a valid /24 prefix ({exc})") return if not network.is_private: rep.warn("network.subnet_prefix", f"{network} is not a private range — this stack is designed to be LAN-only " "(docs/network-integration.md §1)") # 192.168.0.x and 192.168.1.x are what nearly every café, hotel and consumer router # uses, so a WireGuard split tunnel routing this VLAN would collide with them — # docs/network-integration.md §2.1 spells this out. if prefix in ("192.168.0", "192.168.1"): rep.warn("network.subnet_prefix", f"'{prefix}' collides with typical café/hotel LANs; a WireGuard split tunnel " "routing it will break connectivity on those networks (network-integration.md §2.1)") gw = _get(cfg, "network.gateway_last_octet") if not isinstance(gw, int) or not 1 <= gw <= 254: rep.error("network.gateway_last_octet", f"must be an int 1-254, got {gw!r}") dns = _get(cfg, "network.dns_servers") if not isinstance(dns, list) or not dns: rep.error("network.dns_servers", "must be a non-empty list of IP addresses") else: for entry in dns: try: ipaddress.ip_address(entry) except ValueError: rep.error("network.dns_servers", f"'{entry}' is not a valid IP address") ssid = _get(cfg, "network.wifi.ssid", "") psk = _get(cfg, "network.wifi.psk", "") if ssid and not psk: rep.error("network.wifi.psk", "an SSID is set but the PSK is empty") if psk and not ssid: rep.error("network.wifi.ssid", "a PSK is set but the SSID is empty") if psk and len(psk) < 8: rep.error("network.wifi.psk", "WPA2 pre-shared keys are at least 8 characters") def validate_hosts(cfg: dict, rep: Report) -> None: """The twinning check. The container host and the LLM host derive each other's addresses from these two numbers, so the one thing that must hold is that they are two distinct, valid, same-subnet addresses — get that right and no builder can produce a mismatched pair. """ octets: dict[str, int] = {} for host in ("container_host", "llm_host"): hostname = _require(cfg, f"{host}.hostname", str, rep) if hostname and not HOSTNAME_RE.match(hostname): rep.error(f"{host}.hostname", f"'{hostname}' is not a valid DNS label (lowercase letters, digits, hyphens; " "no leading/trailing hyphen)") octet = _get(cfg, f"{host}.ip_last_octet") if not isinstance(octet, int) or not 1 <= octet <= 254: rep.error(f"{host}.ip_last_octet", f"must be an int 1-254, got {octet!r}") else: octets[host] = octet disk = _require(cfg, f"{host}.install_disk", str, rep) if disk and not disk.startswith("/dev/"): rep.error(f"{host}.install_disk", f"'{disk}' should be a device path like /dev/sda or /dev/nvme0n1") _require(cfg, f"{host}.admin_username", str, rep) if len(octets) == 2: if octets["container_host"] == octets["llm_host"]: rep.error("llm_host.ip_last_octet", f"the container host and LLM host both want .{octets['llm_host']} — " "they are two separate machines and need two addresses") gw = _get(cfg, "network.gateway_last_octet") for host, octet in octets.items(): if octet == gw: rep.error(f"{host}.ip_last_octet", f".{octet} is the gateway address") tier = _get(cfg, "llm_host.tier") if tier not in TIERS: rep.error("llm_host.tier", f"must be one of {sorted(TIERS)}, got {tier!r}") for key in ("text_model_gpu", "text_model_cpu", "vision_model"): _require(cfg, f"llm_host.{key}", str, rep) for key in ("max_loaded_models", "num_parallel"): value = _get(cfg, f"llm_host.{key}") if not isinstance(value, int) or value < 1: rep.error(f"llm_host.{key}", f"must be a positive int, got {value!r}") keep_alive = _get(cfg, "llm_host.keep_alive") if not isinstance(keep_alive, str) or not re.match(r"^\d+[smh]$|^-1$", str(keep_alive)): rep.error("llm_host.keep_alive", f"must look like '30m', '2h', '600s' or '-1', got {keep_alive!r}") def validate_ports(cfg: dict, rep: Report) -> None: """Port uniqueness is the single highest-value check here: a duplicate produces a container that silently fails to bind, or two services fighting over one port with whichever won last boot answering. Both are miserable to diagnose from the symptom. """ ports = _get(cfg, "ports") if not isinstance(ports, dict): rep.error("ports", "missing or not an object") return seen: dict[int, list[str]] = {} for name, value in ports.items(): if name.startswith("_"): continue if not isinstance(value, int) or not 1 <= value <= 65535: rep.error(f"ports.{name}", f"must be an int 1-65535, got {value!r}") continue if value < 1024: rep.warn(f"ports.{name}", f"{value} is a privileged port (<1024)") seen.setdefault(value, []).append(name) for value, names in sorted(seen.items()): if len(names) > 1: rep.error("ports", f"port {value} is claimed by {len(names)} services: {', '.join(sorted(names))}") def validate_secrets(cfg: dict, rep: Report) -> None: enable = _get(cfg, "container_host.enable", {}) or {} for token_name, flag, used_by in TOKEN_REQUIREMENTS: value = _get(cfg, f"secrets.{token_name}", "") or "" if not enable.get(flag): if value: rep.warn(f"secrets.{token_name}", f"set, but container_host.enable.{flag} is false — it won't be used") continue if not value: rep.error(f"secrets.{token_name}", f"required because container_host.enable.{flag} is true (used by {used_by}). " "Generate one with: openssl rand -hex 32") # Placeholder BEFORE length: a long placeholder ("changeme-changeme-changeme-...") # would otherwise sail past the length check, and a short one would be reported # as merely too short — which invites someone to pad it rather than generate one. elif _looks_like_placeholder(value): rep.error(f"secrets.{token_name}", f"'{value[:24]}' is a placeholder, not a generated secret. " "Generate one with: openssl rand -hex 32") elif len(value) < MIN_TOKEN_LEN: rep.error(f"secrets.{token_name}", f"only {len(value)} characters; use at least {MIN_TOKEN_LEN} " "(openssl rand -hex 32)") mqtt_user = _get(cfg, "secrets.mqtt_username", "") or "" mqtt_pass = _get(cfg, "secrets.mqtt_password", "") or "" if mqtt_user and not mqtt_pass: rep.error("secrets.mqtt_password", "an MQTT username is set but the password is empty") if not mqtt_user: rep.warn("secrets.mqtt_username", "empty — every kiosk will connect to Mosquitto anonymously. Fine while " "allow_anonymous is on; revisit before that changes") # OPNsense's API key pair. Half-configured is the failure worth catching: an # address with no credentials (or credentials with no address) looks configured # and produces a network digest that silently reports nothing. opnsense_url = str(_get(cfg, "opnsense.base_url", "") or "").strip() opnsense_key = str(_get(cfg, "secrets.opnsense_api_key", "") or "").strip() opnsense_secret = str(_get(cfg, "secrets.opnsense_api_secret", "") or "").strip() if opnsense_url and not (opnsense_key and opnsense_secret): rep.error("secrets.opnsense_api_key", "opnsense.base_url is set, so both opnsense_api_key and " "opnsense_api_secret are needed. OPNsense mints them: System -> Access " "-> Users -> API keys; give that user only the 'Services: Intrusion " "Detection' privilege") if (opnsense_key or opnsense_secret) and not opnsense_url: rep.warn("opnsense.base_url", "empty, but an OPNsense API key is set — nothing will use it until this " "points at the firewall (e.g. https://opnsense.home.lan)") if not (_get(cfg, "secrets.ha_token", "") or ""): rep.warn("secrets.ha_token", "empty — identity's /register and /presence can't reach Home Assistant until " "this is a real Long-Lived Access Token. It cannot be generated ahead of time; " "fill it in and re-run the builder once HA is up") key = _get(cfg, "secrets.ssh_authorized_key", "") or "" if key and not re.match(r"^(ssh-(rsa|ed25519|dss)|ecdsa-sha2-\S+) \S+", key): rep.error("secrets.ssh_authorized_key", "doesn't look like an OpenSSH public key (should start 'ssh-ed25519 AAAA...')") if key and "PRIVATE KEY" in key: rep.error("secrets.ssh_authorized_key", "this is a PRIVATE key — put the .pub here instead") if not key: rep.warn("secrets.ssh_authorized_key", "empty — the built images will have no way in over SSH. Fine for a kiosk you " "only ever touch physically, painful for a headless host") def validate_proxy(cfg: dict, rep: Report) -> None: proxy = _get(cfg, "proxy") or {} if not proxy.get("enabled"): return hostname = (proxy.get("hostname") or "").strip() if not hostname: rep.error("proxy.hostname", "required when the proxy is enabled — it's what the " "certificate is issued for") elif not re.match(r"^[a-z0-9]([a-z0-9.-]{0,253}[a-z0-9])?$", hostname): rep.error("proxy.hostname", f"'{hostname}' is not a valid hostname or IP") elif hostname == "home.example.lan": rep.warn("proxy.hostname", "still the template's placeholder — set a name that " "actually resolves to the container host") mode = proxy.get("tls", "internal") if mode not in {"internal", "custom"}: rep.error("proxy.tls", f"must be 'internal' or 'custom', got {mode!r}") elif mode == "custom": for key in ("cert_file", "key_file"): if not (proxy.get(key) or "").strip(): rep.error(f"proxy.{key}", "required when proxy.tls is 'custom'") else: rep.warn("proxy.tls", "'internal' means Caddy runs its own CA — browsers warn until you install " "its root (tools/export-proxy-ca.sh). A kiosk cannot dismiss that warning, " "which is why the kiosk images still use plain HTTP; see proxy/README.md") def validate_kiosks(cfg: dict, rep: Report) -> None: kiosks = _get(cfg, "kiosks") if kiosks is None: rep.warn("kiosks", "no kiosks defined — only the core pair will be buildable") return if not isinstance(kiosks, list): rep.error("kiosks", "must be a list") return hostnames: dict[str, int] = {} for index, kiosk in enumerate(kiosks): where = f"kiosks[{index}]" if not isinstance(kiosk, dict): rep.error(where, "must be an object") continue ktype = kiosk.get("type") if ktype not in KIOSK_TYPES: rep.error(f"{where}.type", f"must be one of {sorted(KIOSK_TYPES)}, got {ktype!r}") hostname = kiosk.get("hostname") if not isinstance(hostname, str) or not HOSTNAME_RE.match(hostname or ""): rep.error(f"{where}.hostname", f"{hostname!r} is not a valid DNS label") else: if hostname in hostnames: rep.error(f"{where}.hostname", f"'{hostname}' is already used by kiosks[{hostnames[hostname]}] — " "hostnames identify these devices on the network and in HA, so they must be unique") hostnames[hostname] = index if not kiosk.get("friendly_name"): rep.error(f"{where}.friendly_name", "missing — this is the name shown on the HA device") if not kiosk.get("kiosk_username"): rep.error(f"{where}.kiosk_username", "missing") # Audio endpoints share the hostname namespace with kiosks: they are all devices on # one network, and two of anything answering to the same name is the same problem # regardless of what kind of device they are. for index, endpoint in enumerate(_get(cfg, "audio_endpoints", []) or []): where = f"audio_endpoints[{index}]" if not isinstance(endpoint, dict): rep.error(where, "must be an object") continue hostname = endpoint.get("hostname") if not isinstance(hostname, str) or not HOSTNAME_RE.match(hostname or ""): rep.error(f"{where}.hostname", f"{hostname!r} is not a valid DNS label") elif hostname in hostnames: rep.error(f"{where}.hostname", f"'{hostname}' is already used by kiosks[{hostnames[hostname]}] — " "kiosks and audio endpoints share one hostname namespace") else: hostnames[hostname] = index if endpoint.get("arch") not in ARCHITECTURES: rep.error(f"{where}.arch", f"must be one of {sorted(ARCHITECTURES)}, got {endpoint.get('arch')!r}") if not endpoint.get("friendly_name"): rep.error(f"{where}.friendly_name", "missing") # Cross-check: a kiosk that talks to a disabled service will build fine and then # fail at runtime with a connection error, which is exactly the class of "works on # paper" mistake this file exists to catch. enable = _get(cfg, "container_host.enable", {}) or {} needs = { "door-panel": [("identity", "identity"), ("pantry_vision", "pantry-vision")], "kitchen-display": [("identity", "identity"), ("pantry_vision", "pantry-vision")], } for index, kiosk in enumerate(kiosks): if not isinstance(kiosk, dict): continue for flag, label in needs.get(kiosk.get("type", ""), []): if not enable.get(flag): rep.error(f"kiosks[{index}]", f"a {kiosk.get('type')} needs {label}, but container_host.enable.{flag} is false") def validate(cfg: dict) -> Report: rep = Report() validate_network(cfg, rep) validate_household(cfg, rep) validate_hosts(cfg, rep) validate_ports(cfg, rep) validate_secrets(cfg, rep) validate_kiosks(cfg, rep) validate_proxy(cfg, rep) return rep def main(argv: list[str]) -> int: path = Path(argv[1]) if len(argv) > 1 else Path(__file__).resolve().parent.parent / "CoreSystemConfig.json" if not path.exists(): template = path.parent / "CoreSystemConfig.json.template" print(f"error: {path} not found.", file=sys.stderr) if template.exists(): print(f" Copy the template and fill it in:\n" f" cp {template.name} {path.name}\n" f" $EDITOR {path.name}", file=sys.stderr) return 2 try: cfg = json.loads(path.read_text()) except json.JSONDecodeError as exc: # Line and column, because a trailing comma in a 200-line JSON file is otherwise # a genuinely annoying thing to find. print(f"error: {path} is not valid JSON — line {exc.lineno}, column {exc.colno}: {exc.msg}", file=sys.stderr) return 2 rep = validate(cfg) for warning in rep.warnings: print(f" warn {warning}") for error in rep.errors: print(f" ERROR {error}", file=sys.stderr) if rep.errors: print(f"\n{len(rep.errors)} error(s), {len(rep.warnings)} warning(s) — {path.name} is not usable yet.", file=sys.stderr) return 1 print(f"\n{path.name} is valid ({len(rep.warnings)} warning(s)).") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))