132 lines
5.3 KiB
Python
132 lines
5.3 KiB
Python
"""Чистые функции валидации состояния (стадия ``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
|