refactor: перенос логики работы с конфигурациями Xray из класса XrayConfig в классы Outbound и OutboundFactory

README.md

@@ -7,10 +7,9 @@ user show (<username> | -a)
 user modify <username> [-a <protocol>:<target> [<protocol>:<target> ...]] [-d <protocol>:<target> [<protocol>:<target> ...]]
 user delete (<username> | -a)
 
-user outbound add <username> -p <protocol> -t <target> [--<param> <value> ...]
-user outbound show <username> (-a | -p <protocol> | -t <target>)
-user outbound modify <username> -p <protocol> -t <target> [--<param> <value> ...]
-user outbound delete <username> (-a | -p <protocol> | -t <target>)
+user outbound set <username> -protocol [--target <target_id>] [--<param> <value> ...]
+user outbound show <username> (-a | -p <protocol> [--target <target_id>])
+user outbound delete <username> (-a | -p <protocol> | --target <target_id>)
 
 user profile build (<username> | -a)
 user profile show (<username> | -a)

src/xray_manager/core/__init__.py

@@ -1,5 +1,13 @@
-from .connection import Connection, ConnectionFactory, ConnectionKey
+from .outbound import (
+    OUTBOUND_FIELDS,
+    Outbound,
+    OutboundFactory,
+    OutboundSpec,
+    ShadowsocksOutbound,
+    VlessOutbound,
+)
 from .profile import Profile, ProfileFactory, ProfileStorage
+from .target import Target, TargetStorage
 from .user import User, UserFactory
 from .xray_config import XrayConfig
 from .xray_manager_config import XrayManagerConfig

src/xray_manager/core/connection.py

