122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
"""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 not isinstance(value, str) or 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, ...]:
|
|
if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
|
|
raise ValidationError("ports must be a sequence of PORT[:TYPE] strings")
|
|
ports: dict[int, PortRule] = {}
|
|
for value in values:
|
|
if not isinstance(value, str):
|
|
raise ValidationError("each port specification must be a string")
|
|
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
|