test: добавление файла для проверки работоспособности ShadowsocksOutbound

src/xray_manager/core/outbound.py

@@ -21,7 +21,7 @@ class OutboundSpec:
     target: str
 
 
-O = TypeVar("O", bound=Outbound)
+O = TypeVar("O", bound="Outbound")
 T = TypeVar("T")
 
 
@@ -50,24 +50,36 @@ class OutboundFactory:
         return list(cls._registry)
 
     @classmethod
-    def from_link(cls, link: str) -> Outbound:
+    def from_link(
+        cls, xray_manager_config: XrayManagerConfig, link: str
+    ) -> Outbound:
         scheme = link.split("://", 1)[0]
 
         if scheme not in cls._scheme_registry:
             raise ValueError(f"Unsupported link scheme: {scheme}")
 
         outbound_cls = cls._scheme_registry[scheme]
-        return outbound_cls.from_link(link)
+        targets = xray_manager_config.targets
+        return outbound_cls.from_link(targets, link)
 
     @classmethod
-    def from_inbound(cls, host: str, client: dict, inbound: dict):
+    def from_inbound(
+        cls,
+        xray_manager_config: XrayManagerConfig,
+        host: str,
+        client: dict,
+        inbound: dict,
+    ):
         protocol = inbound.get("protocol")
 
         if protocol not in cls._registry:
             raise ValueError(f"Unsupported protocol: {protocol}")
 
         outbound_cls = cls._registry[protocol]
-        username, outbound = outbound_cls.from_inbound(host, client, inbound)
+        targets = xray_manager_config.targets
+        username, outbound = outbound_cls.from_inbound(
+            targets, host, client, inbound
+        )
         return username, outbound
 
     @classmethod
@@ -85,7 +97,15 @@ class OutboundFactory:
 
         inbound = xray_config.find_managed_inbound_by_protocol(spec.protocol)
         outbound_cls = cls._registry[spec.protocol]
-        return outbound_cls.from_scrath(xray_config.host, inbound, spec.target)
+
+        targets = xray_manager_config.targets
+        target_pretty_name = targets.get(spec.target, {}).get(
+            "pretty_name", spec.target
+        )
+
+        return outbound_cls.from_scratch(
+            spec.target, target_pretty_name, xray_config.host, inbound
+        )
 
 
 class Outbound(ABC):
@@ -94,17 +114,25 @@ class Outbound(ABC):
 
     @classmethod
     @abstractmethod
-    def from_link(cls, link: str) -> Outbound: ...
+    def from_link(
+        cls, targets: dict[str, dict[str, str]], link: str
+    ) -> Outbound: ...
 
     @classmethod
     @abstractmethod
     def from_inbound(
-        cls, host: str, client: dict, inbound: dict
+        cls,
+        targets: dict[str, dict[str, str]],
+        host: str,
+        client: dict,
+        inbound: dict,
     ) -> tuple[str, Outbound]: ...
 
     @classmethod
     @abstractmethod
-    def from_scrath(cls, host: str, inbound: dict, target: str) -> Outbound: ...
+    def from_scratch(
+        cls, target: str, target_pretty_name: str, host: str, inbound: dict
+    ) -> Outbound: ...
 
     @abstractmethod
     def to_link(self) -> str: ...
@@ -127,6 +155,16 @@ class Outbound(ABC):
         username, target = email.rsplit("-", 1)
         return username, target
 
+    @staticmethod
+    def find_target_by_pretty_name(
+        targets: dict[str, dict[str, str]], pretty_name: str
+    ) -> str | None:
+        search_name = pretty_name.strip()
+        for target, target_data in targets.items():
+            if target_data.get("pretty_name", "").strip() == search_name:
+                return target
+        return None
+
     @abstractmethod
     def __hash__(self) -> int: ...
 