@@ -1,625 +0,0 @@
-from __future__ import annotations
-
-import secrets
-import uuid
-from abc import ABC, abstractmethod
-from collections import defaultdict
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, TypeVar
-from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
-
-from ..utils import private_to_public
-
-if TYPE_CHECKING:
-    from .xray_config import XrayConfig
-
-C = TypeVar("C", bound="Connection")
-T = TypeVar("T")
-
-
-class ConnectionFactory:
-    _registry: dict[str, type["Connection"]] = {}
-    _scheme_registry: dict[str, type["Connection"]] = {}
-
-    @classmethod
-    def register(cls, conn_cls: type[C]) -> type[C]:
-        protocol = conn_cls.PROTOCOL
-        scheme = conn_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] = conn_cls
-        cls._scheme_registry[scheme] = conn_cls
-
-        return conn_cls
-
-    @classmethod
-    def get_protocols(cls) -> list[str]:
-        return list(cls._registry)
-
-    @classmethod
-    def from_link(cls, link: str) -> "Connection":
-        scheme = link.split("://", 1)[0]
-
-        if scheme not in cls._scheme_registry:
-            raise ValueError(f"Unsupported link scheme: {scheme}")
-
-        conn_cls = cls._scheme_registry[scheme]
-        return conn_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}")
-
-        conn_cls = cls._registry[protocol]
-        username, connection = conn_cls.from_inbound(host, client, inbound)
-        return username, connection
-
-    @classmethod
-    def from_spec(
-        cls, key: "ConnectionKey", xray_config: "XrayConfig"
-    ) -> "Connection":
-        if key.protocol not in cls._registry:
-            raise ValueError(f"Unsupported protocol: {key.protocol}")
-
-        if key.exit_point not in xray_config.get_exit_points():
-            raise ValueError(f"Exit point {key.exit_point} not found")
-
-        inbound = xray_config.find_managed_inbound_by_protocol(key.protocol)
-        conn_cls = cls._registry[key.protocol]
-        return conn_cls.from_spec(xray_config.host, inbound, key.exit_point)
-
-
-class Connection(ABC):
-    PROTOCOL: str
-    LINK_SCHEME: str
-    exit_point: str
-
-    @classmethod
-    @abstractmethod
-    def from_link(cls, link: str) -> "Connection": ...
-
-    @classmethod
-    @abstractmethod
-    def from_inbound(
-        cls, host: str, client: dict, inbound: dict
-    ) -> tuple[str, "Connection"]: ...
-
-    @classmethod
-    @abstractmethod
-    def from_spec(
-        cls, host: str, inbound: dict, exit_point: str
-    ) -> "Connection": ...
-
-    @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, exit_point = email.rsplit("-", 1)
-        return username, exit_point
-
-    @abstractmethod
-    def __hash__(self) -> int: ...
-
-    @abstractmethod
-    def __eq__(self, other: object) -> bool: ...
-
-
-@dataclass(frozen=True)
-class ConnectionKey:
-    protocol: str
-    exit_point: str
-
-
-@ConnectionFactory.register
-@dataclass(eq=False)
-class ShadowsocksConnection(Connection):
-    PROTOCOL = "shadowsocks"
-    LINK_SCHEME = "ss"
-
-    host: str
-    port: int
-    method: str
-    server_password: str
-    client_password: str
-    exit_point: str
-
-    @classmethod
-    def from_link(cls, link: str) -> "ShadowsocksConnection":
-        import base64
-
-        prefix_b64, rest = link[5:].split("@", 1)
-        host_port, exit_point = 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,
-        )
-
-    @classmethod
-    def from_inbound(
-        cls, host: str, client: dict, inbound: dict
-    ) -> tuple[str, "ShadowsocksConnection"]:
-        port = inbound["port"]
-        method = inbound["settings"]["method"]
-        server_password = inbound["settings"]["password"]
-        client_password = client["password"]
-        email = client["email"]
-        username, exit_point = cls.split_client_email(email)
-        connection = cls(
-            host=host,
-            port=port,
-            method=method,
-            server_password=server_password,
-            client_password=client_password,
-            exit_point=exit_point,
-        )
-        return username, connection
-
-    @classmethod
-    def from_spec(
-        cls, host: str, inbound: dict, exit_point: str
-    ) -> "ShadowsocksConnection":
-        port = inbound["port"]
-        method = inbound["settings"]["method"]
-        server_password = inbound["settings"]["password"]
-        client_password = cls.generate_password(method)
-
-        return cls(
-            host=host,
-            port=port,
-            method=method,
-            server_password=server_password,
-            client_password=client_password,
-            exit_point=exit_point,
-        )
-
-    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 = self.exit_point
-        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.exit_point}",
-            "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):
-            _, exit_point = self.split_client_email(c["email"])
-            return (priority.get(exit_point, 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.exit_point}"
-        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.method,
-                self.server_password,
-                self.client_password,
-                self.exit_point,
-            )
-        )
-
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, ShadowsocksConnection):
-            return NotImplemented
-
-        return (
-            self.host == other.host
-            and self.port == other.port
-            and self.method == other.method
-            and self.server_password == other.server_password
-            and self.client_password == other.client_password
-            and self.exit_point == other.exit_point
-        )
-
-
-@ConnectionFactory.register
-@dataclass(eq=False)
-class VlessConnection(Connection):
-    PROTOCOL = "vless"
-    LINK_SCHEME = "vless"
-
-    host: str
-    port: int
-    id: str
-    security: str
-    encryption: str
-    public_key: str
-    network_type: str
-    sni: str
-    short_id: str
-    exit_point: str
-
-    flow: str = "xtls-rprx-vision"
-    header_type: str = "none"
-    finger_print: str = "chrome"
-
-    @classmethod
-    def from_link(cls, link: str) -> "VlessConnection":
-        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"),
-            id=require_str(parsed.username, "id"),
-            security=get_required("security"),
-            encryption=get_required("encryption"),
-            public_key=get_required("pbk"),
-            header_type=get_required("headerType"),
-            finger_print=get_required("fp"),
-            network_type=get_required("type"),
-            flow=get_required("flow"),
-            sni=get_required("sni"),
-            short_id=get_required("sid"),
-            exit_point=require_str(parsed.fragment, "exit_point"),
-        )
-
-    @classmethod
-    def from_inbound(
-        cls, host: str, client: dict, inbound: dict
-    ) -> tuple[str, "VlessConnection"]:
-        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, exit_point = cls.split_client_email(email)
-        connection = cls(
-            host=host,
-            port=port,
-            id=id,
-            security=security,
-            encryption=encryption,
-            public_key=public_key,
-            network_type=network_type,
-            flow=flow,
-            sni=sni,
-            short_id=short_id,
-            exit_point=exit_point,
-        )
-        return username, connection
-
-    @classmethod
-    def from_spec(
-        cls, host: str, inbound: dict, exit_point: str
-    ) -> "VlessConnection":
-        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,
-            id=id,
-            security=security,
-            encryption=encryption,
-            public_key=public_key,
-            network_type=network_type,
-            sni=sni,
-            short_id=short_id,
-            exit_point=exit_point,
-        )
-
-    def to_link(self) -> str:
-        port = 443
-        netloc = f"{self.id}@{self.host}:{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.exit_point)
-        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.exit_point}",
-            "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
-            _, exit_point = self.split_client_email(client["email"])
-            return (priority.get(exit_point, 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.exit_point}"
-
-        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.id,
-                self.security,
-                self.encryption,
-                self.public_key,
-                self.header_type,
-                self.finger_print,
-                self.network_type,
-                self.flow,
-                self.sni,
-                self.short_id,
-                self.exit_point,
-            )
-        )
-
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, VlessConnection):
-            return NotImplemented
-
-        return (
-            self.host == other.host
-            and self.port == other.port
-            and self.id == other.id
-            and self.security == other.security
-            and self.encryption == other.encryption
-            and self.public_key == other.public_key
-            and self.header_type == other.header_type
-            and self.finger_print == other.finger_print
-            and self.network_type == other.network_type
-            and self.flow == other.flow
-            and self.sni == other.sni
-            and self.short_id == other.short_id
-            and self.exit_point == other.exit_point
-        )

src/xray_manager/core/outbound.py

