1import json
  2from enum import Enum
  3from typing import Optional, List
  4
  5from pydantic import ConfigDict, BaseModel, Field, field_validator
  6
  7
  8class XTLSFlows(Enum):
  9    NONE = ""
 10    VISION = "xtls-rprx-vision"
 11
 12
 13class ShadowsocksMethods(Enum):
 14    AES_128_GCM = "aes-128-gcm"
 15    AES_256_GCM = "aes-256-gcm"
 16    CHACHA20_POLY1305 = "chacha20-ietf-poly1305"
 17
 18
 19class ProxyTypes(str, Enum):
 20    # proxy_type = protocol
 21
 22    VMess = "vmess"
 23    VLESS = "vless"
 24    Trojan = "trojan"
 25    Shadowsocks = "shadowsocks"
 26
 27    @property
 28    def settings_model(self):
 29        if self == self.VMess:
 30            return VMessSettings
 31        if self == self.VLESS:
 32            return VLESSSettings
 33        if self == self.Trojan:
 34            return TrojanSettings
 35        if self == self.Shadowsocks:
 36            return ShadowsocksSettings
 37
 38
 39class ProxySettings(BaseModel):
 40    @classmethod
 41    def from_dict(cls, proxy_type: ProxyTypes, _dict: dict):
 42        return ProxyTypes(proxy_type).settings_model.parse_obj(_dict)
 43
 44    def dict(self, *, no_obj=False, **kwargs):
 45        if no_obj:
 46            return json.loads(self.json())
 47        return super().dict(**kwargs)
 48
 49
 50class VMessSettings(ProxySettings):
 51    id: str = Field(nullable=False)
 52
 53
 54class VLESSSettings(ProxySettings):
 55    id: str = Field(nullable=False)
 56    flow: XTLSFlows = XTLSFlows.NONE
 57
 58
 59class TrojanSettings(ProxySettings):
 60    password: str = Field(nullable=False)
 61
 62
 63class ShadowsocksSettings(ProxySettings):
 64    password: str = Field(nullable=False)
 65    method: ShadowsocksMethods = ShadowsocksMethods.CHACHA20_POLY1305
 66
 67
 68class InboundHostSecurity(str, Enum):
 69    inbound_default = "inbound_default"
 70    none = "none"
 71    tls = "tls"
 72
 73
 74InboundHostALPN = Enum(
 75    "ProxyHostALPN",
 76    {
 77        "none": "",
 78        "h2": "h2",
 79        "http/1.1": "http/1.1",
 80        "h2,http/1.1": "h2,http/1.1",
 81    },
 82)
 83
 84
 85InboundHostFingerprint = Enum(
 86    "ProxyHostFingerprint",
 87    {
 88        "none": "",
 89        "chrome": "chrome",
 90        "firefox": "firefox",
 91        "safari": "safari",
 92        "ios": "ios",
 93        "android": "android",
 94        "edge": "edge",
 95        "360": "360",
 96        "qq": "qq",
 97        "random": "random",
 98        "randomized": "randomized",
 99    },
100)
101
102
103class FormatVariables(dict):
104    def __missing__(self, key):
105        return key.join("{}")
106
107
108class FragmentSettings(BaseModel):
109    packets: str = Field(pattern=r"^(:?tlshello|[\d-]{1,32})$")
110    length: str = Field(pattern=r"^[\d-]{1,32}$")
111    interval: str = Field(pattern=r"^[\d-]{1,32}$")
112
113
114class InboundHost(BaseModel):
115    remark: str
116    address: str
117    port: Optional[int] = Field(None)
118    sni: Optional[str] = Field(None)
119    host: Optional[str] = Field(None)
120    path: Optional[str] = Field(None)
121    security: InboundHostSecurity = InboundHostSecurity.inbound_default
122    alpn: InboundHostALPN = InboundHostALPN.none
123    fingerprint: InboundHostFingerprint = InboundHostFingerprint.none
124    allowinsecure: Optional[bool] = None
125    is_disabled: Optional[bool] = None
126    mux: bool = Field(False)
127    fragment: Optional[FragmentSettings] = Field(None)
128    model_config = ConfigDict(from_attributes=True)
129
130    @field_validator("remark", "address", "path")
131    @classmethod
132    def validate_fmt_variables(cls, v: str) -> str:
133        if not v:
134            return v
135
136        v.format_map(FormatVariables())
137
138        return v
139
140
141class InboundHostResponse(InboundHost):
142    id: int
143
144
145class InboundBase(BaseModel):
146    id: Optional[int] = Field(None)
147    tag: str = Field()
148    protocol: ProxyTypes = Field()
149    config: str
150    node_id: Optional[int] = Field(None)
151    model_config = ConfigDict(from_attributes=True)
152
153
154from app.models.service import ServiceBase
155from app.models.node import NodeBase
156
157
158class Inbound(InboundBase):
159    services: Optional[List[ServiceBase]] = None
160    node: Optional[NodeBase] = None