gitrusprusandCursor
5811ecb134
Persist settings in CA, expose /admin/vpn and /internal/vpn, and route Telegram (and optional web/nlp) traffic through mihomo SOCKS when enabled. Co-authored-by: Cursor <cursoragent@cursor.com>
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""VPN / proxy settings shared by CA admin UI and CP workers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from contracts.queues import known_source_types
|
|
|
|
VpnMode = Literal["subscription", "socks5", "http"]
|
|
|
|
VPN_MODES: tuple[str, ...] = ("subscription", "socks5", "http")
|
|
|
|
|
|
class VpnSettings(BaseModel):
|
|
enabled: bool = False
|
|
mode: VpnMode = "subscription"
|
|
subscription_url: str | None = None
|
|
subscription_interval_seconds: int = Field(default=3600, ge=60, le=86400)
|
|
host: str | None = None
|
|
port: int | None = Field(default=None, ge=1, le=65535)
|
|
username: str | None = None
|
|
password: str | None = None
|
|
proxied_source_types: list[str] = Field(default_factory=list)
|
|
|
|
@field_validator("subscription_url", "host", "username", "password", mode="before")
|
|
@classmethod
|
|
def empty_to_none(cls, value: object) -> object:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
return stripped or None
|
|
return value
|
|
|
|
@field_validator("proxied_source_types")
|
|
@classmethod
|
|
def validate_source_types(cls, value: list[str]) -> list[str]:
|
|
known = set(known_source_types())
|
|
cleaned: list[str] = []
|
|
seen: set[str] = set()
|
|
for raw in value or []:
|
|
item = str(raw).strip()
|
|
if not item or item in seen:
|
|
continue
|
|
if item not in known:
|
|
raise ValueError(
|
|
f"Unknown source_type {item!r}. "
|
|
f"Allowed: {', '.join(sorted(known))}"
|
|
)
|
|
seen.add(item)
|
|
cleaned.append(item)
|
|
return cleaned
|
|
|
|
@model_validator(mode="after")
|
|
def require_fields_when_enabled(self) -> "VpnSettings":
|
|
if not self.enabled:
|
|
return self
|
|
if self.mode == "subscription":
|
|
if not self.subscription_url:
|
|
raise ValueError("subscription_url required when mode=subscription")
|
|
elif self.mode in ("socks5", "http"):
|
|
if not self.host:
|
|
raise ValueError(f"host required when mode={self.mode}")
|
|
if self.port is None:
|
|
raise ValueError(f"port required when mode={self.mode}")
|
|
return self
|
|
|
|
def proxies_source(self, source_type: str) -> bool:
|
|
return self.enabled and source_type in self.proxied_source_types
|