@@ -140,17 +178,18 @@ _ShadowsocksFields = make_dataclass(
 
 
 @OutboundFactory.register
-@dataclass
 class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
     PROTOCOL = "shadowsocks"
     LINK_SCHEME = "ss"
 
     @classmethod
-    def from_link(cls, link: str) -> ShadowsocksOutbound:
+    def from_link(
+        cls, targets: dict[str, dict[str, str]], link: str
+    ) -> ShadowsocksOutbound:
         import base64
 
         prefix_b64, rest = link[5:].split("@", 1)
-        host_port, target_pretty_name = rest.split("#", 1)
+        host_port, quoted_target_pretty_name = rest.split("#", 1)
         host, port_str = host_port.split(":")
         port = int(port_str)
 
@@ -158,13 +197,182 @@ class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
         prefix_str = prefix_bytes.decode()
         method, server_password, client_password = prefix_str.split(":")
 
+        target_pretty_name = unquote(quoted_target_pretty_name)
+        target = cls.find_target_by_pretty_name(targets, target_pretty_name)
+        if target is None:
+            target = "unknown"
+
         return cls(
             host=host,
             port=port,
+            target=target,
+            target_pretty_name=target_pretty_name,
             method=method,
             server_password=server_password,
             client_password=client_password,
-            exit_point=exit_point,
+        )
+
+    @classmethod
+    def from_inbound(
+        cls,
+        targets: dict[str, dict[str, str]],
+        host: str,
+        client: dict,
+        inbound: dict,
+    ) -> tuple[str, ShadowsocksOutbound]:
+        port = inbound["port"]
+        method = inbound["settings"]["method"]
+        server_password = inbound["settings"]["password"]
+        client_password = client["password"]
+        email = client["email"]
+        username, target = cls.split_client_email(email)
+
+        target_pretty_name = targets.get(target, {}).get("pretty_name", target)
+
+        outbound = cls(
+            host=host,
+            port=port,
+            target=target,
+            target_pretty_name=target_pretty_name,
+            method=method,
+            server_password=server_password,
+            client_password=client_password,
+        )
+        return username, outbound
+
+    @classmethod
+    def from_scratch(
+        cls, target: str, target_pretty_name: str, host: str, inbound: dict
+    ) -> ShadowsocksOutbound:
+        port = inbound["port"]
+        method = inbound["settings"]["method"]
+        server_password = inbound["settings"]["password"]
+        client_password = cls.generate_password(method)
+
+        return cls(
+            host=host,
+            port=port,
+            target=target,
+            target_pretty_name=target_pretty_name,
+            method=method,
+            server_password=server_password,
+            client_password=client_password,
+        )
+
+    def to_link(self) -> str:
+        import base64
+
+        prefix = f"{self.method}:{self.server_password}:{self.client_password}"
+        prefix_b64 = base64.urlsafe_b64encode(prefix.encode()).decode()
+        tag = quote(self.target_pretty_name)
+        link = f"ss://{prefix_b64}@{self.host}:{self.port}#{tag}"
+        return link
+
+    @staticmethod
+    def generate_password(method: str) -> str:
+        import base64
+        import secrets
+
+        key_lengths = {
+            "2022-blake3-aes-128-gcm": 16,
+            "2022-blake3-aes-256-gcm": 32,
+            "2022-blake3-chacha20-poly1305": 32,
+        }
+
+        try:
+            length = key_lengths[method]
+        except KeyError:
+            raise ValueError(f"Unsupported shadowsocks method: {method}")
+
+        key = secrets.token_bytes(length)
+
+        return base64.b64encode(key).decode()
+
+    def matches_inbound(self, inbound: dict):
+        if inbound.get("protocol") != self.PROTOCOL:
+            return False
+
+        if inbound.get("port") != self.port:
+            return False
+
+        settings = inbound.get("settings", {})
+
+        if settings.get("method") != self.method:
+            return False
+
+        if settings.get("password") != self.server_password:
+            return False
+
+        return True
+
+    def add_to_inbound(self, inbound: dict, username: str) -> None:
+        client = {
+            "email": f"{username}-{self.target}",
+            "password": self.client_password,
+        }
+
+        clients = inbound["settings"].setdefault("clients", [])
+        clients.append(client)
+
+        clients_by_user = defaultdict(list)
+        for client in clients:
+            username, _ = self.split_client_email(client["email"])
+            clients_by_user[username].append(client)
+
+        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"])
+
+        for user_clients in clients_by_user.values():
+            user_clients.sort(key=sort_key)
+
+        sorted_clients = []
+        for username in sorted(clients_by_user):
+            sorted_clients.extend(clients_by_user[username])
+
+        inbound["settings"]["clients"] = sorted_clients
+
+    def delete_from_inbound(self, inbound: dict, username: str) -> None:
+        target_email = f"{username}-{self.target}"
+        settings = inbound["settings"]
+        clients = settings["clients"]
+
+        filtered_clients = []
+        for client in clients:
+            if client["email"] != target_email:
+                filtered_clients.append(client)
+
+        inbound["settings"]["clients"] = filtered_clients
+
+    @property
+    def protocol(self) -> str:
+        return self.PROTOCOL
+
+    def __hash__(self) -> int:
+        return hash(
+            (
+                self.host,
+                self.port,
+                self.target,
+                self.method,
+                self.server_password,
+                self.client_password,
+            )
+        )
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, ShadowsocksOutbound):
+            return NotImplemented
+
+        return (
+            self.host == other.host
+            and self.port == other.port
+            and self.target == other.target
+            and self.method == other.method
+            and self.server_password == other.server_password
+            and self.client_password == other.client_password
         )
 
 
