1from enum import Enum
  2
  3from pydantic import ConfigDict, BaseModel, Field, field_validator
  4
  5from app.models.node import Node
  6
  7
  8class ProxyTypes(str, Enum):
  9    # proxy_type = protocol
 10    VMess = "vmess"
 11    VLESS = "vless"
 12    Trojan = "trojan"
 13    Shadowsocks = "shadowsocks"
 14    Shadowsocks2022 = "shadowsocks2022"
 15    Hysteria2 = "hysteria2"
 16    WireGuard = "wireguard"
 17    TUIC = "tuic"
 18    ShadowTLS = "shadowtls"
 19
 20
 21class InboundHostSecurity(str, Enum):
 22    inbound_default = "inbound_default"
 23    none = "none"
 24    tls = "tls"
 25
 26
 27InboundHostALPN = Enum(
 28    "ProxyHostALPN",
 29    {
 30        "none": "none",
 31        "h2": "h2",
 32        "http/1.1": "http/1.1",
 33        "h2,http/1.1": "h2,http/1.1",
 34        "h3": "h3",
 35        "h3,h2": "h3,h2",
 36        "h3,h2,http/1.1": "h3,h2,http/1.1",
 37    },
 38)
 39
 40
 41InboundHostFingerprint = Enum(
 42    "ProxyHostFingerprint",
 43    {
 44        "none": "",
 45        "chrome": "chrome",
 46        "firefox": "firefox",
 47        "safari": "safari",
 48        "ios": "ios",
 49        "android": "android",
 50        "edge": "edge",
 51        "360": "360",
 52        "qq": "qq",
 53        "random": "random",
 54        "randomized": "randomized",
 55    },
 56)
 57
 58
 59class FormatVariables(dict):
 60    def __missing__(self, key):
 61        return key.join("{}")
 62
 63
 64class FragmentSettings(BaseModel):
 65    packets: str = Field(pattern=r"^(:?tlshello|[\d-]{1,32})$")
 66    length: str = Field(pattern=r"^[\d-]{1,32}$")
 67    interval: str = Field(pattern=r"^[\d-]{1,32}$")
 68
 69
 70class XrayNoise(BaseModel):
 71    type: str = Field(pattern=r"^(:?rand|str|base64)$")
 72    packet: str = Field()
 73    delay: str = Field(pattern=r"^\d{1,10}(-\d{1,10})?$")
 74
 75
 76class XMuxSettings(BaseModel):
 77    max_concurrency: str | None = Field(
 78        None, pattern=r"^\d{1,10}(-\d{1,10})?$"
 79    )
 80    max_connections: str | None = Field(
 81        None, pattern=r"^\d{1,10}(-\d{1,10})?$"
 82    )
 83    max_reuse_times: str | None = Field(
 84        None, pattern=r"^\d{1,10}(-\d{1,10})?$"
 85    )
 86    max_lifetime: str | None = Field(None, pattern=r"^\d{1,10}(-\d{1,10})?$")
 87    max_request_times: str | None = Field(None)
 88    keep_alive_period: int | None = Field(None)
 89
 90
 91class SplitHttpSettings(BaseModel):
 92    mode: str | None = None
 93    no_grpc_header: bool | None = None
 94    padding_bytes: str | None = None
 95    xmux: XMuxSettings | None = None
 96
 97
 98class SingBoxMuxSettings(BaseModel):
 99    max_connections: int | None = Field(None)
100    max_streams: int | None = Field(None)
101    min_streams: int | None = Field(None)
102    padding: bool | None = Field(None)
103
104
105class MuxCoolSettings(BaseModel):
106    concurrency: int | None = None
107    xudp_concurrency: int | None = None
108    xudp_proxy_443: str | None = None
109
110
111class MuxSettings(BaseModel):
112    protocol: str
113    sing_box_mux_settings: SingBoxMuxSettings | None = None
114    mux_cool_settings: MuxCoolSettings | None = None
115
116
117class InboundHost(BaseModel):
118    remark: str
119    address: str
120    uuid: str | None = None
121    password: str | None = None
122    protocol: ProxyTypes | str | None = None
123    network: str | None = None
124    port: int | None = Field(None)
125    sni: str | None = Field(None)
126    host: str | None = Field(None)
127    path: str | None = Field(None)
128    security: InboundHostSecurity = InboundHostSecurity.inbound_default
129    alpn: InboundHostALPN = InboundHostALPN.none
130    fingerprint: InboundHostFingerprint = InboundHostFingerprint.none
131    allowinsecure: bool | None = False
132    is_disabled: bool | None = False
133    fragment: FragmentSettings | None = Field(None)
134    noise: list[XrayNoise] | None = None
135    http_headers: dict[str, str] | None = Field(default_factory=dict)
136    mtu: int | None = None
137    dns_servers: str | None = None
138    allowed_ips: str | None = None
139    header_type: str | None = None
140    reality_public_key: str | None = None
141    reality_short_ids: list[str] | None = None
142    flow: str | None = None
143    shadowtls_version: int | None = None
144    shadowsocks_method: str | None = None
145    splithttp_settings: SplitHttpSettings | None = None
146    early_data: int | None = None
147    mux_settings: MuxSettings | None = None
148    universal: bool = False
149    service_ids: list[int] = Field(default_factory=list)
150    weight: int = 1
151    inbound_id: int | None = None
152    chain_ids: list[int] = Field(default_factory=list)
153
154    model_config = ConfigDict(from_attributes=True)
155
156    @field_validator("remark", "address", "path")
157    @classmethod
158    def validate_fmt_variables(cls, v: str) -> str:
159        if not v:
160            return v
161
162        v.format_map(FormatVariables())
163
164        return v
165
166    @field_validator("alpn", mode="before")
167    @classmethod
168    def validate_alpn(cls, v):
169        if not v:
170            return InboundHostALPN.none
171        return v
172
173
174class InboundHostResponse(InboundHost):
175    id: int
176
177
178class Inbound(BaseModel):
179    id: int
180    tag: str
181    protocol: ProxyTypes
182    config: str
183    node: Node
184    service_ids: list[int]
185    model_config = ConfigDict(from_attributes=True)