@@ -8,11 +8,171 @@ 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:
+    from .target import Target, TargetStorage
+    from .xray_config import XrayConfig
+
+
+@dataclass(frozen=True)
+class OutboundSpec:
+    protocol: str
+    target_id: 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, target_storage: TargetStorage, 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(target_storage, link)
+
+    # устаревший метод, удалить после рефакторинга
+    @classmethod
+    def from_inbound(
+        cls,
+        target_storage: TargetStorage,
+        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(
+            target_storage, host, client, inbound
+        )
+        return username, outbound
+
+    @classmethod
+    def from_xray_inbound(
+        cls, ts: TargetStorage, client: dict, inbound: dict
+    ) -> Outbound:
+        protocol = inbound.get("protocol")
+
+        if protocol not in cls._registry:
+            raise ValueError(f"Unsupported protocol: {protocol}")
+
+        outbound_cls = cls._registry[protocol]
+        outbound = outbound_cls.from_xray_inbound(ts, client, inbound)
+        return outbound
+
+    @classmethod
+    def from_spec(
+        cls,
+        xray_config: XrayConfig,
+        target_storage: TargetStorage,
+        spec: OutboundSpec,
+    ) -> Outbound:
+        if spec.protocol not in cls._registry:
+            raise ValueError(f"Unsupported protocol: {spec.protocol}")
+
+        target = target_storage.load_by_id(spec.target_id)
+
+        inbound = xray_config.find_managed_inbound_by_protocol(spec.protocol)
+        outbound_cls = cls._registry[spec.protocol]
+
+        return outbound_cls.from_scratch(target, xray_config.host, inbound)
+
 
 class Outbound(ABC):
-    pass
+    PROTOCOL: str
+    LINK_SCHEME: str
+
+    host: str
+    port: int
+    target: Target
+
+    @classmethod
+    @abstractmethod
+    def from_link(
+        cls, target_storage: TargetStorage, link: str
+    ) -> Outbound: ...
+
+    @classmethod
+    @abstractmethod
+    def from_inbound(
+        cls,
+        target_storage: TargetStorage,
+        host: str,
+        client: dict,
+        inbound: dict,
+    ) -> tuple[str, Outbound]: ...
+
+    @classmethod
+    @abstractmethod
+    def from_xray_inbound(
+        cls, ts: TargetStorage, client: dict, inbound: dict
+    ) -> Outbound: ...
+
+    @classmethod
+    @abstractmethod
+    def from_scratch(
+        cls, target: Target, host: str, inbound: dict
+    ) -> 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_id = email.rsplit("-", 1)
+        return username, target_id
+
+    @abstractmethod
+    def __hash__(self) -> int: ...
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool: ...
 
 
 _ShadowsocksFields = make_dataclass(
@@ -20,14 +180,494 @@ _ShadowsocksFields = make_dataclass(
 )
 
 
+@OutboundFactory.register
 class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
-    pass
+    PROTOCOL = "shadowsocks"
+    LINK_SCHEME = "ss"
+
+    @classmethod
+    def from_link(
+        cls, target_storage: TargetStorage, link: str
+    ) -> ShadowsocksOutbound:
+        import base64
+
+        prefix_b64, rest = link[5:].split("@", 1)
+        host_port, quoted_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(":")
+
+        target_pretty_name = unquote(quoted_target_pretty_name)
+        target = target_storage.load_by_pretty(target_pretty_name)
+
+        return cls(
+            host=host,
+            port=port,
+            target=target,
+            method=method,
+            server_password=server_password,
+            client_password=client_password,
+        )
+
+    @classmethod
+    def from_inbound(
+        cls,
+        target_storage: TargetStorage,
+        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_id = cls.split_client_email(email)
+
+        target = target_storage.load_by_id(target_id)
+
+        outbound = cls(
+            host=host,
+            port=port,
+            target=target,
+            method=method,
+            server_password=server_password,
+            client_password=client_password,
+        )
+        return username, outbound
+
+    @classmethod
+    def from_scratch(
+        cls, target: Target, 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,
+            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)
+        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.id}",
+            "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_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)
+
+        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.id}"
+        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
+        )
 
 
-_VlessFields = make_dataclass(
-    "_VlessFieldss", fields=OUTBOUND_FIELDS["vless"]
-)
+_VlessFields = make_dataclass("_VlessFields", fields=OUTBOUND_FIELDS["vless"])
 
 
+@OutboundFactory.register
 class VlessOutbound(_VlessFields, Outbound):
-    pass
+    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
+        )

src/xray_manager/core/outbound_fields.py

@@ -1,20 +1,19 @@
 from dataclasses import field
 
+from .target import Target
 
-_BASE_FIELDS = [
-    ("host", str),
-    ("port", str),
-    ("target", str),
-]
+_BASE_FIELDS = [("host", str), ("port", str), ("target", Target)]
 
 
 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 +23,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")),
     ],
 }

src/xray_manager/core/profile.py

@@ -1,16 +1,11 @@
 from __future__ import annotations
 
-import json
-import secrets
-import uuid
-from abc import ABC, abstractmethod
-from collections import defaultdict
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, TypeVar
-from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
+from typing import TYPE_CHECKING
 
-from .connection import ConnectionFactory
+from .outbound import OutboundFactory
+from .target import TargetStorage
 from .user import User
 
 if TYPE_CHECKING:
@@ -23,7 +18,7 @@ class Profile:
     user: User
 
     def to_text(self):
