"""Чистые функции валидации состояния (стадия ``validate`` перед apply, issue #20). Валидация отделена от БД: те же функции вызывают и репозиторий (точечные мутации), и движок атомарного применения (``validate -> stage -> apply``). """ from __future__ import annotations import ipaddress import re from .errors import ValidationError from .models import DEVICE_POLICIES, RULE_KINDS, Mode _NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_.:-]{0,62}$") _DOMAIN_RE = re.compile(r"^(\*\.)?([a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$") _IPSET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$") _MAC_RE = re.compile(r"^(?:[0-9a-f]{2}:){5}[0-9a-f]{2}$") _CRON_FIELD_RE = re.compile(r"^[0-9*,/-]{1,20}$") _PROVIDER_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,31}$") def normalize_mac(mac: str) -> str: """Привести MAC к канонической форме ``aa:bb:cc:dd:ee:ff``.""" cleaned = mac.strip().replace("-", ":").lower() if len(cleaned) == 12 and ":" not in cleaned: cleaned = ":".join(cleaned[i : i + 2] for i in range(0, 12, 2)) if not _MAC_RE.match(cleaned): raise ValidationError("mac", f"некорректный MAC-адрес: {mac!r}") return cleaned def validate_name(field: str, value: str) -> str: """Проверить идентификатор (имя правила/конфига устройства).""" if not _NAME_RE.match(value.strip()): raise ValidationError( field, f"допустимы строчные латинские буквы, цифры и «_.:-», до 63 символов: {value!r}" ) return value.strip() def validate_rule_kind(kind: str) -> str: if kind not in RULE_KINDS: raise ValidationError("kind", f"ожидается одно из {RULE_KINDS}, получено {kind!r}") return kind def validate_match(kind: str, match: str) -> str: """Проверить образец правила по типу записи.""" validate_rule_kind(kind) value = match.strip() if not value: raise ValidationError("match", "образец правила не может быть пустым") if kind == "domain": if not _DOMAIN_RE.match(value): raise ValidationError("match", f"некорректный домен: {value!r}") elif kind == "cidr": try: ipaddress.ip_network(value, strict=False) except ValueError as exc: raise ValidationError("match", f"некорректная сеть CIDR: {value!r}") from exc elif not _IPSET_RE.match(value): raise ValidationError("match", f"некорректное имя ipset: {value!r}") return value def validate_exception_match(match: str) -> str: """Проверить образец исключения: домен, имя ipset или сеть CIDR.""" value = match.strip() if not value: raise ValidationError("match", "образец исключения не может быть пустым") if _DOMAIN_RE.match(value) or _IPSET_RE.match(value): return value try: ipaddress.ip_network(value, strict=False) except ValueError as exc: raise ValidationError( "match", f"ожидается домен, сеть CIDR или имя ipset: {match!r}" ) from exc return value def validate_target(target: str) -> str: value = target.strip() if not _IPSET_RE.match(value): raise ValidationError("target", f"некорректная цель маршрутизации: {target!r}") return value def validate_priority(priority: int) -> int: if not isinstance(priority, int) or isinstance(priority, bool): raise ValidationError("priority", f"ожидается целое число, получено {priority!r}") if not 0 <= priority <= 65535: raise ValidationError("priority", f"вне диапазона 0..65535: {priority}") return priority def validate_mode(mode: str | Mode) -> Mode: try: return Mode(mode) except ValueError as exc: raise ValidationError( "mode", f"ожидается одно из {Mode.values()}, получено {mode!r}" ) from exc def validate_cron(cron: str) -> str: fields = cron.split() if len(fields) != 5 or not all(_CRON_FIELD_RE.match(part) for part in fields): raise ValidationError("cron", f"ожидается 5-полевое cron-выражение: {cron!r}") return " ".join(fields) def validate_provider(provider: str) -> str: value = provider.strip().lower() if not _PROVIDER_RE.match(value): raise ValidationError("provider", f"некорректное имя провайдера GeoIP: {provider!r}") return value def validate_device_policy(policy: str) -> str: if policy not in DEVICE_POLICIES: raise ValidationError("policy", f"ожидается одно из {DEVICE_POLICIES}, получено {policy!r}") return policy def validate_config_path(path: str) -> str: value = path.strip() if not value.startswith("/") or ".." in value.split("/"): raise ValidationError("path", f"ожидается абсолютный путь без «..»: {path!r}") return value