feat: завершение работы над классом VlessOutbound

src/xray_manager/core/outbound.py

@@ -8,6 +8,7 @@ from dataclasses import dataclass, make_dataclass
 from typing import TYPE_CHECKING, TypeVar
 from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
 
+from ..utils import private_to_public
 from .outbound_fields import OUTBOUND_FIELDS
 
 if TYPE_CHECKING:
@@ -282,7 +283,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
 
     def add_to_inbound(self, inbound: dict, username: str) -> None:
         client = {
-            "email": f"{username}-{self.target}",
+            "email": f"{username}-{self.target.id}",
             "password": self.client_password,
         }
 
@@ -297,8 +298,8 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         priority = {"default": 0, "managed": 1, "relay": 2}
 
         def sort_key(c):
-            _, target = self.split_client_email(c["email"])
-            return (priority.get(target, 100), c["email"])
+            _, target_id = self.split_client_email(c["email"])
+            return (priority.get(target_id, 100), c["email"])
 
         for user_clients in clients_by_user.values():
             user_clients.sort(key=sort_key)
@@ -310,7 +311,7 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         inbound["settings"]["clients"] = sorted_clients
 
     def delete_from_inbound(self, inbound: dict, username: str) -> None:
-        target_email = f"{username}-{self.target}"
+        target_email = f"{username}-{self.target.id}"
         settings = inbound["settings"]
         clients = settings["clients"]
 
@@ -358,3 +359,292 @@ _VlessFields = make_dataclass("_VlessFields", fields=OUTBOUND_FIELDS["vless"])
 class VlessOutbound(_VlessFields, Outbound):
     PROTOCOL = "vless"
     LINK_SCHEME = "vless"
