diff --git a/src/xray_manager/core/outbound.py b/src/xray_manager/core/outbound.py index 74beb35..8ea8ebb 100644 --- a/src/xray_manager/core/outbound.py +++ b/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"