nixos-servers/scripts/create-host/validators.py

"""Validation functions for host configuration."""

import re
from pathlib import Path
from typing import Optional


def validate_hostname_format(hostname: str) -> None:
    """
    Validate hostname format according to RFC 1123.

    Args:
        hostname: Hostname to validate

    Raises:
        ValueError: If hostname format is invalid
    """
    # RFC 1123: lowercase, alphanumeric, hyphens, max 63 chars
    pattern = r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$"

    if not re.match(pattern, hostname):
        raise ValueError(
            f"Invalid hostname '{hostname}'. "
            "Must be lowercase alphanumeric with hyphens, "
            "start and end with alphanumeric, max 63 characters."
        )


def validate_hostname_unique(hostname: str, repo_root: Path) -> None:
    """
    Validate that hostname is unique in the repository.

    Args:
        hostname: Hostname to check
        repo_root: Path to repository root

    Raises:
        ValueError: If hostname already exists
    """
    # Check if host directory exists
    host_dir = repo_root / "hosts" / hostname
    if host_dir.exists():
        raise ValueError(f"Host directory already exists: {host_dir}")

    # Check if hostname exists in flake.nix
    flake_path = repo_root / "flake.nix"
    if flake_path.exists():
        flake_content = flake_path.read_text()
        # Look for pattern like "      hostname = "
        hostname_pattern = rf'^\s+{re.escape(hostname)}\s*='
        if re.search(hostname_pattern, flake_content, re.MULTILINE):
            raise ValueError(f"Hostname '{hostname}' already exists in flake.nix")


def validate_ip_format(ip: str) -> None:
    """
    Validate IP address format with CIDR notation.

    Args:
        ip: IP address with CIDR (e.g., "10.69.13.50/24")

    Raises:
        ValueError: If IP format is invalid
    """
    if not ip:
        return

    # Check CIDR notation
    if "/" not in ip:
        raise ValueError(f"IP address must include CIDR notation (e.g., {ip}/24)")

    ip_part, cidr_part = ip.rsplit("/", 1)

    # Validate CIDR is /24
    if cidr_part != "24":
        raise ValueError(f"CIDR notation must be /24, got /{cidr_part}")

    # Validate IP format
    octets = ip_part.split(".")
    if len(octets) != 4:
        raise ValueError(f"Invalid IP address format: {ip_part}")

    try:
        octet_values = [int(octet) for octet in octets]
    except ValueError:
        raise ValueError(f"Invalid IP address format: {ip_part}")

    # Check each octet is 0-255
    for i, value in enumerate(octet_values):
        if not 0 <= value <= 255:
            raise ValueError(f"Invalid octet value {value} in IP address")

    # Check last octet is 1-254
    if not 1 <= octet_values[3] <= 254:
        raise ValueError(
            f"Last octet must be 1-254, got {octet_values[3]}"
        )


def validate_ip_subnet(ip: str) -> None:
    """
    Validate that IP address is in the correct subnet (10.69.13.0/24).

    Args:
        ip: IP address with CIDR (e.g., "10.69.13.50/24")

    Raises:
        ValueError: If IP is not in correct subnet
    """
    if not ip:
        return

    validate_ip_format(ip)

    ip_part = ip.split("/")[0]
    octets = ip_part.split(".")

    # Check subnet is 10.69.13.x
    if octets[:3] != ["10", "69", "13"]:
        raise ValueError(
            f"IP address must be in 10.69.13.0/24 subnet, got {ip_part}"
        )


def validate_ip_unique(ip: Optional[str], repo_root: Path) -> None:
    """
    Validate that IP address is not already in use.

    Args:
        ip: IP address with CIDR to check (None for DHCP)
        repo_root: Path to repository root

    Raises:
        ValueError: If IP is already in use
    """
    if not ip:
        return  # DHCP mode, no uniqueness check needed

    # Extract just the IP part without CIDR for searching
    ip_part = ip.split("/")[0]

    # Check all hosts/*/configuration.nix files
    hosts_dir = repo_root / "hosts"
    if hosts_dir.exists():
        for config_file in hosts_dir.glob("*/configuration.nix"):
            content = config_file.read_text()
            if ip_part in content:
                raise ValueError(
                    f"IP address {ip_part} already in use in {config_file}"
                )

    # Check terraform/vms.tf
    terraform_file = repo_root / "terraform" / "vms.tf"
    if terraform_file.exists():
        content = terraform_file.read_text()
        if ip_part in content:
            raise ValueError(
                f"IP address {ip_part} already in use in {terraform_file}"
            )