OpenVPN-Monitoring-Simple/APP_PROFILER/schemas.py

import ipaddress
import re
from pydantic import BaseModel, Field, field_validator, model_validator
from services import validation as v
from typing import List, Optional, Literal
from datetime import datetime

# --- PKI Settings Schemas ---
class PKISettingBase(BaseModel):
    fqdn_ca: str = "ovpn-ca"
    fqdn_server: str = "ovpn-srv"
    easyrsa_dn: str = "cn_only"
    easyrsa_req_country: str = "RU"
    easyrsa_req_province: str = "Moscow"
    easyrsa_req_city: str = "Moscow"
    easyrsa_req_org: str = "SomeORG"
    easyrsa_req_email: str = "info@someorg.local"
    easyrsa_req_ou: str = "IT"
    easyrsa_key_size: int = 2048
    easyrsa_ca_expire: int = 3650
    easyrsa_cert_expire: int = 3649
    easyrsa_cert_renew: int = 30
    easyrsa_crl_days: int = 3649
    easyrsa_batch: bool = True

class PKISettingUpdate(PKISettingBase):
    @field_validator("fqdn_ca", "fqdn_server")
    @classmethod
    def _check_fqdn(cls, val):
        if not re.match(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$", val) or ".." in val:
            raise ValueError("Invalid name (letters, digits, . _ -; max 64)")
        return val

    @field_validator("easyrsa_dn")
    @classmethod
    def _check_dn(cls, val):
        if val not in ("cn_only", "org"):
            raise ValueError("easyrsa_dn must be 'cn_only' or 'org'")
        return val

    @field_validator("easyrsa_req_country")
    @classmethod
    def _check_country(cls, val):
        if not re.match(r"^[A-Z]{2}$", val):
            raise ValueError("Country must be a 2-letter uppercase code")
        return val

    @field_validator("easyrsa_req_province", "easyrsa_req_city", "easyrsa_req_org", "easyrsa_req_ou")
    @classmethod
    def _check_dn_text(cls, val):
        if not re.match(r"^[A-Za-z0-9 .,_-]{0,64}$", val):
            raise ValueError("Only letters, digits, space and . , _ - are allowed (max 64)")
        return val

    @field_validator("easyrsa_req_email")
    @classmethod
    def _check_email(cls, val):
        if not re.match(r"^[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9.-]{1,190}$", val):
            raise ValueError("Invalid email")
        return val

    @field_validator("easyrsa_key_size")
    @classmethod
    def _check_key_size(cls, val):
        if val not in (2048, 3072, 4096):
            raise ValueError("Key size must be 2048, 3072 or 4096")
        return val

    @field_validator("easyrsa_ca_expire", "easyrsa_cert_expire", "easyrsa_cert_renew", "easyrsa_crl_days")
    @classmethod
    def _check_days(cls, val):
        if not 1 <= val <= 36500:
            raise ValueError("Days must be between 1 and 36500")
        return val

class PKISetting(PKISettingBase):
    id: int
    class Config:
        from_attributes = True

# --- System Settings Schemas ---
class SystemSettingsBase(BaseModel):
    protocol: Literal['tcp', 'udp'] = "udp"
    port: int = 1194
    vpn_network: str = "172.20.1.0"
    vpn_netmask: str = "255.255.255.0"
    tunnel_type: Literal['FULL', 'SPLIT'] = "FULL"
    split_routes: List[str] = Field(default_factory=list)
    duplicate_cn: bool = False
    crl_verify: bool = False
    client_to_client: bool = False
    user_defined_dns: bool = False
    dns_servers: List[str] = Field(default_factory=list)
    user_defined_cdscripts: bool = False
    connect_script: str = ""
    disconnect_script: str = ""
    management_interface: bool = False
    management_interface_address: str = "127.0.0.1"
    management_interface_address: str = "127.0.0.1"
    management_port: int = 7505
    public_ip: Optional[str] = None
    tun_mtu: Optional[int] = None
    mssfix: Optional[int] = None

class SystemSettingsUpdate(SystemSettingsBase):
    @field_validator("port", "management_port")
    @classmethod
    def _check_port(cls, val):
        if not 1 <= val <= 65535:
            raise ValueError("Port must be between 1 and 65535")
        return val

    @field_validator("vpn_network")
    @classmethod
    def _check_network(cls, val):
        try:
            ipaddress.IPv4Address(val)
        except ValueError:
            raise ValueError("Invalid IPv4 network address")
        return val

    @field_validator("vpn_netmask")
    @classmethod
    def _check_netmask(cls, val):
        if not v.valid_netmask(val):
            raise ValueError("Invalid netmask")
        return val

    @model_validator(mode="after")
    def _check_subnet(self):
        try:
            net = ipaddress.IPv4Network(f"{self.vpn_network}/{self.vpn_netmask}", strict=True)
        except ValueError:
            raise ValueError("vpn_network is not a valid network address for vpn_netmask")
        if not 8 <= net.prefixlen <= 30:
            raise ValueError("VPN subnet prefix must be between /8 and /30")
        return self

    @field_validator("split_routes")
    @classmethod
    def _check_routes(cls, val):
        if len(val) > 256:
            raise ValueError("Too many routes (max 256)")
        for r in val:
            if not v.valid_route(r):
                raise ValueError(f"Invalid route: {r[:40]!r} (use a.b.c.d/nn or 'a.b.c.d mask')")
        return val

    @field_validator("dns_servers")
    @classmethod
    def _check_dns(cls, val):
        if len(val) > 8:
            raise ValueError("Too many DNS servers (max 8)")
        for d in val:
            try:
                ipaddress.ip_address(d)
            except ValueError:
                raise ValueError(f"Invalid DNS server address: {d[:40]!r}")
        return val

    @field_validator("connect_script", "disconnect_script")
    @classmethod
    def _check_script_path(cls, val):
        if val and not v.SCRIPT_RE.match(val):
            raise ValueError("Script path must be " + v.SCRIPTS_DIR + "/<name> (letters, digits, . _ -)")
        return val

    @field_validator("management_interface_address")
    @classmethod
    def _check_mgmt_addr(cls, val):
        try:
            if not ipaddress.ip_address(val).is_loopback:
                raise ValueError
        except ValueError:
            raise ValueError("Management interface must listen on a loopback address")
        return val

    @field_validator("public_ip")
    @classmethod
    def _check_public_ip(cls, val):
        if val in (None, ""):
            return val
        if not v.valid_host(val):
            raise ValueError("public_ip must be an IP address or a hostname")
        return val

    @field_validator("tun_mtu")
    @classmethod
    def _check_mtu(cls, val):
        if val is not None and not 576 <= val <= 9000:
            raise ValueError("tun_mtu must be between 576 and 9000")
        return val

    @field_validator("mssfix")
    @classmethod
    def _check_mss(cls, val):
        if val is not None and not 536 <= val <= 1500:
            raise ValueError("mssfix must be between 536 and 1500")
        return val

class SystemSettings(SystemSettingsBase):
    id: int
    class Config:
        from_attributes = True

class ConfigResponse(BaseModel):
    server: Optional[SystemSettings] = None
    pki: Optional[PKISetting] = None

# --- User Profile Schemas ---
class UserProfileBase(BaseModel):
    username: str = Field(..., pattern=r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")

class UserProfileCreate(UserProfileBase):
    pass

class UserProfile(UserProfileBase):
    id: int
    status: str
    created_at: datetime
    revoked_at: Optional[datetime] = None
    expiration_date: Optional[datetime] = None
    days_remaining: Optional[int] = None
    is_revoked: bool = False
    is_expired: bool = False
    file_path: Optional[str] = None
    
    class Config:
        from_attributes = True