vpn/app/models.py

from __future__ import annotations

import calendar
import secrets
import uuid
from datetime import UTC, date, datetime
from enum import StrEnum
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, field_validator


def utc_now() -> datetime:
    return datetime.now(UTC)


class EngineState(StrEnum):
    STOPPED = "stopped"
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    ERROR = "error"
    UNSUPPORTED = "unsupported"


class EngineKind(StrEnum):
    SIGNBOX = "signbox"
    FREETURN = "freeturn"


class RunMode(StrEnum):
    VPN = "vpn"
    PROXY = "proxy"
    BOTH = "both"


class ProxyAccess(StrEnum):
    LOCAL = "local"
    HOTSPOT = "hotspot"


class ProxyProtocol(StrEnum):
    SOCKS5 = "socks5"
    HTTP = "http"


class AppPolicy(StrEnum):
    PROXY = "proxy"
    DIRECT = "direct"
    VOID = "void"


class Strategy(StrEnum):
    BASIC = "basic"
    ENHANCED = "enhanced"


class CapabilitySet(BaseModel):
    supports_vpn: bool = False
    supports_proxy: bool = False
    supports_socks: bool = False
    supports_http: bool = False
    supports_udp: bool = False
    supports_mobile: bool = False
    supports_desktop: bool = False
    supports_stats: bool = False
    supports_hot_reload: bool = False
    supports_both: bool = False


class ProfileMetadata(BaseModel):
    model_config = ConfigDict(extra="forbid")

    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str = "Новый профиль"
    created_at: datetime = Field(default_factory=utc_now)
    updated_at: datetime = Field(default_factory=utc_now)


class SignBoxProfile(ProfileMetadata):
    kind: EngineKind = EngineKind.SIGNBOX
    strategy: Strategy = Strategy.BASIC
    config: dict[str, Any] = Field(default_factory=dict)
    mtu: int = Field(default=1280, ge=576, le=9000)
    split_tunneling: bool = False
    raw_config: str | None = None


class FreeTurnProfile(ProfileMetadata):
    kind: EngineKind = EngineKind.FREETURN
    key: str = ""
    options: dict[str, Any] = Field(default_factory=dict)
    preset: int = Field(default=1, ge=1, le=4)
    stability_over_load: bool = False
    captcha_always_browser: bool = False

    @property
    def stream_settings(self) -> tuple[int, int]:
        values = {1: (4, 4), 2: (8, 4), 3: (12, 6), 4: (24, 6)}
        threads, per_credential = values[self.preset]
        if self.stability_over_load and self.preset >= 3:
            per_credential = 4
        return threads, per_credential


class ProxySettings(BaseModel):
    enabled: bool = True
    access: ProxyAccess = ProxyAccess.LOCAL
    socks_enabled: bool = True
    socks_port: int = Field(default=1080, ge=1, le=65535)
    socks_random_port: bool = False
    http_enabled: bool = True
    http_port: int = Field(default=8080, ge=1, le=65535)
    http_random_port: bool = False
    authentication: bool = False
    username: str = ""
    password: str = ""
    bind_address: str = "127.0.0.1"

    @field_validator("username", "password")
    @classmethod
    def no_control_chars(cls, value: str) -> str:
        if any(ord(char) < 32 for char in value):
            raise ValueError("credentials must not contain control characters")
        return value

    def validate_for_access(self) -> None:
        if self.access is ProxyAccess.HOTSPOT and not self.authentication:
            return
        if self.authentication and (not self.username or not self.password):
            raise ValueError("proxy authentication requires username and password")


class VpnSettings(BaseModel):
    mode: RunMode = RunMode.VPN
    tun_address: str = "10.0.0.1"
    tun_prefix: int = Field(default=24, ge=8, le=30)
    mtu: int = Field(default=1280, ge=576, le=9000)
    split_tunneling: bool = False
    app_rules: dict[str, AppPolicy] = Field(default_factory=dict)

    @field_validator("app_rules")
    @classmethod
    def validate_rules(cls, value: dict[str, AppPolicy]) -> dict[str, AppPolicy]:
        # A VOID rule must never be represented as a VpnService exclusion.
        return dict(value)


class RuntimeRequest(BaseModel):
    engine: EngineKind
    mode: RunMode
    profile_id: str
    proxy: ProxySettings = Field(default_factory=ProxySettings)
    vpn: VpnSettings = Field(default_factory=VpnSettings)
    engine_config: dict[str, Any] = Field(default_factory=dict)
    config_revision: int = Field(default=0, ge=0)


class RuntimeStats(BaseModel):
    upload_bytes: int = 0
    download_bytes: int = 0
    upload_bps: int = 0
    download_bps: int = 0
    uptime_seconds: int = 0


class RuntimeSnapshot(BaseModel):
    state: EngineState = EngineState.STOPPED
    engine: EngineKind | None = None
    mode: RunMode | None = None
    profile_id: str | None = None
    message: str | None = None
    stats: RuntimeStats = Field(default_factory=RuntimeStats)
    proxy_addresses: list[str] = Field(default_factory=list)
    captcha_url: str | None = None
    revision: int = 0


class PaymentInfo(BaseModel):
    monthly_price_rub: int = Field(ge=0)
    payment_day: int = Field(ge=1, le=31)

    def due_date(self, year: int, month: int) -> date:
        last_day = calendar.monthrange(year, month)[1]
        return date(year, month, min(self.payment_day, last_day))


class DeviceSubscription(BaseModel):
    device_id: str
    device_token: str
    subscription_url: str


def generate_device_token() -> str:
    return secrets.token_urlsafe(32)


def generate_device_id() -> str:
    return str(uuid.uuid4())