+
+    @classmethod
+    def from_link(
+        cls, target_storage: TargetStorage, link: str
+    ) -> VlessOutbound:
+        parsed = urlparse(link)
+        query = parse_qs(parsed.query)
+
+        def require(value: T | None, name: str) -> T:
+            if value is None:
+                raise ValueError(f"Missing required field: {name}")
+            return value
+
+        def require_str(value: str | None, name: str) -> str:
+            value = require(value, name)
+            if value == "":
+                raise ValueError(f"Empty value for required field: {name}")
+            return unquote(value)
+
+        def get_required(key: str) -> str:
+            values = query.get(key)
+            if not values:
+                raise ValueError(f"Missing required query param: {key}")
+            raw = values[0]
+            if raw is None or raw == "":
+                raise ValueError(f"Empty value for required query param: {key}")
+            return unquote(raw)
+
+        return cls(
+            host=require_str(parsed.hostname, "host"),
+            port=require(parsed.port, "port"),
+            target=target_storage.load_by_pretty(
+                require_str(parsed.fragment, "target_pretty_name")
+            ),
+            id=require_str(parsed.username, "id"),
+            security=get_required("security"),
+            encryption=get_required("encryption"),
+            public_key=get_required("pbk"),
+            network_type=get_required("type"),
+            sni=get_required("sni"),
+            short_id=get_required("sid"),
+            flow=get_required("flow"),
+            header_type=get_required("headerType"),
+            finger_print=get_required("fp"),
+        )
+
+    @classmethod
+    def from_inbound(
+        cls,
+        target_storage: TargetStorage,
+        host: str,
+        client: dict,
+        inbound: dict,
+    ) -> tuple[str, VlessOutbound]:
+        settings = inbound["settings"]
+        stream = inbound["streamSettings"]
+        reality = stream["realitySettings"]
+
+        port = inbound["port"]
+        id = client["id"]
+        flow = client["flow"]
+        encryption = settings["decryption"]
+        network_type = stream["network"]
+        security = stream["security"]
+        sni = reality["serverNames"][0]
+        public_key = private_to_public(reality["privateKey"])
+
+        clients = settings["clients"]
+        short_ids = reality["shortIds"]
+
+        try:
+            index = clients.index(client)
+        except ValueError:
+            raise RuntimeError(f"Client not found in inbound: {client}")
+
+        try:
+            short_id = short_ids[index]
+        except IndexError:
+            raise RuntimeError(f"No shortId for client at index {index}")
+
+        email = client["email"]
+        username, target_id = cls.split_client_email(email)
+
+        connection = cls(
+            host=host,
+            port=port,
+            target=target_storage.load_by_id(target_id),
+            id=id,
+            security=security,
+            encryption=encryption,
+            public_key=public_key,
+            network_type=network_type,
+            sni=sni,
+            short_id=short_id,
+            flow=flow,
+        )
+        return username, connection
+
+    @classmethod
+    def from_scratch(
+        cls, target: Target, host: str, inbound: dict
+    ) -> VlessOutbound:
+        settings = inbound["settings"]
+        stream = inbound["streamSettings"]
+        reality = stream["realitySettings"]
+
+        port = inbound["port"]
+        encryption = settings["decryption"]
+        network_type = stream["network"]
+        security = stream["security"]
+        sni = reality["serverNames"][0]
+        public_key = private_to_public(reality["privateKey"])
+
+        id = cls.generate_id()
+        short_id = cls.generate_short_id()
+
+        return cls(
+            host=host,
+            port=port,
+            target=target,
+            id=id,
+            security=security,
+            encryption=encryption,
+            public_key=public_key,
+            network_type=network_type,
+            sni=sni,
+            short_id=short_id,
+        )
+
+    def to_link(self) -> str:
+        netloc = f"{self.id}@{self.host}:{self.port}"
+        params = {
+            "security": self.security,
+            "encryption": self.encryption,
+            "pbk": self.public_key,
+            "headerType": self.header_type,
+            "fp": self.finger_print,
+            "type": self.network_type,
+            "flow": self.flow,
+            "sni": self.sni,
+            "sid": self.short_id,
+        }
+        query = urlencode(params, quote_via=quote)
+        fragment = quote(self.target.pretty)
+        return f"vless://{netloc}?{query}#{fragment}"
+
+    @staticmethod
+    def generate_id() -> str:
+        return str(uuid.uuid4())
+
+    @staticmethod
+    def generate_short_id() -> str:
+        return secrets.token_hex(8)
+
+    def matches_inbound(self, inbound: dict) -> bool:
+        if inbound.get("protocol") != self.PROTOCOL:
+            return False
+
+        if inbound.get("port") != self.port:
+            return False
+
+        settings = inbound.get("settings", {})
+        if settings.get("decryption") != self.encryption:
+            return False
+
+        stream_settings = inbound.get("streamSettings", {})
+        if stream_settings.get("network") != self.network_type:
+            return False
+        if stream_settings.get("security") != self.security:
+            return False
+
+        reality_settings = stream_settings.get("realitySettings", {})
+        if self.sni not in reality_settings.get("serverNames", []):
+            return False
+
+        inbound_private_key = reality_settings.get("privateKey")
+        if not inbound_private_key:
+            raise ValueError("Reality Settings do not have private key")
+        inbound_public_key = private_to_public(inbound_private_key)
+        if inbound_public_key != self.public_key:
+            return False
+
+        return True
+
+    def add_to_inbound(self, inbound: dict, username: str) -> None:
+        client = {
+            "email": f"{username}-{self.target.id}",
+            "id": self.id,
+            "flow": self.flow,
+        }
+
+        clients = inbound["settings"].setdefault("clients", [])
+        reality = inbound["streamSettings"].setdefault("realitySettings", {})
+        short_ids = reality.setdefault("shortIds", [])
+
+        clients.append(client)
+        short_ids.append(self.short_id)
+
+        clients_and_sids_by_user = defaultdict(list)
+        for client, sid in zip(clients, short_ids):
+            username, _ = self.split_client_email(client["email"])
+            clients_and_sids_by_user[username].append((client, sid))
+
+        priority = {"default": 0, "managed": 1, "relay": 2}
+
+        def sort_key(pair):
+            client, _ = pair
+            _, target_id = self.split_client_email(client["email"])
+            return (priority.get(target_id, 100), client["email"])
+
+        for pair in clients_and_sids_by_user.values():
+            pair.sort(key=sort_key)
+
+        sorted_clients = []
+        sorted_short_ids = []
+        for username in sorted(clients_and_sids_by_user):
+            for client, sid in clients_and_sids_by_user[username]:
+                sorted_clients.append(client)
+                sorted_short_ids.append(sid)
+
+        inbound["settings"]["clients"] = sorted_clients
+        reality["shortIds"] = sorted_short_ids
+
+    def delete_from_inbound(self, inbound: dict, username: str) -> None:
+        target_email = f"{username}-{self.target.id}"
+
+        clients = inbound["settings"].get("clients", [])
+        reality = inbound["streamSettings"].get("realitySettings", {})
+        short_ids = reality.get("shortIds", [])
+
+        if not clients or not short_ids:
+            return
+
+        filtered_pairs = [
+            (client, sid)
+            for client, sid in zip(clients, short_ids)
+            if client["email"] != target_email
+        ]
+
+        inbound["settings"]["clients"] = [
+            client for client, _ in filtered_pairs
+        ]
+
+        inbound["streamSettings"]["realitySettings"]["shortIds"] = [
+            sid for _, sid in filtered_pairs
+        ]
+
+    @property
+    def protocol(self) -> str:
+        return self.PROTOCOL
+
+    def __hash__(self) -> int:
+        return hash(
+            (
+                self.host,
+                self.port,
+                self.target,
+                self.id,
+                self.security,
+                self.encryption,
+                self.public_key,
+                self.network_type,
+                self.sni,
+                self.short_id,
+                self.flow,
+                self.header_type,
+                self.finger_print,
+            )
+        )
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, VlessOutbound):
+            return NotImplemented
+
+        return (
+            self.host == other.host
+            and self.port == other.port
+            and self.target == other.target
+            and self.id == other.id
+            and self.security == other.security
+            and self.encryption == other.encryption
+            and self.public_key == other.public_key
+            and self.network_type == other.network_type
+            and self.sni == other.sni
+            and self.short_id == other.short_id
+            and self.flow == other.flow
+            and self.header_type == other.header_type
+            and self.finger_print == other.finger_print
+        )