twothatITandClaude Sonnet 5
189c49d505
Some customers only regain access to their NetBird web UI after a manual container restart, and until now that required someone to notice and do it by hand. The scheduler now optionally probes each customer's actual dashboard URL over HTTP(S) every poll interval (not just Docker container state, since a container can be "running" while the web UI behind it still doesn't respond) and restarts that customer's containers if it doesn't answer, gated by a per-deployment cooldown to avoid restart loops on a persistently broken customer. New system setting "Automatic Restart of Unreachable Containers" (default off) with a configurable cooldown. Every automatic restart is logged in the affected customer's own log, same as a manual restart. Co-Authored-By: Claude Sonnet 5 <[email protected]>
317 lines
12 KiB
Python
317 lines
12 KiB
Python
"""Input validation with Pydantic models for all API endpoints."""
|
|
|
|
import re
|
|
from typing import Optional
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth
|
|
# ---------------------------------------------------------------------------
|
|
class LoginRequest(BaseModel):
|
|
"""Login credentials."""
|
|
|
|
username: str = Field(..., min_length=1, max_length=100)
|
|
password: str = Field(..., min_length=1)
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
"""Password change payload."""
|
|
|
|
current_password: str = Field(..., min_length=1)
|
|
new_password: str = Field(..., min_length=12, max_length=128)
|
|
|
|
|
|
class MfaTokenRequest(BaseModel):
|
|
"""Request containing only an MFA token (for setup initiation)."""
|
|
|
|
mfa_token: str = Field(..., min_length=1)
|
|
|
|
|
|
class MfaVerifyRequest(BaseModel):
|
|
"""MFA TOTP verification payload."""
|
|
|
|
mfa_token: str = Field(..., min_length=1)
|
|
totp_code: str = Field(..., min_length=6, max_length=6)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Customer
|
|
# ---------------------------------------------------------------------------
|
|
class CustomerCreate(BaseModel):
|
|
"""Payload to create a new customer."""
|
|
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
company: Optional[str] = Field(None, max_length=255)
|
|
subdomain: str = Field(..., min_length=1, max_length=63)
|
|
email: str = Field(..., max_length=255)
|
|
max_devices: int = Field(default=20, ge=1, le=10000)
|
|
notes: Optional[str] = None
|
|
|
|
@field_validator("subdomain")
|
|
@classmethod
|
|
def validate_subdomain(cls, v: str) -> str:
|
|
"""Subdomain must be lowercase alphanumeric + hyphens, no leading/trailing hyphen."""
|
|
v = v.lower().strip()
|
|
if not re.match(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$", v):
|
|
raise ValueError(
|
|
"Subdomain must be lowercase, alphanumeric with hyphens, "
|
|
"2-63 chars, no leading/trailing hyphen."
|
|
)
|
|
return v
|
|
|
|
@field_validator("email")
|
|
@classmethod
|
|
def validate_email(cls, v: str) -> str:
|
|
"""Basic email format check."""
|
|
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid email address.")
|
|
return v.lower().strip()
|
|
|
|
|
|
class CustomerUpdate(BaseModel):
|
|
"""Payload to update an existing customer."""
|
|
|
|
name: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
company: Optional[str] = Field(None, max_length=255)
|
|
email: Optional[str] = Field(None, max_length=255)
|
|
max_devices: Optional[int] = Field(None, ge=1, le=10000)
|
|
notes: Optional[str] = None
|
|
status: Optional[str] = None
|
|
|
|
@field_validator("email")
|
|
@classmethod
|
|
def validate_email(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Basic email format check."""
|
|
if v is None:
|
|
return v
|
|
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid email address.")
|
|
return v.lower().strip()
|
|
|
|
@field_validator("status")
|
|
@classmethod
|
|
def validate_status(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Status must be one of the allowed values."""
|
|
if v is None:
|
|
return v
|
|
allowed = {"active", "inactive", "deploying", "error"}
|
|
if v not in allowed:
|
|
raise ValueError(f"Status must be one of: {', '.join(sorted(allowed))}")
|
|
return v
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Settings
|
|
# ---------------------------------------------------------------------------
|
|
class SystemConfigUpdate(BaseModel):
|
|
"""Payload to update system configuration."""
|
|
|
|
base_domain: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
admin_email: Optional[str] = Field(None, max_length=255)
|
|
npm_api_url: Optional[str] = Field(None, max_length=500)
|
|
npm_api_email: Optional[str] = Field(None, max_length=255) # NPM login email
|
|
npm_api_password: Optional[str] = None # NPM login password, encrypted before storage
|
|
netbird_management_image: Optional[str] = Field(None, max_length=255)
|
|
netbird_signal_image: Optional[str] = Field(None, max_length=255)
|
|
netbird_relay_image: Optional[str] = Field(None, max_length=255)
|
|
netbird_dashboard_image: Optional[str] = Field(None, max_length=255)
|
|
data_dir: Optional[str] = Field(None, max_length=500)
|
|
docker_network: Optional[str] = Field(None, max_length=100)
|
|
relay_base_port: Optional[int] = Field(None, ge=1024, le=65535)
|
|
dashboard_base_port: Optional[int] = Field(None, ge=1024, le=65535)
|
|
branding_name: Optional[str] = Field(None, max_length=255)
|
|
branding_subtitle: Optional[str] = Field(None, max_length=255)
|
|
default_language: Optional[str] = Field(None, max_length=10)
|
|
ssl_mode: Optional[str] = Field(None, max_length=20)
|
|
wildcard_cert_id: Optional[int] = Field(None, ge=0)
|
|
mfa_enabled: Optional[bool] = None
|
|
azure_enabled: Optional[bool] = None
|
|
azure_tenant_id: Optional[str] = Field(None, max_length=255)
|
|
azure_client_id: Optional[str] = Field(None, max_length=255)
|
|
azure_client_secret: Optional[str] = None # encrypted before storage
|
|
azure_allowed_group_id: Optional[str] = Field(
|
|
None, max_length=255,
|
|
description="Azure AD group object ID. If set, only members of this group can log in."
|
|
)
|
|
# Windows DNS
|
|
dns_enabled: Optional[bool] = None
|
|
dns_server: Optional[str] = Field(None, max_length=255)
|
|
dns_username: Optional[str] = Field(None, max_length=255)
|
|
dns_password: Optional[str] = None # plaintext, encrypted before storage
|
|
dns_zone: Optional[str] = Field(None, max_length=255)
|
|
dns_record_ip: Optional[str] = Field(None, max_length=45)
|
|
# LDAP
|
|
ldap_enabled: Optional[bool] = None
|
|
ldap_server: Optional[str] = Field(None, max_length=255)
|
|
ldap_port: Optional[int] = Field(None, ge=1, le=65535)
|
|
ldap_use_ssl: Optional[bool] = None
|
|
ldap_bind_dn: Optional[str] = Field(None, max_length=500)
|
|
ldap_bind_password: Optional[str] = None # plaintext, encrypted before storage
|
|
ldap_base_dn: Optional[str] = Field(None, max_length=500)
|
|
ldap_user_filter: Optional[str] = Field(None, max_length=255)
|
|
ldap_group_dn: Optional[str] = Field(None, max_length=500)
|
|
# Update management
|
|
git_repo_url: Optional[str] = Field(None, max_length=500)
|
|
git_branch: Optional[str] = Field(None, max_length=100)
|
|
git_token: Optional[str] = None # plaintext, encrypted before storage
|
|
# Automatic NetBird image update check/apply
|
|
auto_update_check_enabled: Optional[bool] = None
|
|
auto_update_check_time: Optional[str] = Field(None, max_length=5)
|
|
auto_update_apply_enabled: Optional[bool] = None
|
|
# Master default for the NetBird client (peer) automatic-updates feature
|
|
netbird_client_auto_update_version: Optional[str] = Field(None, max_length=50)
|
|
netbird_client_auto_update_always: Optional[bool] = None
|
|
# Automatic restart of unreachable customer containers
|
|
auto_restart_unreachable_enabled: Optional[bool] = None
|
|
auto_restart_cooldown_minutes: Optional[int] = Field(None, ge=1, le=1440)
|
|
|
|
@field_validator("auto_update_check_time")
|
|
@classmethod
|
|
def validate_auto_update_check_time(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Must be HH:MM in 24h format."""
|
|
if v is None:
|
|
return v
|
|
import re
|
|
if not re.fullmatch(r"([01]\d|2[0-3]):[0-5]\d", v):
|
|
raise ValueError("auto_update_check_time must be in HH:MM 24h format")
|
|
return v
|
|
|
|
@field_validator("ssl_mode")
|
|
@classmethod
|
|
def validate_ssl_mode(cls, v: Optional[str]) -> Optional[str]:
|
|
"""SSL mode must be 'letsencrypt' or 'wildcard'."""
|
|
if v is None:
|
|
return v
|
|
allowed = {"letsencrypt", "wildcard"}
|
|
if v not in allowed:
|
|
raise ValueError(f"ssl_mode must be one of: {', '.join(sorted(allowed))}")
|
|
return v
|
|
|
|
@field_validator("base_domain")
|
|
@classmethod
|
|
def validate_domain(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Validate domain format."""
|
|
if v is None:
|
|
return v
|
|
pattern = r"^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid domain format.")
|
|
return v.lower().strip()
|
|
|
|
@field_validator("npm_api_url")
|
|
@classmethod
|
|
def validate_npm_url(cls, v: Optional[str]) -> Optional[str]:
|
|
"""NPM URL must start with http(s)://."""
|
|
if v is None:
|
|
return v
|
|
if not re.match(r"^https?://", v):
|
|
raise ValueError("NPM API URL must start with http:// or https://")
|
|
return v.rstrip("/")
|
|
|
|
@field_validator("admin_email")
|
|
@classmethod
|
|
def validate_email(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Validate admin email."""
|
|
if v is None:
|
|
return v
|
|
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid email address.")
|
|
return v.lower().strip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NetBird client (peer) automatic updates
|
|
# ---------------------------------------------------------------------------
|
|
class NetbirdClientAutoUpdatePayload(BaseModel):
|
|
"""Push a client auto-update version/mode to one or all customers."""
|
|
|
|
version: str = Field(..., max_length=50, description="'latest', 'disabled', or a version e.g. '0.61.0'")
|
|
always: bool = False
|
|
|
|
|
|
class NetbirdApiTokenPayload(BaseModel):
|
|
"""Manually register a NetBird Personal Access Token for a customer.
|
|
|
|
Needed for customers deployed before automatic PAT capture existed —
|
|
create a PAT once in that customer's own NetBird dashboard
|
|
(Settings > Service Users / Personal Access Tokens) and paste it here.
|
|
"""
|
|
|
|
token: str = Field(..., min_length=10, max_length=500)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Users
|
|
# ---------------------------------------------------------------------------
|
|
class UserCreate(BaseModel):
|
|
"""Payload to create a new local user."""
|
|
|
|
username: str = Field(..., min_length=3, max_length=100)
|
|
password: str = Field(..., min_length=8, max_length=128)
|
|
email: Optional[str] = Field(None, max_length=255)
|
|
default_language: Optional[str] = Field(None, max_length=10)
|
|
|
|
@field_validator("username")
|
|
@classmethod
|
|
def validate_username(cls, v: str) -> str:
|
|
if not re.match(r"^[a-zA-Z0-9_.-]+$", v):
|
|
raise ValueError("Username may only contain letters, digits, dots, hyphens, and underscores.")
|
|
return v.strip()
|
|
|
|
@field_validator("email")
|
|
@classmethod
|
|
def validate_user_email(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return v
|
|
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid email address.")
|
|
return v.lower().strip()
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
"""Payload to update an existing user."""
|
|
|
|
email: Optional[str] = Field(None, max_length=255)
|
|
is_active: Optional[bool] = None
|
|
role: Optional[str] = Field(None, max_length=20)
|
|
default_language: Optional[str] = Field(None, max_length=10)
|
|
|
|
@field_validator("email")
|
|
@classmethod
|
|
def validate_user_email(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return v
|
|
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
|
if not re.match(pattern, v):
|
|
raise ValueError("Invalid email address.")
|
|
return v.lower().strip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Query params
|
|
# ---------------------------------------------------------------------------
|
|
class CustomerListParams(BaseModel):
|
|
"""Query parameters for listing customers."""
|
|
|
|
page: int = Field(default=1, ge=1)
|
|
per_page: int = Field(default=25, ge=1, le=100)
|
|
search: Optional[str] = None
|
|
status: Optional[str] = None
|
|
|
|
@field_validator("status")
|
|
@classmethod
|
|
def validate_status(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Filter status validation."""
|
|
if v is None or v == "":
|
|
return None
|
|
allowed = {"active", "inactive", "deploying", "error"}
|
|
if v not in allowed:
|
|
raise ValueError(f"Status must be one of: {', '.join(sorted(allowed))}")
|
|
return v
|