wip

src/xray_manager/core/outbound.py

@@ -10,9 +10,128 @@ from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
 
 from .outbound_fields import OUTBOUND_FIELDS
 
+if TYPE_CHECKING:
+    from .xray_config import XrayConfig
+    from .xray_manager_config import XrayManagerConfig
+
+
+@dataclass(frozen=True)
+class OutboundSpec:
+    protocol: str
+    target: str
+
+
+O = TypeVar("O", bound=Outbound)
+T = TypeVar("T")
+
+
+class OutboundFactory:
+    _registry: dict[str, type[Outbound]] = {}
+    _scheme_registry: dict[str, type[Outbound]] = {}
+
+    @classmethod
+    def register(cls, outbound_cls: type[O]) -> type[O]:
+        protocol = outbound_cls.PROTOCOL
+        scheme = outbound_cls.LINK_SCHEME
+
+        if protocol in cls._registry:
+            raise RuntimeError(f"Protocol already registered: {protocol}")
+
+        if scheme in cls._scheme_registry:
+            raise RuntimeError(f"Scheme already registered: {scheme}")
+
+        cls._registry[protocol] = outbound_cls
+        cls._scheme_registry[scheme] = outbound_cls
+
+        return outbound_cls
+
+    @classmethod
+    def get_protocols(cls) -> list[str]:
+        return list(cls._registry)
+
+    @classmethod
+    def from_link(cls, 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)
+
+    @classmethod
+    def from_inbound(cls, 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)
+        return username, outbound
+
+    @classmethod
+    def from_spec(
+        cls,
+        xray_config: XrayConfig,
+        xray_manager_config: XrayManagerConfig,
+        spec: OutboundSpec,
+    ) -> Outbound:
+        if spec.protocol not in cls._registry:
+            raise ValueError(f"Unsupported protocol: {spec.protocol}")
+
+        if spec.target not in xray_manager_config.get_targets_list():
+            raise ValueError(f"Target {spec.target} not found")
+
+        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)
+
 
 class Outbound(ABC):
-    pass
+    PROTOCOL: str
+    LINK_SCHEME: str
+
+    @classmethod
+    @abstractmethod
+    def from_link(cls, link: str) -> Outbound: ...
+
+    @classmethod
+    @abstractmethod
+    def from_inbound(
+        cls, host: str, client: dict, inbound: dict
+    ) -> tuple[str, Outbound]: ...
+
+    @classmethod
+    @abstractmethod
+    def from_scrath(cls, host: str, inbound: dict, target: str) -> Outbound: ...
+
+    @abstractmethod
+    def to_link(self) -> str: ...
+
+    @abstractmethod
+    def matches_inbound(self, inbound: dict) -> bool: ...
+
+    @abstractmethod
+    def add_to_inbound(self, inbound: dict, username: str) -> None: ...
+
+    @abstractmethod
+    def delete_from_inbound(self, inbound: dict, username: str) -> None: ...
+
+    @property
+    @abstractmethod
+    def protocol(self) -> str: ...
+
+    @staticmethod
+    def split_client_email(email: str) -> tuple[str, str]:
+        username, target = email.rsplit("-", 1)
+        return username, target
+
+    @abstractmethod
+    def __hash__(self) -> int: ...
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool: ...
 
 
 _ShadowsocksFields = make_dataclass(
@@ -20,14 +139,39 @@ _ShadowsocksFields = make_dataclass(
 )
 
 
+@OutboundFactory.register
+@dataclass
 class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
-    pass
+    PROTOCOL = "shadowsocks"
+    LINK_SCHEME = "ss"
 
+    @classmethod
+    def from_link(cls, link: str) -> ShadowsocksOutbound:
+        import base64
 
-_VlessFields = make_dataclass(
+        prefix_b64, rest = link[5:].split("@", 1)
-    "_VlessFieldss", fields=OUTBOUND_FIELDS["vless"]
+        host_port, target_pretty_name = rest.split("#", 1)
+        host, port_str = host_port.split(":")
+        port = int(port_str)
+
+        prefix_bytes = base64.urlsafe_b64decode(prefix_b64)
+        prefix_str = prefix_bytes.decode()
+        method, server_password, client_password = prefix_str.split(":")
+
+        return cls(
+            host=host,
+            port=port,
+            method=method,
+            server_password=server_password,
+            client_password=client_password,
+            exit_point=exit_point,
         )
 
 
+_VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
+
+
+@OutboundFactory.register
+@dataclass
 class VlessOutbound(_VlessFields, Outbound):
     pass