1import json
  2from enum import Enum
  3from typing import Optional, Union, List
  4# from uuid import UUID, uuid4
  5
  6from pydantic import ConfigDict, BaseModel, Field, validator
  7
  8
  9class XTLSFlows(Enum):
 10    NONE = ''
 11    VISION = 'xtls-rprx-vision'
 12
 13
 14class ShadowsocksMethods(Enum):
 15    AES_128_GCM = 'aes-128-gcm'
 16    AES_256_GCM = 'aes-256-gcm'
 17    CHACHA20_POLY1305 = 'chacha20-ietf-poly1305'
 18
 19
 20class ProxyTypes(str, Enum):
 21    # proxy_type = protocol
 22
 23    VMess = "vmess"
 24    VLESS = "vless"
 25    Trojan = "trojan"
 26    Shadowsocks = "shadowsocks"
 27
 28    @property
 29    def settings_model(self):
 30        if self == self.VMess:
 31            return VMessSettings
 32        if self == self.VLESS:
 33            return VLESSSettings
 34        if self == self.Trojan:
 35            return TrojanSettings
 36        if self == self.Shadowsocks:
 37            return ShadowsocksSettings
 38
 39
 40class ProxySettings(BaseModel):
 41    @classmethod
 42    def from_dict(cls, proxy_type: ProxyTypes, _dict: dict):
 43        return ProxyTypes(proxy_type).settings_model.parse_obj(_dict)
 44
 45    def dict(self, *, no_obj=False, **kwargs):
 46        if no_obj:
 47            return json.loads(self.json())
 48        return super().dict(**kwargs)
 49
 50
 51class VMessSettings(ProxySettings):
 52    id: str = Field(nullable=False)
 53
 54
 55class VLESSSettings(ProxySettings):
 56    id: str = Field(nullable=False)
 57    flow: XTLSFlows = XTLSFlows.NONE
 58
 59
 60class TrojanSettings(ProxySettings):
 61    password: str = Field(nullable=False)
 62
 63
 64class ShadowsocksSettings(ProxySettings):
 65    password: str = Field(nullable=False)
 66    method: ShadowsocksMethods = ShadowsocksMethods.CHACHA20_POLY1305
 67
 68
 69class InboundHostSecurity(str, Enum):
 70    inbound_default = "inbound_default"
 71    none = "none"
 72    tls = "tls"
 73
 74
 75InboundHostALPN = Enum(
 76    "ProxyHostALPN",
 77    {
 78        "none": "",
 79        "h2": "h2",
 80        "http/1.1": "http/1.1",
 81        "h2,http/1.1": "h2,http/1.1",
 82    },
 83)
 84
 85
 86InboundHostFingerprint = Enum(
 87    "ProxyHostFingerprint",
 88    {
 89        "none": "",
 90        "chrome": "chrome",
 91        "firefox": "firefox",
 92        "safari": "safari",
 93        "ios": "ios",
 94        "android": "android",
 95        "edge": "edge",
 96        "360": "360",
 97        "qq": "qq",
 98        "random": "random",
 99        "randomized": "randomized",
100    },
101)
102
103
104class FormatVariables(dict):
105    def __missing__(self, key):
106        return key.join("{}")
107
108
109class InboundHost(BaseModel):
110    remark: str
111    address: str
112    port: Optional[int] = Field(None, nullable=True)
113    sni: Optional[str] = Field(None, nullable=True)
114    host: Optional[str] = Field(None, nullable=True)
115    path: Optional[str] = Field(None, nullable=True)
116    security: InboundHostSecurity = InboundHostSecurity.inbound_default
117    alpn: InboundHostALPN = InboundHostALPN.none
118    fingerprint: InboundHostFingerprint = InboundHostFingerprint.none
119    allowinsecure: Union[bool, None] = None
120    is_disabled: Union[bool, None] = None
121    model_config = ConfigDict(from_attributes=True)
122
123    # TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.
124    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
125    @validator("remark", pre=False, always=True)
126    def validate_remark(cls, v):
127        try:
128            v.format_map(FormatVariables())
129        except ValueError as exc:
130            raise ValueError("Invalid formatting variables")
131
132        return v
133
134    # TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.
135    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
136    @validator("address", pre=False, always=True)
137    def validate_address(cls, v):
138        try:
139            v.format_map(FormatVariables())
140        except ValueError as exc:
141            raise ValueError("Invalid formatting variables")
142
143        return v
144
145    # TODO[pydantic]: We couldn't refactor the `validator`, please replace it by `field_validator` manually.
146    # Check https://docs.pydantic.dev/dev-v2/migration/#changes-to-validators for more information.
147    @validator("path", pre=False, always=True)
148    def validate_path(cls, v):
149        if not v:
150            return v
151        try:
152            v.format_map(FormatVariables())
153        except ValueError:
154            raise ValueError("Invalid formatting variables")
155
156        return v
157
158
159class InboundHostResponse(InboundHost):
160    id: int
161
162
163class InboundBase(BaseModel):
164    id: Optional[int] = Field(None)
165    tag: str = Field()
166    protocol: ProxyTypes = Field()
167    config: str
168    node_id: Optional[int] = Field(None)
169    model_config = ConfigDict(from_attributes=True)
170
171
172from app.models.service import ServiceBase
173from app.models.node import NodeBase
174
175
176class Inbound(InboundBase):
177    services: Optional[List[ServiceBase]] = None
178    node: Optional[NodeBase] = None