@@ -172,6 +380,6 @@ _VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
 
 
 @OutboundFactory.register
-@dataclass
 class VlessOutbound(_VlessFields, Outbound):
-    pass
+    PROTOCOL = "vless"
+    LINK_SCHEME = "vless"

src/xray_manager/core/outbound_fields.py

@@ -1,20 +1,22 @@
 from dataclasses import field
 
-
 _BASE_FIELDS = [
     ("host", str),
     ("port", str),
     ("target", str),
+    ("target_pretty_name", str),
 ]
 
 
 OUTBOUND_FIELDS = {
-    "shadowsocks": _BASE_FIELDS + [
+    "shadowsocks": _BASE_FIELDS
+    + [
         ("method", str),
         ("server_password", str),
         ("client_password", str),
     ],
-    "vless": _BASE_FIELDS + [
+    "vless": _BASE_FIELDS
+    + [
         ("id", str),
         ("security", str),
         ("encryption", str),
@@ -24,6 +26,6 @@ OUTBOUND_FIELDS = {
         ("short_id", str),
         ("flow", str, field(default="xtls-rprx-vision")),
         ("header_type", str, field(default="none")),
-        ("finger_print", str, field(default="chrome"))
+        ("finger_print", str, field(default="chrome")),
     ],
 }

tests/shadowsocks_outbound_sandbox.py

@@ -0,0 +1,47 @@
+from xray_manager.core.outbound import (
+    OutboundFactory,
+    OutboundSpec,
+    ShadowsocksOutbound,
+)
+from xray_manager.core.xray_config import XrayConfig
+from xray_manager.core.xray_manager_config import XrayManagerConfig
+
+inbound = {
+    "tag": "shadowsocks",
+    "xrm": True,
+    "listen": "0.0.0.0",
+    "port": 8443,
+    "protocol": "shadowsocks",
+    "settings": {
+        "method": "2022-blake3-aes-256-gcm",
+        "password": "Z8WtMeWzZVh1F6/5URGrs8vWdB3CLN5y7A9D1U0Q65E=",
+        "clients": [
+            {
+                "email": "test-fr1",
+                "password": "zV3qwuj5TUng+mBMfqdInp0ih7i9ykSIzUig3v5d1Bg=",
+            },
+            {
+                "email": "test2-us1",
+                "password": "ZNtMDJ5WuaGa2ko2jMbbZguddcMY2TTirsb9cNYHffw=",
+            },
+        ],
+        "network": "tcp,udp",
+    },
+}
+
+if __name__ == "__main__":
+    DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
+    xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
+
+    target = "us1"
+    target_pretty_name = xray_manager_config.targets.get(target, {}).get(
+        "pretty_name", target
+    )
+
+    outbound = ShadowsocksOutbound.from_scratch(
+        target, target_pretty_name, "127.0.0.1", inbound
+    )
+
+    link = outbound.to_link()
+    o = ShadowsocksOutbound.from_link(xray_manager_config.targets, link)
+    print(o)