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