-        links = [connection.to_link() for connection in self.user.connections]
+        links = [outbound.to_link() for outbound in self.user.outbounds]
         text = "\n".join(links)
         return text
 
@@ -94,12 +89,16 @@ class ProfileStorage:
             if line.strip()
         ]
 
-        connections = [ConnectionFactory.from_link(link) for link in links]
-        user = User(username, connections)
+        target_storage = TargetStorage(self.xrmc)
+
+        outbounds = [
+            OutboundFactory.from_link(target_storage, link) for link in links
+        ]
+        user = User(username, outbounds)
 
         profile = Profile(folder_name=path.parent.name, user=user)
         print(
-            f"[DEBUG] Loaded profile: folder='{profile.folder_name}', connections={len(connections)}"
+            f"[DEBUG] Loaded profile: folder='{profile.folder_name}', outbounds={len(outbounds)}"
         )
         return profile
 

src/xray_manager/core/target.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .xray_manager_config import XrayManagerConfig
+
+
+@dataclass(frozen=True)
+class Target:
+    id: str
+    pretty: str
+    route: str
+
+
+class TargetStorage:
+    def __init__(self, xray_manager_config: XrayManagerConfig):
+        self.xrmc = xray_manager_config
+
+    @property
+    def target_dict(self) -> dict[str, dict[str, str]]:
+        return self.xrmc.config.get("target", {})
+
+    def load_by_id(self, target_id: str) -> Target:
+        if target_id not in self.target_dict:
+            raise ValueError(f"Target not found: {target_id}")
+
+        target_data = self.target_dict[target_id]
+
+        if "route" not in target_data:
+            raise ValueError(
+                f"Missing required 'route' field for target: {target_id}"
+            )
+
+        return Target(
+            id=target_id,
+            pretty=target_data.get("pretty_name", target_id),
+            route=target_data["route"],
+        )
+
+    def load_by_pretty(self, target_pretty_name: str) -> Target:
+        for target_id, target_data in self.target_dict.items():
+            if target_data.get("pretty_name", target_id) == target_pretty_name:
+                return self.load_by_id(target_id)
+
+        raise ValueError(f"Target pretty name not found: {target_pretty_name}")
+
+    def load_all_targets(self) -> list[Target]:
+        targets = [
+            self.load_by_id(target_id) for target_id in self.target_dict.keys()
+        ]
+        return targets
+
+    def save_target(self, target: Target):
+        self.xrmc.set_target(target.id, target.pretty, target.route)
+
+    def delete_target(self, target: Target):
+        if target not in self.load_all_targets():
+            raise ValueError(f"Target not found: {target.id}")
+
+        self.xrmc.delete_target(target.id)

src/xray_manager/core/user.py

@@ -3,53 +3,98 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
-from .connection import ConnectionFactory, ConnectionKey
+from .outbound import Outbound, OutboundFactory, OutboundSpec
 
 if TYPE_CHECKING:
-    from .connection import Connection
-    from .xray_config import XrayConfig
+    from .target import TargetStorage
+    from .xray_config import XrayConfig, XrayStorage
 
 
 @dataclass
 class User:
     username: str
-    connections: list[Connection] = field(default_factory=list)
+    outbounds: list[Outbound] = field(default_factory=list)
 
