| """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<name>(?:{_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.""" |
| |
| 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}.") |
| |
| 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 |
|
|
