"""Conservative v1 field validation, not a full Kubernetes schema validator.""" import re from decimal import Context, Decimal, localcontext from typing import Never from .models import ErrorCode, FieldKind, Scalar, YamlEditError, is_scalar _MAX_QUANTITY = Decimal(2**63 - 1) _MAX_REPLICAS = 2**31 - 1 _MAX_QUANTITY_TEXT = 128 _MAX_IMAGE_TEXT = 512 _MAX_IMAGE_NAME = 255 _MILLICORES = 1000 _QUANTITY_PRECISION = 256 _DECIMAL = r"(?:\d+(?:\.\d+)?|\.\d+)" _CPU = re.compile(rf"({_DECIMAL})(m)?", re.ASCII) _MEMORY = re.compile(rf"({_DECIMAL})([KMGTPE]i|[kMGTPE])?", re.ASCII) _COMPONENT = r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*" _HOST = r"[a-z0-9]+(?:[.-][a-z0-9]+)*(?::[0-9]{1,5})?" _IMAGE = re.compile( rf"(?P(?:{_HOST}/)?{_COMPONENT}(?:/{_COMPONENT})*)" r"(?::[A-Za-z0-9_][A-Za-z0-9_.-]{0,127})?" r"(?:@sha256:[a-fA-F0-9]{64})?" ) def _invalid(message: str) -> Never: raise YamlEditError(ErrorCode.INVALID_VALUE, message) def quantity(value: Scalar, *, cpu: bool) -> Decimal: """Normalize our supported quantity subset to cores or bytes, exactly.""" # Check integers before decimal formatting, which has a process-wide digit limit. if type(value) is int and not 0 <= value <= _MAX_QUANTITY: _invalid("Resource quantity must be nonnegative and at most 2^63-1.") text = str(value) if len(text) > _MAX_QUANTITY_TEXT: _invalid("Resource quantity is too long.") match = (_CPU if cpu else _MEMORY).fullmatch(text) if match is None: unit_help = ( "decimal cores or millicores (0.5, 250m)" if cpu else "bytes, SI or binary units (500M, 256Mi)" ) _invalid(f"Resource quantity must be nonnegative {unit_help}.") # Do not inherit caller precision, exponent limits, or arithmetic traps. with localcontext(Context(prec=_QUANTITY_PRECISION)): amount = Decimal(match[1]) suffix = match[2] or "" if cpu: if suffix == "m": amount /= _MILLICORES if amount * _MILLICORES != (amount * _MILLICORES).to_integral_value(): _invalid("CPU precision cannot be finer than one millicore.") elif suffix: power = "kMGTPE".index("k" if suffix[0] == "K" else suffix[0]) + 1 amount *= Decimal(1024 if suffix.endswith("i") else 1000) ** power if amount > _MAX_QUANTITY: _invalid("Resource quantity exceeds 2^63-1.") return amount def validate_value(kind: FieldKind, value: object) -> Scalar: """Validate a requested or existing field value without coercing its type.""" if not is_scalar(value): _invalid("Expected a finite number or string, not a boolean, null, or collection.") if kind == FieldKind.REPLICAS: if type(value) is not int or not 0 <= value <= _MAX_REPLICAS: _invalid("Replicas must be an integer from 0 through 2147483647.") elif kind == FieldKind.IMAGE: match = ( _IMAGE.fullmatch(value) if isinstance(value, str) and len(value) <= _MAX_IMAGE_TEXT else None ) if match is None or len(match["name"]) > _MAX_IMAGE_NAME: _invalid("Expected an image reference such as nginx:1.27 or ghcr.io/acme/api:v2.") else: quantity(value, cpu=kind in (FieldKind.CPU_REQUEST, FieldKind.CPU_LIMIT)) return value