"""Validation and unit conversion for command-line resource options.""" from __future__ import annotations import math import re from collections.abc import Sequence from decimal import Decimal, InvalidOperation from .errors import ValidationError from .domain import PortRule _RAM_PATTERN = re.compile(r"\s*(\d+(?:\.\d+)?)\s*([BKMG]i?B?)?\s*", re.IGNORECASE) _USERNAME_PATTERN = re.compile(r"[a-z_][a-z0-9_-]{0,31}\$?") _MIB_FACTORS = { "b": Decimal(1) / Decimal(1024 * 1024), "k": Decimal(1) / Decimal(1024), "kb": Decimal(1) / Decimal(1024), "ki": Decimal(1) / Decimal(1024), "kib": Decimal(1) / Decimal(1024), "m": Decimal(1), "mb": Decimal(1), "mi": Decimal(1), "mib": Decimal(1), "g": Decimal(1024), "gb": Decimal(1024), "gi": Decimal(1024), "gib": Decimal(1024), } def parse_ram(value: str) -> int: """Parse a MiB-default RAM value, including B/K/M/G suffixes.""" match = _RAM_PATTERN.fullmatch(value) if not match: raise ValidationError(f"invalid RAM value: {value}") try: amount = Decimal(match.group(1)) except InvalidOperation as error: raise ValidationError(f"invalid RAM value: {value}") from error unit = (match.group(2) or "MiB").lower() mib = int(amount * _MIB_FACTORS[unit]) if mib < 128: raise ValidationError("RAM must be at least 128 MiB") return mib def parse_cpu(value: str) -> float: """Parse a positive finite CPU capacity request.""" try: cpu = float(value) except ValueError as error: raise ValidationError(f"invalid CPU value: {value}") from error if not math.isfinite(cpu) or cpu <= 0: raise ValidationError("CPU must be a finite value greater than 0") return cpu def parse_username(value: str) -> str: """Validate a conventional Linux account name.""" if not _USERNAME_PATTERN.fullmatch(value): raise ValidationError( "username must start with a lowercase letter or underscore and contain " "at most 32 lowercase letters, digits, underscores, or hyphens" ) return value def parse_password(value: str) -> str: """Validate a guest password without including it in error messages.""" if not value: raise ValidationError("password must not be empty") if len(value) > 128: raise ValidationError("password must be at most 128 characters") if any(ord(character) < 32 or ord(character) == 127 for character in value): raise ValidationError("password must not contain control characters") return value def parse_name(value: str) -> str: if re.fullmatch(r"[a-z][a-z0-9-]{0,62}", value) is None: raise ValidationError("VM name must be 1-63 lowercase letters, digits, or hyphens, starting with a letter") if re.fullmatch(r"vm-[0-9a-f]{32}", value): raise ValidationError("VM names must not impersonate generated VM IDs") return value def parse_ports(values: Sequence[str]) -> tuple[PortRule, ...]: ports: dict[int, PortRule] = {} for value in values: for item in value.split(","): match = re.fullmatch(r"\s*(\d+)(?::(TCP|HTTP|HTTPS))?\s*", item, re.IGNORECASE) if match is None: raise ValidationError(f"invalid private port: {item!r}; use PORT[:TCP|HTTP|HTTPS]") rule = PortRule(int(match.group(1)), match.group(2) or "TCP") if rule.port in ports and ports[rule.port] != rule: raise ValidationError(f"conflicting labels for TCP port {rule.port}") ports[rule.port] = rule return tuple(ports[number] for number in sorted(ports)) def parse_timeout(value: str) -> float: try: timeout = float(value) except (ValueError, TypeError) as error: raise ValidationError("timeout must be a finite non-negative number of seconds") from error if not math.isfinite(timeout) or timeout < 0: raise ValidationError("timeout must be a finite non-negative number of seconds") return timeout