File size: 3,391 Bytes
8d20d7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """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."""
# 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
|