railiance-infra/scripts/inventory_contract.py

#!/usr/bin/env python3
"""Validate the S1 host inventory without contacting a provider or host."""

from __future__ import annotations

import argparse
import ipaddress
import json
import re
import sys
from pathlib import Path
from typing import Any

import yaml


SCHEMA_VERSION = "1.0"
PROVIDERS = {"hosteurope", "hetzner"}
LIFECYCLE_MODES = {"adopted", "provider-managed"}
BASELINE_PROFILES = {"ufw-managed", "external-firewall"}
HETZNER_REQUIRED = {"server_type", "region", "image", "role"}
HETZNER_OPTIONAL = {"labels"}
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,62}$")
USER_RE = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$")


class InventoryError(ValueError):
    """The inventory does not satisfy the S1 contract."""


def _nonempty_string(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def validate_inventory(payload: Any) -> dict[str, Any]:
    if not isinstance(payload, dict):
        raise InventoryError("inventory must be a YAML object")
    if str(payload.get("schema_version")) != SCHEMA_VERSION:
        raise InventoryError(f"schema_version must be {SCHEMA_VERSION!r}")
    servers = payload.get("servers")
    if not isinstance(servers, list) or not servers:
        raise InventoryError("servers must be a non-empty list")

    errors: list[str] = []
    names: set[str] = set()
    for index, server in enumerate(servers):
        label = f"servers[{index}]"
        if not isinstance(server, dict):
            errors.append(f"{label} must be an object")
            continue
        name = server.get("name")
        if not _nonempty_string(name) or not NAME_RE.fullmatch(name):
            errors.append(f"{label}.name must be a stable host identifier")
            name = label
        elif name in names:
            errors.append(f"{label}.name duplicates {name!r}")
        else:
            names.add(name)
        label = str(name)

        provider = server.get("provider")
        lifecycle = server.get("lifecycle_mode")
        profile = server.get("baseline_profile")
        ssh_user = server.get("ssh_user")
        if provider not in PROVIDERS:
            errors.append(f"{label}: provider must be one of {sorted(PROVIDERS)}")
        if lifecycle not in LIFECYCLE_MODES:
            errors.append(
                f"{label}: lifecycle_mode must be one of {sorted(LIFECYCLE_MODES)}"
            )
        if profile not in BASELINE_PROFILES:
            errors.append(
                f"{label}: baseline_profile must be one of {sorted(BASELINE_PROFILES)}"
            )
        if not _nonempty_string(ssh_user) or not USER_RE.fullmatch(ssh_user):
            errors.append(f"{label}: ssh_user must be a valid Unix user name")

        if lifecycle == "adopted":
            address = server.get("ip")
            try:
                ipaddress.ip_address(address)
            except (TypeError, ValueError):
                errors.append(f"{label}: adopted hosts require a literal ip address")
            if "provisioning" in server:
                errors.append(
                    f"{label}: adopted hosts must not carry provider provisioning fields"
                )
        elif lifecycle == "provider-managed":
            if provider != "hetzner":
                errors.append(
                    f"{label}: provider-managed is currently implemented only for hetzner"
                )
            if "ip" in server:
                errors.append(
                    f"{label}: provider-managed addresses come from provider output; remove ip"
                )
            provisioning = server.get("provisioning")
            if not isinstance(provisioning, dict):
                errors.append(f"{label}: provider-managed hosts require provisioning")
            else:
                missing = sorted(
                    key
                    for key in HETZNER_REQUIRED
                    if not _nonempty_string(provisioning.get(key))
                )
                unknown = sorted(
                    set(provisioning) - HETZNER_REQUIRED - HETZNER_OPTIONAL
                )
                if missing:
                    errors.append(
                        f"{label}: provisioning missing {', '.join(missing)}"
                    )
                if unknown:
                    errors.append(
                        f"{label}: provisioning has unknown fields {', '.join(unknown)}"
                    )
                labels = provisioning.get("labels", [])
                if not isinstance(labels, list) or not all(
                    _nonempty_string(item) for item in labels
                ):
                    errors.append(
                        f"{label}: provisioning.labels must be a list of strings"
                    )

    if errors:
        raise InventoryError("inventory contract failed:\n- " + "\n- ".join(errors))
    return payload


def load_inventory(path: Path) -> dict[str, Any]:
    try:
        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError) as exc:
        raise InventoryError(f"cannot read {path}: {exc}") from exc
    return validate_inventory(payload)


def managed_hetzner_servers(payload: dict[str, Any]) -> list[dict[str, Any]]:
    validate_inventory(payload)
    return [
        server
        for server in payload["servers"]
        if server["provider"] == "hetzner"
        and server["lifecycle_mode"] == "provider-managed"
    ]


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "inventory", nargs="?", type=Path, default=Path("inventory/servers.yaml")
    )
    parser.add_argument(
        "--managed-hetzner", action="store_true", help="print selected names"
    )
    args = parser.parse_args()
    try:
        payload = load_inventory(args.inventory)
    except InventoryError as exc:
        print(exc, file=sys.stderr)
        return 1
    selected = managed_hetzner_servers(payload)
    result = {
        "ok": True,
        "schema_version": SCHEMA_VERSION,
        "server_count": len(payload["servers"]),
        "managed_hetzner": [server["name"] for server in selected],
    }
    print(json.dumps(result, sort_keys=True))
    return 0


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