my-homelab-configs/scripts/validate-tailnet-policy

#!/usr/bin/env python3
"""Validate the repo-managed Tailscale policy without contacting Tailscale."""

from __future__ import annotations

import json
import pathlib
import re
import sys


REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
POLICY_PATH = REPO_ROOT / "infra" / "tailscale" / "tailnet-policy.hujson"
INVENTORY_PATH = REPO_ROOT / "homelab.yml"


def strip_hujson(source: str) -> str:
    output: list[str] = []
    index = 0
    in_string = False
    escaped = False

    while index < len(source):
        char = source[index]
        next_char = source[index + 1] if index + 1 < len(source) else ""

        if in_string:
            output.append(char)
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == '"':
                in_string = False
            index += 1
            continue

        if char == '"':
            in_string = True
            output.append(char)
            index += 1
            continue

        if char == "/" and next_char == "/":
            index += 2
            while index < len(source) and source[index] not in "\r\n":
                index += 1
            continue

        if char == "/" and next_char == "*":
            index += 2
            while index + 1 < len(source) and source[index : index + 2] != "*/":
                index += 1
            index += 2
            continue

        output.append(char)
        index += 1

    normalized = "".join(output)
    cleaned: list[str] = []
    index = 0
    in_string = False
    escaped = False
    while index < len(normalized):
        char = normalized[index]
        if in_string:
            cleaned.append(char)
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == '"':
                in_string = False
            index += 1
            continue

        if char == '"':
            in_string = True
            cleaned.append(char)
            index += 1
            continue

        if char == ",":
            lookahead = index + 1
            while lookahead < len(normalized) and normalized[lookahead].isspace():
                lookahead += 1
            if lookahead < len(normalized) and normalized[lookahead] in "]}":
                index += 1
                continue

        cleaned.append(char)
        index += 1

    return "".join(cleaned)


def parse_simple_inventory(path: pathlib.Path) -> dict[str, str]:
    values: dict[str, str] = {}
    stack: list[tuple[int, str]] = []
    pattern = re.compile(r"^(\s*)([A-Za-z_][A-Za-z0-9_]*):(?:\s*(.*?))?\s*$")

    for raw_line in path.read_text(encoding="utf-8").splitlines():
        if not raw_line.strip() or raw_line.lstrip().startswith("#") or raw_line.lstrip().startswith("- "):
            continue
        match = pattern.match(raw_line)
        if not match:
            continue
        indent = len(match.group(1))
        key = match.group(2)
        value = (match.group(3) or "").strip()
        while stack and stack[-1][0] >= indent:
            stack.pop()
        path_key = ".".join([item[1] for item in stack] + [key])
        if value == "":
            stack.append((indent, key))
            continue
        if " #" in value:
            value = value.split(" #", 1)[0].strip()
        values[path_key] = value.strip("\"'")

    return values


def require(condition: bool, message: str, failures: list[str]) -> None:
    if not condition:
        failures.append(message)


def main() -> int:
    failures: list[str] = []
    policy = json.loads(strip_hujson(POLICY_PATH.read_text(encoding="utf-8")))
    inventory = parse_simple_inventory(INVENTORY_PATH)

    require(isinstance(policy.get("hosts"), dict), "policy must define hosts", failures)
    require(isinstance(policy.get("acls"), list) and policy["acls"], "policy must define non-empty acls", failures)
    require(isinstance(policy.get("tagOwners"), dict), "policy must define tagOwners", failures)

    hosts = policy.get("hosts", {})
    if isinstance(hosts, dict):
        expected_hosts = {
            "debian": inventory.get("hosts.debian.tailscale_ip"),
            "rpi4": inventory.get("hosts.rpi4.tailscale_ip"),
            "oci-edge": inventory.get("hosts.oci_edge.tailscale_ip"),
            "traefik-lan": inventory.get("network.metallb.traefik_ip"),
        }
        for name, expected_value in expected_hosts.items():
            require(bool(expected_value), f"homelab.yml missing inventory value for {name}", failures)
            require(hosts.get(name) == expected_value, f"policy host {name}={hosts.get(name)!r}, expected {expected_value!r}", failures)

    for index, acl in enumerate(policy.get("acls", [])):
        require(acl.get("action") == "accept", f"acl[{index}] action must be accept", failures)
        require(isinstance(acl.get("src"), list) and acl["src"], f"acl[{index}] must have non-empty src list", failures)
        require(isinstance(acl.get("dst"), list) and acl["dst"], f"acl[{index}] must have non-empty dst list", failures)

    if failures:
        print("tailnet policy validation failed:", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1

    print("tailnet policy static validation passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())