467 lines
15 KiB
Python
467 lines
15 KiB
Python
"""Strict TOML model for the GovOPlaN Kubernetes VM lab."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import ipaddress
|
|
from pathlib import Path
|
|
import re
|
|
import tomllib
|
|
from typing import Any, Mapping
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
SCHEMA_VERSION = 1
|
|
_NAME = re.compile(r"^[a-z][a-z0-9-]{1,47}$")
|
|
_HOSTNAME = re.compile(
|
|
r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
|
|
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$"
|
|
)
|
|
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
_K3S_VERSION = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+\+k3s[0-9]+$")
|
|
_SSH_TARGET = re.compile(r"^(?:[A-Za-z0-9_.-]+@)?[A-Za-z0-9_.:-]+$")
|
|
_BRIDGE = re.compile(r"^[A-Za-z0-9_.:-]{1,32}$")
|
|
_ROLE = {"control", "worker", "state"}
|
|
_MODE = {"rehearsal", "acceptance"}
|
|
|
|
|
|
class LabConfigError(ValueError):
|
|
"""Raised when the lab inventory cannot be used safely."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LabNode:
|
|
name: str
|
|
role: str
|
|
address: ipaddress.IPv4Address
|
|
hypervisor: str
|
|
failure_domain: str
|
|
mac_address: str
|
|
cpus: int
|
|
memory_mib: int
|
|
disk_gib: int
|
|
|
|
@property
|
|
def is_local(self) -> bool:
|
|
return self.hypervisor == "local"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class NetworkConfig:
|
|
prefix_length: int
|
|
gateway: ipaddress.IPv4Address
|
|
dns_servers: tuple[ipaddress.IPv4Address, ...]
|
|
bridge: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImageConfig:
|
|
url: str
|
|
sha256: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class K3sConfig:
|
|
version: str
|
|
binary_url: str
|
|
binary_sha256: str
|
|
install_script_url: str
|
|
install_script_sha256: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ReleaseConfig:
|
|
manifest_url: str
|
|
manifest_sha256: str
|
|
keyring_url: str
|
|
keyring_sha256: str
|
|
channel: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LabConfig:
|
|
source: Path
|
|
schema_version: int
|
|
name: str
|
|
mode: str
|
|
state_directory: Path
|
|
vm_image_directory: str
|
|
ssh_user: str
|
|
ssh_private_key: Path
|
|
ssh_public_key: Path
|
|
namespace: str
|
|
public_host: str
|
|
s3_host: str
|
|
ingress_class: str
|
|
module_set: str
|
|
api_replicas: int
|
|
web_replicas: int
|
|
worker_replicas: int
|
|
db_connection_limit: int
|
|
network: NetworkConfig
|
|
image: ImageConfig
|
|
k3s: K3sConfig
|
|
release: ReleaseConfig
|
|
nodes: tuple[LabNode, ...]
|
|
|
|
@property
|
|
def controls(self) -> tuple[LabNode, ...]:
|
|
return tuple(node for node in self.nodes if node.role == "control")
|
|
|
|
@property
|
|
def workers(self) -> tuple[LabNode, ...]:
|
|
return tuple(node for node in self.nodes if node.role == "worker")
|
|
|
|
@property
|
|
def state_node(self) -> LabNode:
|
|
return next(node for node in self.nodes if node.role == "state")
|
|
|
|
@property
|
|
def primary_control(self) -> LabNode:
|
|
return self.controls[0]
|
|
|
|
@property
|
|
def public_url(self) -> str:
|
|
return f"https://{self.public_host}"
|
|
|
|
@property
|
|
def s3_url(self) -> str:
|
|
return f"https://{self.s3_host}:9443"
|
|
|
|
@property
|
|
def evidence_capable(self) -> bool:
|
|
worker_domains = {node.failure_domain for node in self.workers}
|
|
worker_hypervisors = {node.hypervisor for node in self.workers}
|
|
return (
|
|
self.mode == "acceptance"
|
|
and len(worker_domains) == len(self.workers)
|
|
and len(worker_hypervisors) == len(self.workers)
|
|
and self.state_node.failure_domain not in worker_domains
|
|
and self.state_node.hypervisor not in worker_hypervisors
|
|
)
|
|
|
|
|
|
def load_config(path: Path) -> LabConfig:
|
|
source = path.expanduser().resolve()
|
|
try:
|
|
raw = tomllib.loads(source.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise LabConfigError(f"lab configuration does not exist: {source}") from exc
|
|
except tomllib.TOMLDecodeError as exc:
|
|
raise LabConfigError(f"lab configuration is not valid TOML: {exc}") from exc
|
|
root = _mapping(raw, "lab configuration")
|
|
_only_keys(
|
|
root,
|
|
{
|
|
"schema_version",
|
|
"name",
|
|
"mode",
|
|
"state_directory",
|
|
"vm_image_directory",
|
|
"ssh_user",
|
|
"ssh_private_key",
|
|
"ssh_public_key",
|
|
"namespace",
|
|
"public_host",
|
|
"s3_host",
|
|
"ingress_class",
|
|
"module_set",
|
|
"api_replicas",
|
|
"web_replicas",
|
|
"worker_replicas",
|
|
"db_connection_limit",
|
|
"network",
|
|
"image",
|
|
"k3s",
|
|
"release",
|
|
"nodes",
|
|
},
|
|
"lab configuration",
|
|
)
|
|
schema_version = _integer(root, "schema_version")
|
|
if schema_version != SCHEMA_VERSION:
|
|
raise LabConfigError(
|
|
f"schema_version must be {SCHEMA_VERSION}; found {schema_version}"
|
|
)
|
|
name = _pattern(root, "name", _NAME)
|
|
mode = _choice(root, "mode", _MODE)
|
|
base = source.parent
|
|
state_directory = _path(root, "state_directory", base)
|
|
vm_image_directory = _absolute_posix_path(root, "vm_image_directory")
|
|
ssh_user = _pattern(root, "ssh_user", re.compile(r"^[a-z_][a-z0-9_-]{0,31}$"))
|
|
ssh_private_key = _path(root, "ssh_private_key", base)
|
|
ssh_public_key = _path(root, "ssh_public_key", base)
|
|
namespace = _pattern(root, "namespace", _NAME)
|
|
public_host = _pattern(root, "public_host", _HOSTNAME)
|
|
s3_host = _pattern(root, "s3_host", _HOSTNAME)
|
|
if public_host == s3_host:
|
|
raise LabConfigError("public_host and s3_host must be different")
|
|
ingress_class = _pattern(root, "ingress_class", _NAME)
|
|
module_set = _choice(root, "module_set", {"core", "base", "full"})
|
|
api_replicas = _bounded_integer(root, "api_replicas", 2, 32)
|
|
web_replicas = _bounded_integer(root, "web_replicas", 2, 32)
|
|
worker_replicas = _bounded_integer(root, "worker_replicas", 2, 64)
|
|
db_connection_limit = _bounded_integer(root, "db_connection_limit", 50, 10000)
|
|
network = _parse_network(_mapping(root.get("network"), "network"))
|
|
image = _parse_image(_mapping(root.get("image"), "image"))
|
|
k3s = _parse_k3s(_mapping(root.get("k3s"), "k3s"))
|
|
release = _parse_release(_mapping(root.get("release"), "release"))
|
|
raw_nodes = root.get("nodes")
|
|
if not isinstance(raw_nodes, list) or not raw_nodes:
|
|
raise LabConfigError("nodes must be a non-empty array of tables")
|
|
nodes = tuple(_parse_node(item, index=index) for index, item in enumerate(raw_nodes))
|
|
config = LabConfig(
|
|
source=source,
|
|
schema_version=schema_version,
|
|
name=name,
|
|
mode=mode,
|
|
state_directory=state_directory,
|
|
vm_image_directory=vm_image_directory.rstrip("/"),
|
|
ssh_user=ssh_user,
|
|
ssh_private_key=ssh_private_key,
|
|
ssh_public_key=ssh_public_key,
|
|
namespace=namespace,
|
|
public_host=public_host,
|
|
s3_host=s3_host,
|
|
ingress_class=ingress_class,
|
|
module_set=module_set,
|
|
api_replicas=api_replicas,
|
|
web_replicas=web_replicas,
|
|
worker_replicas=worker_replicas,
|
|
db_connection_limit=db_connection_limit,
|
|
network=network,
|
|
image=image,
|
|
k3s=k3s,
|
|
release=release,
|
|
nodes=nodes,
|
|
)
|
|
_validate_topology(config)
|
|
return config
|
|
|
|
|
|
def _parse_network(raw: Mapping[str, Any]) -> NetworkConfig:
|
|
_only_keys(raw, {"prefix_length", "gateway", "dns_servers", "bridge"}, "network")
|
|
prefix_length = _bounded_integer(raw, "prefix_length", 8, 30)
|
|
gateway = _ipv4(raw, "gateway")
|
|
dns_raw = raw.get("dns_servers")
|
|
if not isinstance(dns_raw, list) or not dns_raw or len(dns_raw) > 4:
|
|
raise LabConfigError("network.dns_servers must contain 1-4 IPv4 addresses")
|
|
dns_servers = tuple(_ipv4_value(value, "network.dns_servers") for value in dns_raw)
|
|
bridge = _pattern(raw, "bridge", _BRIDGE)
|
|
return NetworkConfig(prefix_length, gateway, dns_servers, bridge)
|
|
|
|
|
|
def _parse_image(raw: Mapping[str, Any]) -> ImageConfig:
|
|
_only_keys(raw, {"url", "sha256"}, "image")
|
|
return ImageConfig(
|
|
url=_https_url(raw, "url"),
|
|
sha256=_pattern(raw, "sha256", _SHA256),
|
|
)
|
|
|
|
|
|
def _parse_k3s(raw: Mapping[str, Any]) -> K3sConfig:
|
|
_only_keys(
|
|
raw,
|
|
{
|
|
"version",
|
|
"binary_url",
|
|
"binary_sha256",
|
|
"install_script_url",
|
|
"install_script_sha256",
|
|
},
|
|
"k3s",
|
|
)
|
|
return K3sConfig(
|
|
version=_pattern(raw, "version", _K3S_VERSION),
|
|
binary_url=_https_url(raw, "binary_url"),
|
|
binary_sha256=_pattern(raw, "binary_sha256", _SHA256),
|
|
install_script_url=_https_url(raw, "install_script_url"),
|
|
install_script_sha256=_pattern(raw, "install_script_sha256", _SHA256),
|
|
)
|
|
|
|
|
|
def _parse_release(raw: Mapping[str, Any]) -> ReleaseConfig:
|
|
_only_keys(
|
|
raw,
|
|
{"manifest_url", "manifest_sha256", "keyring_url", "keyring_sha256", "channel"},
|
|
"release",
|
|
)
|
|
return ReleaseConfig(
|
|
manifest_url=_https_url(raw, "manifest_url"),
|
|
manifest_sha256=_pattern(raw, "manifest_sha256", _SHA256),
|
|
keyring_url=_https_url(raw, "keyring_url"),
|
|
keyring_sha256=_pattern(raw, "keyring_sha256", _SHA256),
|
|
channel=_pattern(raw, "channel", _NAME),
|
|
)
|
|
|
|
|
|
def _parse_node(value: object, *, index: int) -> LabNode:
|
|
raw = _mapping(value, f"nodes[{index}]")
|
|
_only_keys(
|
|
raw,
|
|
{
|
|
"name",
|
|
"role",
|
|
"address",
|
|
"hypervisor",
|
|
"failure_domain",
|
|
"mac_address",
|
|
"cpus",
|
|
"memory_mib",
|
|
"disk_gib",
|
|
},
|
|
f"nodes[{index}]",
|
|
)
|
|
hypervisor = _string(raw, "hypervisor")
|
|
if hypervisor != "local" and not _SSH_TARGET.fullmatch(hypervisor):
|
|
raise LabConfigError(
|
|
f"nodes[{index}].hypervisor must be 'local' or a simple SSH target"
|
|
)
|
|
mac_address = _string(raw, "mac_address").lower()
|
|
try:
|
|
octets = mac_address.split(":")
|
|
valid_mac = len(octets) == 6 and all(
|
|
len(octet) == 2 and 0 <= int(octet, 16) <= 255 for octet in octets
|
|
)
|
|
except ValueError:
|
|
valid_mac = False
|
|
if not valid_mac:
|
|
raise LabConfigError(f"nodes[{index}].mac_address is not a canonical MAC address")
|
|
return LabNode(
|
|
name=_pattern(raw, "name", _NAME),
|
|
role=_choice(raw, "role", _ROLE),
|
|
address=_ipv4(raw, "address"),
|
|
hypervisor=hypervisor,
|
|
failure_domain=_pattern(raw, "failure_domain", _NAME),
|
|
mac_address=mac_address,
|
|
cpus=_bounded_integer(raw, "cpus", 1, 64),
|
|
memory_mib=_bounded_integer(raw, "memory_mib", 2048, 262144),
|
|
disk_gib=_bounded_integer(raw, "disk_gib", 16, 4096),
|
|
)
|
|
|
|
|
|
def _validate_topology(config: LabConfig) -> None:
|
|
names = [node.name for node in config.nodes]
|
|
addresses = [node.address for node in config.nodes]
|
|
mac_addresses = [node.mac_address for node in config.nodes]
|
|
for label, values in (
|
|
("node names", names),
|
|
("node addresses", addresses),
|
|
("node MAC addresses", mac_addresses),
|
|
):
|
|
if len(values) != len(set(values)):
|
|
raise LabConfigError(f"{label} must be unique")
|
|
too_long = [
|
|
node.name
|
|
for node in config.nodes
|
|
if len(f"{config.name}-{node.name}") > 63
|
|
]
|
|
if too_long:
|
|
raise LabConfigError(
|
|
"lab name plus node name must fit the 63-character libvirt domain limit: "
|
|
+ ", ".join(too_long)
|
|
)
|
|
if len(config.controls) not in {1, 3}:
|
|
raise LabConfigError("the lab requires exactly one or three control-plane nodes")
|
|
if len(config.workers) < 2:
|
|
raise LabConfigError("the lab requires at least two worker nodes")
|
|
if sum(node.role == "state" for node in config.nodes) != 1:
|
|
raise LabConfigError("the lab requires exactly one external shared-state node")
|
|
network = ipaddress.ip_network(
|
|
f"{config.network.gateway}/{config.network.prefix_length}", strict=False
|
|
)
|
|
if any(node.address not in network for node in config.nodes):
|
|
raise LabConfigError("every node address must be in the configured IPv4 network")
|
|
if config.network.gateway in addresses:
|
|
raise LabConfigError("the network gateway cannot also be a node address")
|
|
if config.mode == "acceptance" and not config.evidence_capable:
|
|
raise LabConfigError(
|
|
"acceptance mode requires each worker and the shared-state node to use "
|
|
"distinct hypervisors and failure_domain values"
|
|
)
|
|
|
|
|
|
def _mapping(value: object, label: str) -> Mapping[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise LabConfigError(f"{label} must be a table")
|
|
return value
|
|
|
|
|
|
def _only_keys(raw: Mapping[str, Any], allowed: set[str], label: str) -> None:
|
|
unexpected = sorted(set(raw) - allowed)
|
|
if unexpected:
|
|
raise LabConfigError(f"{label} contains unsupported keys: {', '.join(unexpected)}")
|
|
|
|
|
|
def _string(raw: Mapping[str, Any], key: str) -> str:
|
|
value = raw.get(key)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise LabConfigError(f"{key} must be a non-empty string")
|
|
return value.strip()
|
|
|
|
|
|
def _integer(raw: Mapping[str, Any], key: str) -> int:
|
|
value = raw.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool):
|
|
raise LabConfigError(f"{key} must be an integer")
|
|
return value
|
|
|
|
|
|
def _bounded_integer(raw: Mapping[str, Any], key: str, minimum: int, maximum: int) -> int:
|
|
value = _integer(raw, key)
|
|
if not minimum <= value <= maximum:
|
|
raise LabConfigError(f"{key} must be between {minimum} and {maximum}")
|
|
return value
|
|
|
|
|
|
def _pattern(raw: Mapping[str, Any], key: str, pattern: re.Pattern[str]) -> str:
|
|
value = _string(raw, key)
|
|
if pattern.fullmatch(value) is None:
|
|
raise LabConfigError(f"{key} has an unsupported format")
|
|
return value
|
|
|
|
|
|
def _choice(raw: Mapping[str, Any], key: str, choices: set[str]) -> str:
|
|
value = _string(raw, key)
|
|
if value not in choices:
|
|
raise LabConfigError(f"{key} must be one of: {', '.join(sorted(choices))}")
|
|
return value
|
|
|
|
|
|
def _ipv4(raw: Mapping[str, Any], key: str) -> ipaddress.IPv4Address:
|
|
return _ipv4_value(_string(raw, key), key)
|
|
|
|
|
|
def _ipv4_value(value: object, label: str) -> ipaddress.IPv4Address:
|
|
if not isinstance(value, str):
|
|
raise LabConfigError(f"{label} must contain strings")
|
|
try:
|
|
parsed = ipaddress.ip_address(value)
|
|
except ValueError as exc:
|
|
raise LabConfigError(f"{label} contains an invalid IP address") from exc
|
|
if not isinstance(parsed, ipaddress.IPv4Address):
|
|
raise LabConfigError(f"{label} supports IPv4 only in schema version 1")
|
|
return parsed
|
|
|
|
|
|
def _path(raw: Mapping[str, Any], key: str, base: Path) -> Path:
|
|
value = Path(_string(raw, key)).expanduser()
|
|
return (value if value.is_absolute() else base / value).resolve()
|
|
|
|
|
|
def _absolute_posix_path(raw: Mapping[str, Any], key: str) -> str:
|
|
value = _string(raw, key)
|
|
if not value.startswith("/") or ".." in Path(value).parts:
|
|
raise LabConfigError(f"{key} must be an absolute path without '..'")
|
|
return value
|
|
|
|
|
|
def _https_url(raw: Mapping[str, Any], key: str) -> str:
|
|
value = _string(raw, key)
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
|
|
raise LabConfigError(f"{key} must be a credential-free HTTPS URL")
|
|
if parsed.fragment:
|
|
raise LabConfigError(f"{key} must not contain a fragment")
|
|
return value
|