-    def modify_connection_by_key(
+    def modify_outbounds(
         self,
-        xray_config: "XrayConfig",
-        add: list[ConnectionKey] = [],
-        delete: list[ConnectionKey] = [],
+        xray_config: XrayConfig,
+        target_storage: TargetStorage,
+        to_add: list[OutboundSpec] | None = None,
+        to_delete: list[OutboundSpec] | None = None,
     ):
-        key_map = {
-            ConnectionKey(c.protocol, c.exit_point): c for c in self.connections
+        to_add = to_add or []
+        to_delete = to_delete or []
+
+        user_specs = {
+            OutboundSpec(o.protocol, o.target.id): o for o in self.outbounds
         }
 
-        for key in delete:
-            key_map.pop(key, None)
+        for spec in to_delete:
+            user_specs.pop(spec, None)
 
-        for key in add:
-            if key not in key_map:
-                connection = ConnectionFactory.from_spec(key, xray_config)
-                key_map[key] = connection
+        for spec in to_add:
+            if spec not in user_specs:
+                outbound = OutboundFactory.from_spec(
+                    xray_config, target_storage, spec
+                )
+                user_specs[spec] = outbound
 
-        self.connections = list(key_map.values())
+        self.outbounds = list(user_specs.values())
+
+    def add_to_xray_inbounds(self, xrs: XrayStorage):
+        added = False
+        xray_inbounds = xrs.load_inbounds()
+        for outbound in self.outbounds:
+            for xray_inbound in xray_inbounds:
+                if outbound.matches_inbound(xray_inbound):
+                    outbound.add_to_inbound(xray_inbound, self.username)
+                    added = True
+
+        if not added:
+            raise RuntimeError(f"Inbound not found for {outbound}")
 
 
 class UserFactory:
     @staticmethod
     def from_spec(
-        username: str, keys: list[ConnectionKey], xray_config: "XrayConfig"
+        username: str,
+        specs: list[OutboundSpec],
+        xray_config: XrayConfig,
+        target_storage: TargetStorage,
     ) -> User:
-        connections = []
+        outbounds = []
 
-        for key in keys:
-            conn = ConnectionFactory.from_spec(key, xray_config)
-            connections.append(conn)
+        for spec in specs:
+            outbound = OutboundFactory.from_spec(
+                xray_config, target_storage, spec
+            )
+            outbounds.append(outbound)
 
-        user = User(username=username, connections=connections)
+        user = User(username=username, outbounds=outbounds)
         return user
 
     @staticmethod
-    def empty(username: str):
+    def empty(username: str) -> User:
         return User(username, [])
+
+    @staticmethod
+    def from_xray_inbound(username: str, ts: TargetStorage, xrs: XrayStorage):
+        user_outbounds: list[Outbound] = []
+
+        inbounds: list[dict] = xrs.load_inbounds()
+        for inbound in inbounds:
+            protocol = inbound.get("protocol")
+
+            if protocol in OutboundFactory._registry:
+                settings = inbound.get("settings", {})
+                clients = settings.get("clients", [])
+
+                for client in clients:
+                    email = client.get("email", "")
+                    inbound_username, _ = Outbound.split_client_email(email)
+
+                    if inbound_username == username:
+                        user_outbound = OutboundFactory.from_xray_inbound(
+                            ts, client, inbound
+                        )
+                        user_outbounds.append(user_outbound)

src/xray_manager/core/xray_config.py

@@ -5,17 +5,21 @@ from collections import defaultdict
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from .connection import ConnectionFactory
+from .outbound import OutboundFactory
+from .target import TargetStorage
 from .user import User
 
 if TYPE_CHECKING:
-    from .connection import Connection
+    from .outbound import Outbound
 
 
 class XrayConfig:
-    def __init__(self, config_folder: Path, host: str):
+    def __init__(
+        self, config_folder: Path, host: str, target_storage: TargetStorage
+    ):
         self.path = config_folder
         self.host = host
+        self.ts = target_storage
 
         self.inbounds_files = list(self.path.glob("*-in-*.json"))
         self.outbounds_files = list(self.path.glob("*-out-*.json"))
@@ -37,14 +41,14 @@ class XrayConfig:
             encoding="utf-8",
         )
 
-    def _find_inbound(self, connection: Connection) -> tuple[str, dict]:
+    def _find_inbound(self, outbound: Outbound) -> tuple[str, dict]:
         for filename, data in self.inbounds_data.items():
             inbounds = data.get("inbounds", [])
             for inbound in inbounds:
-                if connection.matches_inbound(inbound):
+                if outbound.matches_inbound(inbound):
                     return filename, inbound
 
-        raise RuntimeError(f"Inbound not found for {connection}")
+        raise RuntimeError(f"Inbound not found for {outbound}")
 
     def find_managed_inbound_by_protocol(self, protocol: str) -> dict:
         for data in self.inbounds_data.values():
@@ -62,39 +66,27 @@ class XrayConfig:
             f"Managed inbound not found for protocol: {protocol}"
         )
 
-    def get_exit_points(self) -> set[str]:
-        exit_points = set()
-        for data in self.outbounds_data.values():
-            outbounds = data.get("outbounds", [])
-            for outbound in outbounds:
-                tag: str = outbound.get("tag")
-                if tag:
-                    exit_point = tag.rsplit("-", 1)[1]
-                    exit_points.add(exit_point)
-        exit_points.add("default")
-        return exit_points
-
     def get_users(self) -> list[User]:
-        connection_map: defaultdict[str, list[Connection]] = defaultdict(list)
+        outbound_map: defaultdict[str, list[Outbound]] = defaultdict(list)
 
         for data in self.inbounds_data.values():
             inbounds = data.get("inbounds", [])
             for inbound in inbounds:
                 protocol = inbound.get("protocol")
 
-                if protocol in ConnectionFactory._registry:
+                if protocol in OutboundFactory._registry:
                     settings = inbound.get("settings", {})
                     clients = settings.get("clients", [])
                     for client in clients:
