Files
RelayOps/backend/app/schemas/validators.py
2026-06-07 02:32:28 +03:00

71 lines
2.4 KiB
Python

import ipaddress
import re
USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$")
SSH_USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$")
HOST_LABEL_PATTERN = re.compile(r"^[A-Za-z0-9-]{1,63}$")
def normalize_required_text(value: str, *, field_name: str) -> str:
normalized = value.strip()
if not normalized:
raise ValueError(f"{field_name} must not be empty.")
return normalized
def normalize_optional_text(value: str | None) -> str | None:
if value is None:
return None
normalized = value.strip()
return normalized or None
def validate_username(value: str) -> str:
normalized = normalize_required_text(value, field_name="username")
if not USERNAME_PATTERN.fullmatch(normalized):
raise ValueError(
"username may contain only letters, digits, dot, underscore, and dash."
)
return normalized
def validate_ssh_username(value: str | None) -> str | None:
normalized = normalize_optional_text(value)
if normalized is None:
return None
if not SSH_USERNAME_PATTERN.fullmatch(normalized):
raise ValueError(
"ssh_username may contain only letters, digits, dot, underscore, and dash."
)
return normalized
def validate_host(value: str) -> str:
normalized = normalize_required_text(value, field_name="host")
if any(fragment in normalized for fragment in ("://", "/", "@", " ")):
raise ValueError("host must be a hostname or IP address without URL formatting.")
try:
ipaddress.ip_address(normalized)
return normalized
except ValueError:
labels = normalized.split(".")
if any(not label for label in labels):
raise ValueError("host contains an empty hostname label.") from None
if not all(HOST_LABEL_PATTERN.fullmatch(label) for label in labels):
raise ValueError("host contains invalid hostname characters.") from None
if any(label.startswith("-") or label.endswith("-") for label in labels):
raise ValueError("host labels must not start or end with a dash.") from None
return normalized
def validate_log_source(value: str | None) -> str | None:
normalized = normalize_optional_text(value)
if normalized is None:
return None
if not normalized.startswith("file:/"):
raise ValueError("log_source must use the format 'file:/absolute/path'.")
return normalized