-                        username, connection = ConnectionFactory.from_inbound(
-                            self.host, client, inbound
+                        username, outbound = OutboundFactory.from_inbound(
+                            self.ts, self.host, client, inbound
                         )
-                        connection_map[username].append(connection)
+                        outbound_map[username].append(outbound)
 
         users = []
 
-        for username, connections in connection_map.items():
-            users.append(User(username, connections))
+        for username, outbounds in outbound_map.items():
+            users.append(User(username, outbounds))
 
         return users
 
@@ -107,30 +99,26 @@ class XrayConfig:
     def add_user(self, user: User) -> None:
         modified_files = set()
 
-        for connection in user.connections:
-            filename, inbound = self._find_inbound(connection)
-            connection.add_to_inbound(inbound, user.username)
+        for outbound in user.outbounds:
+            filename, inbound = self._find_inbound(outbound)
+            outbound.add_to_inbound(inbound, user.username)
             modified_files.add(filename)
 
         for filename in modified_files:
             path = self.path / filename
             self._save_json(self.inbounds_data[filename], path)
 
-    def delete_user(self, username: str) -> bool:
-        user = self.get_user(username)
-        if not user:
-            return False
-
+    def delete_user(self, user: User) -> bool:
         deleted = False
         modified_files = set()
 
-        for connection in user.connections:
+        for outbound in user.outbounds:
             try:
-                filename, inbound = self._find_inbound(connection)
+                filename, inbound = self._find_inbound(outbound)
             except RuntimeError:
                 continue
 
-            connection.delete_from_inbound(inbound, username)
+            outbound.delete_from_inbound(inbound, user.username)
             modified_files.add(filename)
             deleted = True
 
@@ -146,22 +134,22 @@ class XrayConfig:
         if current_user is None:
             return False
 
-        old_connections = set(current_user.connections)
-        new_connections = set(user.connections)
+        old_outbounds = set(current_user.outbounds)
+        new_outbounds = set(user.outbounds)
 
-        to_add = new_connections - old_connections
-        to_remove = old_connections - new_connections
+        to_add = new_outbounds - old_outbounds
+        to_remove = old_outbounds - new_outbounds
 
         modified_files = set()
 
-        for connection in to_remove:
-            filename, inbound = self._find_inbound(connection)
-            connection.delete_from_inbound(inbound, user.username)
+        for outbound in to_remove:
+            filename, inbound = self._find_inbound(outbound)
+            outbound.delete_from_inbound(inbound, user.username)
             modified_files.add(filename)
 
-        for connection in to_add:
-            filename, inbound = self._find_inbound(connection)
-            connection.add_to_inbound(inbound, user.username)
+        for outbound in to_add:
+            filename, inbound = self._find_inbound(outbound)
+            outbound.add_to_inbound(inbound, user.username)
             modified_files.add(filename)
 
         for filename in modified_files:
@@ -169,3 +157,19 @@ class XrayConfig:
             self._save_json(self.inbounds_data[filename], path)
 
         return True
+
+
+class XrayStorage:
+    def __init__(self, config_path: Path):
+        self.path = config_path
+
+        self.inbounds_files: list
+        self.outbound_files: list
+        self.routings_files: list
+
+        self.inbounds: list[dict]
+        self.outbouns: list[dict]
+        self.routings: list[dict]
+
+    def load_inbounds(self) -> list[dict]:
+        return self.inbounds

src/xray_manager/core/xray_manager_config.py

@@ -80,3 +80,18 @@ class XrayManagerConfig:
     def delete_profile_folder(self, username: str):
         self.config["profiles"]["folder_mapping"].pop(username, None)
         self._save_json()
+
+    def set_target(
+        self, target_id: str, target_pretty_name: str, target_route: str
+    ):
+        self.config["target"][target_id] = {
+            "pretty_name": target_pretty_name,
+            "route": target_route,
+        }
+        self.config["target"][target_id]["pretty_name"] = target_pretty_name
+        self.config["target"][target_id]["route"] = target_route
+        self._save_json()
+
+    def delete_target(self, target_id: str):
+        self.config["target"].pop(target_id, None)
+        self._save_json()

src/xray_manager/handlers.py

@@ -2,11 +2,12 @@ import os
 from pathlib import Path
 
 from .core import (
-    ConnectionFactory,
-    ConnectionKey,
+    OutboundFactory,
+    OutboundSpec,
     Profile,
     ProfileFactory,
     ProfileStorage,
+    TargetStorage,
     User,
     UserFactory,
     XrayConfig,
@@ -34,11 +35,15 @@ def _resolve_config_path() -> str:
 
 xray_manager_config = XrayManagerConfig(_resolve_config_path())
 
-xray_config = XrayConfig(
-    xray_manager_config.xray_config_folder, xray_manager_config.host
-)
+profile_storage = ProfileStorage(xray_manager_config)
 
-storage = ProfileStorage(xray_manager_config)
+target_storage = TargetStorage(xray_manager_config)
+
+xray_config = XrayConfig(
+    xray_manager_config.xray_config_folder,
+    xray_manager_config.host,
+    target_storage,
+)
 
 
 def _resolve_users(username: str | None, all: bool) -> list[User]:
@@ -54,35 +59,37 @@ def _resolve_users(username: str | None, all: bool) -> list[User]:
 
 def _resolve_profiles(username: str | None, all: bool) -> list[Profile]:
     if all:
-        return storage.load_profiles()
+        return profile_storage.load_profiles()
     if username:
         try:
-            profile = storage.load_profile(username)
+            profile = profile_storage.load_profile(username)
         except (ValueError, FileNotFoundError) as e:
             raise ValueError(f"Cannot load profile for {username}: {e}")
         return [profile]
     raise ValueError("Either username or --all must be specified")
 
 
-def _parse_connection_spec(spec: str) -> ConnectionKey:
-    protocol, exit_point = spec.split(":", 1)
-    if protocol not in ConnectionFactory.get_protocols():
+def _parse_outbound_spec(spec: str) -> OutboundSpec:
+    protocol, target_id = spec.split(":", 1)
+    if protocol not in OutboundFactory.get_protocols():
         raise ValueError(f"Unsupported protocol: {protocol}")
-    if exit_point not in xray_config.get_exit_points():
-        raise ValueError(f"Unsupported exit point: {exit_point}")
-    return ConnectionKey(protocol, exit_point)
+    if target_id not in {
+        target.id for target in target_storage.load_all_targets()
+    }:
+        raise ValueError(f"Target ID not found: {target_id}")
+    return OutboundSpec(protocol, target_id)
 
 
 def _build_user_rows(users: list[User]) -> list[list[str]]:
     rows = []
     for index, user in enumerate(users, start=1):
         first = True
-        for conn in user.connections:
+        for out in user.outbounds:
             if first:
-                row = [index, user.username, conn.exit_point, conn.protocol]
+                row = [index, user.username, out.target.id, out.protocol]
                 first = False
             else:
-                row = ["", "", conn.exit_point, conn.protocol]
+                row = ["", "", out.target.id, out.protocol]
             rows.append(row)
     return rows
 
@@ -91,18 +98,18 @@ def _build_profile_rows(profiles: list[Profile]) -> list[list[str]]:
     rows = []
     for index, profile in enumerate(profiles, start=1):
         first = True
-        for conn in profile.user.connections:
+        for out in profile.user.outbounds:
             if first:
                 row = [
                     index,
                     profile.user.username,
                     profile.folder_name,
-                    conn.exit_point,
-                    conn.protocol,
+                    out.target.id,
+                    out.protocol,
                 ]
                 first = False
             else:
-                row = ["", "", "", conn.exit_point, conn.protocol]
+                row = ["", "", "", out.target.id, out.protocol]
             rows.append(row)
     return rows
 
@@ -113,12 +120,15 @@ def user_add(username: str, connection: list[str], **_):
     if isinstance(user, User):
         raise ValueError(f"User already exists: {username}")
 
-    keys: list[ConnectionKey] = [
-        _parse_connection_spec(spec) for spec in connection
+    specs: list[OutboundSpec] = [
+        _parse_outbound_spec(raw_spec) for raw_spec in connection
     ]
 
     user = UserFactory.from_spec(
-        username=username, keys=keys, xray_config=xray_config
+        username=username,
+        specs=specs,
+        xray_config=xray_config,
+        target_storage=target_storage,
     )
 
     xray_config.add_user(user)
@@ -127,7 +137,7 @@ def user_add(username: str, connection: list[str], **_):
 def user_show(username: str | None, all: bool, **_):
     users = _resolve_users(username, all)
 
-    headers = ["#", "Username", "Exit Point", "Protocol"]
+    headers = ["#", "Username", "Target ID", "Protocol"]
     rows = _build_user_rows(users)
 
     print_table(headers, rows)
@@ -139,10 +149,12 @@ def user_modify(username: str, to_add: list[str], to_delete: list[str], **_):
     if not isinstance(user, User):
         raise ValueError(f"User not found: {username}")
 
-    add_key = [_parse_connection_spec(spec) for spec in to_add]
-    del_key = [_parse_connection_spec(spec) for spec in to_delete]
+    to_add_specs = [_parse_outbound_spec(spec) for spec in to_add]
+    to_del_specs = [_parse_outbound_spec(spec) for spec in to_delete]
 
-    user.modify_connection_by_key(xray_config, add_key, del_key)
+    user.modify_outbounds(
+        xray_config, target_storage, to_add_specs, to_del_specs
+    )
     xray_config.modify_user(user)
 
 
@@ -150,7 +162,7 @@ def user_delete(username: str | None, all: bool, **_):
     users = _resolve_users(username, all)
 
     for user in users:
-        xray_config.delete_user(user.username)
+        xray_config.delete_user(user)
 
 
 def profile_build(username: str | None, all: bool, **_):
@@ -158,19 +170,19 @@ def profile_build(username: str | None, all: bool, **_):
 
     for user in users:
         try:
-            profile = storage.load_profile(user.username)
+            profile = profile_storage.load_profile(user.username)
             profile = ProfileFactory.from_user(user, profile.folder_name)
         except (ValueError, FileNotFoundError) as e:
             print(f"Cannot load profile for {user.username}: {e}")
             profile = ProfileFactory.from_user(user)
 
-        storage.save_profile(profile)
+        profile_storage.save_profile(profile)
 
 
 def profile_show(username: str | None, all: bool, **_):
     profiles = _resolve_profiles(username, all)
 
-    headers = ["#", "Username", "Profile", "Exit Point", "Protocol"]
+    headers = ["#", "Username", "Profile", "Target ID", "Protocol"]
     rows = _build_profile_rows(profiles)
 
     print_table(headers, rows)
@@ -180,14 +192,14 @@ def profile_clear(username: str | None, all: bool, **_):
     profiles = _resolve_profiles(username, all)
 
     for profile in profiles:
-        storage.clear_profile(profile.user.username)
+        profile_storage.clear_profile(profile.user.username)
 
 
 def profile_delete(username: str | None, all: bool, **_):
     profiles = _resolve_profiles(username, all)
 
     for profile in profiles:
-        storage.delete_profile(profile.user.username)
+        profile_storage.delete_profile(profile.user.username)
 
 
 def profile_qrcode(username: str | None, all: bool, **_):

tests/mock_data/etc/xray-manager/config.json

@@ -5,6 +5,38 @@
         "host": "localhost.internal",
         "base_path": "tests/mock_data/var/lib/xray-manager/profiles",
         "file_name": "list.txt",
-        "folder_mapping": {}
+        "folder_mapping": {
+            "test": "OiVqwjq0V9kI",
+            "test2": "P8TC4FUicrvC"
+        }
+    },
+    "target": {
+        "default": {
+            "route": "out-fr1",
+            "pretty_name": "🚩 Default"
+        },
+        "fr1": {
+            "route": "out-fr1",
+            "pretty_name": "🇫🇷 France #1"
+        },
+        "us1": {
+            "route": "out-us1",
+            "pretty_name": "🇺🇸 USA #1"
+        },
+        "us2": {
+            "route": "out-us2",
+            "pretty_name": "🇺🇸 USA #2"
+        }
+    },
+    "override": {
+        "test": {
+            "vless": {
+                "port": 443,
+                "finger_print": "qq",
+                "by_target_id": {
+                    "us2": "chrome"
+                }
+            }
+        }
     }
 }

tests/sandbox/shadowsocks_outbound.py

@@ -0,0 +1,44 @@
+from xray_manager.core.outbound import (
+    OutboundFactory,
+    OutboundSpec,
+    ShadowsocksOutbound,
+)
+from xray_manager.core.target import Target, TargetStorage
+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_storage = TargetStorage(xray_manager_config)
+
+    target = target_storage.load_by_id("default")
+
+    outbound = ShadowsocksOutbound.from_scratch(target, "127.0.0.1", inbound)
+
+    link = outbound.to_link()
+    print(link)
+    o = ShadowsocksOutbound.from_link(target_storage, link)
+    print(o)

tests/sandbox/target.py

@@ -0,0 +1,13 @@
+from xray_manager.core.target import Target, TargetStorage
+from xray_manager.core.xray_manager_config import XrayManagerConfig
+
+DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
+
+
+if __name__ == "__main__":
+    xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
+    storage = TargetStorage(xray_manager_config)
+
+    target = Target("de1", "pretty", "out-de1")
+
+    storage.delete_target(target)

tests/sandbox/vless_outbound.py

@@ -0,0 +1,95 @@
+from xray_manager.core import (
+    OutboundFactory,
+    ProfileStorage,
+    Target,
+    TargetStorage,
+    VlessOutbound,
+    XrayManagerConfig,
+)
+
+inbound = {
+    "tag": "in-vless",
+    "listen": "127.0.0.1",
+    "port": 443,
+    "protocol": "vless",
+    "settings": {
+        "clients": [
+            {
+                "email": "test-fr1",
+                "id": "0eb7a1a8-c24f-4eb2-9f89-8023a10ed070",
+                "flow": "xtls-rprx-vision",
+            },
+            {
+                "email": "test2-us2",
+                "id": "7b6da4ce-bb10-4127-a0c6-243618c62edd",
+                "flow": "xtls-rprx-vision",
+            },
+        ],
+        "decryption": "none",
+    },
+    "streamSettings": {
+        "network": "tcp",
+        "security": "reality",
+        "realitySettings": {
+            "show": False,
+            "dest": "google.com:443",
+            "xver": 0,
+            "serverNames": ["google.com"],
+            "privateKey": "yHJQq57MNRi-0UQm7ymiy2DI17iO_Ovu2HF5EV1ViGg",
+            "minClientVer": "",
+            "maxClientVer": "",
+            "maxTimeDiff": 0,
+            "shortIds": ["91e7c34ea2c07723", "0419a64d06467571"],
+        },
+    },
+}
+
+
+def test_from_scratch(target_storage: TargetStorage) -> VlessOutbound:
+    target = target_storage.load_by_id("fr1")
+    outbound = VlessOutbound.from_scratch(target, "localhost", inbound)
+    print("Test from scratch")
+    print(outbound)
+    return outbound
+
+
+def test_to_link(outbound: VlessOutbound) -> str:
+    link = outbound.to_link()
+    print("Test to link")
+    print(link)
+    return link
+
+
+def test_from_link(target_storage: TargetStorage, link: str):
+    outbound = VlessOutbound.from_link(target_storage, link)
+    print("Test from link")
+    print(outbound)
+
+
+def test_from_inbound(target_storage: TargetStorage):
+    client = {
+        "email": "test-fr1",
+        "id": "0eb7a1a8-c24f-4eb2-9f89-8023a10ed070",
+        "flow": "xtls-rprx-vision",
+    }
+
+    username, outbound = VlessOutbound.from_inbound(
+        target_storage, "localhost", client, inbound
+    )
+
+    print("Test from inbound")
+    print(username)
+    print(outbound)
+
+
+if __name__ == "__main__":
+    DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
+    xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
+    profile_storage = ProfileStorage(xray_manager_config)
+    target_storage = TargetStorage(xray_manager_config)
+
+    outbound = test_from_scratch(target_storage)
+    link = test_to_link(outbound)
+    test_from_link(target_storage, link)
+
+    test_from_inbound(target_storage)