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

1. outbound вместо connection для пользовательских соединений, так как лучше отражает генерируемую сущность
2. добавление сущности server по аналогии с user

README.md

@@ -0,0 +1,36 @@
+# Xray Manager
+
+## Список команд
+```
+user add <username> [-o <protocol>:<target> [<protocol>:<target> ...]]
+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 profile build (<username> | -a)
+user profile show (<username> | -a)
+user profile qrcode (<username> | -a)
+user profile delete (<username> | -a)
+
+target add <target_name> -r <server_outbound_tag> [--pretty "<display_name>"]
+target modify <target_name> [-r <server_outbound_tag>] [--pretty "<display_name>"]
+target show (<target_name> | -a)
+target delete (<target_name> | -a)
+
+protocol <protocol_name>
+
+server inbound add -t <tag> -p <protocol> [--<param> <value> ...]
+server inbound modify -t <tag> -p <protocol> [--<param> <value> ...]
+server inbound show (-a | -t <tag> -p <protocol>)
+server inbound delete (-a | -t <tag> -p <protocol>)
+
+server outbound add -t <tag> -p <protocol> [--<param> <value> ...]
+server outbound modify -t <tag> -p <protocol> [--<param> <value> ...]
+server outbound show (-a | -t <tag> -p <protocol>)
+server outbound delete (-a | -t <tag> -p <protocol>)
+```

src/xray_manager/core/__init__.py

@@ -0,0 +1,5 @@
+from .connection import Connection, ConnectionFactory, ConnectionKey
+from .profile import Profile, ProfileFactory, ProfileStorage
+from .user import User, UserFactory
+from .xray_config import XrayConfig
+from .xray_manager_config import XrayManagerConfig

src/xray_manager/core/connection.py

@@ -1,14 +1,17 @@
-import json
+from __future__ import annotations
+
 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 typing import TypeVar
 from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
 
-from .utils import private_to_public
+from ..utils import private_to_public
+
+if TYPE_CHECKING:
+    from .xray_config import XrayConfig
 
 C = TypeVar("C", bound="Connection")
 T = TypeVar("T")
@@ -620,443 +623,3 @@ class VlessConnection(Connection):
             and self.short_id == other.short_id
             and self.exit_point == other.exit_point
         )
-
-
-@dataclass
-class User:
-    username: str
-    connections: list[Connection] = field(default_factory=list)
-
-    def modify_connection_by_key(
-        self,
-        xray_config: "XrayConfig",
-        add: list[ConnectionKey] = [],
-        delete: list[ConnectionKey] = [],
-    ):
-        key_map = {
-            ConnectionKey(c.protocol, c.exit_point): c for c in self.connections
-        }
-
-        for key in delete:
-            key_map.pop(key, None)
-
-        for key in add:
-            if key not in key_map:
-                connection = ConnectionFactory.from_spec(key, xray_config)
-                key_map[key] = connection
-
-        self.connections = list(key_map.values())
-
-
-class UserFactory:
-    @staticmethod
-    def from_spec(
-        username: str, keys: list[ConnectionKey], xray_config: "XrayConfig"
-    ) -> User:
-        connections = []
-
-        for key in keys:
-            conn = ConnectionFactory.from_spec(key, xray_config)
-            connections.append(conn)
-
-        user = User(username=username, connections=connections)
-        return user
-
-    @staticmethod
-    def empty(username: str):
-        return User(username, [])
-
-
-class XrayManagerConfig:
-    def __init__(self, config_path: str):
-        self.path = Path(config_path)
-        self.config = json.loads(self.path.read_text(encoding="utf-8"))
-
-    def _save_json(self):
-        self.path.write_text(
-            json.dumps(self.config, indent=4, ensure_ascii=False) + "\n",
-            encoding="utf-8",
-        )
-
-    def _update(self, path: list[str], value):
-        ref = self.config
-        for key in path[:-1]:
-            ref = ref[key]
-
-        ref[path[-1]] = value
-
-        self._save_json()
-
-    @property
-    def host(self) -> str:
-        return self.config["host"]
-
-    @host.setter
-    def host(self, value: str):
-        self._update(["host"], value)
-
-    @property
-    def xray_config_folder(self) -> Path:
-        return Path(self.config["xray_config_folder"])
-
-    @xray_config_folder.setter
-    def xray_config_folder(self, value: str):
-        self._update(["xray_config_folder"], value)
-
-    @property
-    def profile_host(self) -> str:
-        return self.config["profiles"]["host"]
-
-    @profile_host.setter
-    def profile_host(self, value: str):
-        self._update(["profiles", "host"], value)
-
-    @property
-    def profile_base_path(self) -> Path:
-        return Path(self.config["profiles"]["base_path"])
-
-    @profile_base_path.setter
-    def profile_base_path(self, value: str):
-        self._update(["profiles", "base_path"], value)
-
-    @property
-    def profile_file_name(self) -> str:
-        return self.config["profiles"]["file_name"]
-
-    @profile_file_name.setter
-    def profile_file_name(self, value: str):
-        self._update(["profiles", "file_name"], value)
-
-    @property
-    def profile_folder_mapping(self) -> dict[str, str]:
-        return self.config["profiles"]["folder_mapping"]
-
-    @property
-    def folder_profile_mapping(self) -> dict[str, str]:
-        return {
-            folder_name: username
-            for username, folder_name in self.profile_folder_mapping.items()
-        }
-
-    def set_profile_folder(self, username: str, folder_name: str):
-        self.config["profiles"]["folder_mapping"][username] = folder_name
-        self._save_json()
-
-    def delete_profile_folder(self, username: str):
-        self.config["profiles"]["folder_mapping"].pop(username, None)
-        self._save_json()
-
-
-class XrayConfig:
-    def __init__(self, config_folder: Path, host: str):
-        self.path = config_folder
-        self.host = host
-
-        self.inbounds_files = list(self.path.glob("*-in-*.json"))
-        self.outbounds_files = list(self.path.glob("*-out-*.json"))
-
-        self.inbounds_data = {
-            f.name: self._load_json(f) for f in self.inbounds_files
-        }
-
-        self.outbounds_data = {
-            f.name: self._load_json(f) for f in self.outbounds_files
-        }
-
-    def _load_json(self, path: Path) -> dict:
-        return json.loads(path.read_text(encoding="utf-8"))
-
-    def _save_json(self, data: dict, path: Path):
-        path.write_text(
-            json.dumps(data, indent=4, ensure_ascii=False) + "\n",
-            encoding="utf-8",
-        )
-
-    def _find_inbound(self, connection: Connection) -> tuple[str, dict]:
-        for filename, data in self.inbounds_data.items():
-            inbounds = data.get("inbounds", [])
-            for inbound in inbounds:
-                if connection.matches_inbound(inbound):
-                    return filename, inbound
-
-        raise RuntimeError(f"Inbound not found for {connection}")
-
-    def find_managed_inbound_by_protocol(self, protocol: str) -> dict:
-        for data in self.inbounds_data.values():
-            inbounds = data.get("inbounds", [])
-            for inbound in inbounds:
-                if not inbound.get("xrm"):
-                    continue
-
-                if inbound.get("protocol") != protocol:
-                    continue
-
-                return inbound
-
-        raise RuntimeError(
-            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)
-
-        for data in self.inbounds_data.values():
-            inbounds = data.get("inbounds", [])
-            for inbound in inbounds:
-                protocol = inbound.get("protocol")
-
-                if protocol in ConnectionFactory._registry:
-                    settings = inbound.get("settings", {})
-                    clients = settings.get("clients", [])
-                    for client in clients:
-                        username, connection = ConnectionFactory.from_inbound(
-                            self.host, client, inbound
-                        )
-                        connection_map[username].append(connection)
-
-        users = []
-
-        for username, connections in connection_map.items():
-            users.append(User(username, connections))
-
-        return users
-
-    def get_user(self, username: str) -> User | None:
-        users = self.get_users()
-        for user in users:
-            if user.username == username:
-                return user
-
-    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)
-            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
-
-        deleted = False
-        modified_files = set()
-
-        for connection in user.connections:
-            try:
-                filename, inbound = self._find_inbound(connection)
-            except RuntimeError:
-                continue
-
-            connection.delete_from_inbound(inbound, username)
-            modified_files.add(filename)
-            deleted = True
-
-        for filename in modified_files:
-            path = self.path / filename
-            self._save_json(self.inbounds_data[filename], path)
-
-        return deleted
-
-    def modify_user(self, user: User) -> bool:
-        current_user = self.get_user(user.username)
-
-        if current_user is None:
-            return False
-
-        old_connections = set(current_user.connections)
-        new_connections = set(user.connections)
-
-        to_add = new_connections - old_connections
-        to_remove = old_connections - new_connections
-
-        modified_files = set()
-
-        for connection in to_remove:
-            filename, inbound = self._find_inbound(connection)
-            connection.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)
-            modified_files.add(filename)
-
-        for filename in modified_files:
-            path = self.path / filename
-            self._save_json(self.inbounds_data[filename], path)
-
-        return True
-
-
-@dataclass
-class Profile:
-    folder_name: str
-    user: User
-
-    def to_text(self):
-        links = [connection.to_link() for connection in self.user.connections]
-        text = "\n".join(links)
-        return text
-
-
-class ProfileFactory:
-    @classmethod
-    def from_user(cls, user: User, folder_name: str | None = None):
-        if not folder_name:
-            folder_name = cls.generate_folder_name()
-
-        return Profile(folder_name=folder_name, user=user)
-
-    @staticmethod
-    def generate_folder_name() -> str:
-        import secrets
-        import string
-
-        folder_name = "".join(
-            secrets.choice(string.ascii_letters + string.digits)
-            for _ in range(12)
-        )
-
-        return folder_name
-
-
-class ProfileStorage:
-    def __init__(self, xray_manager_config: XrayManagerConfig):
-        self.xrmc = xray_manager_config
-
-    def _get_profile_path(self, username: str) -> Path:
-        folder_name = self.xrmc.profile_folder_mapping.get(username)
-
-        if not folder_name:
-            raise ValueError(f"No folder found for username: {username}")
-
-        path = (
-            self.xrmc.profile_base_path
-            / folder_name
-            / self.xrmc.profile_file_name
-        )
-
-        if not path.exists():
-            raise FileNotFoundError(f"Profile path does not exist: {path}")
-
-        print(f"[DEBUG] Found profile path for user '{username}': {path}")
-        return path
-
-    def _create_profile_path(self, username: str, folder_name: str) -> Path:
-        path = (
-            self.xrmc.profile_base_path
-            / folder_name
-            / self.xrmc.profile_file_name
-        )
-        path.parent.mkdir(parents=True, exist_ok=False)
-        path.touch(exist_ok=False)
-
-        self.xrmc.set_profile_folder(username, folder_name)
-
-        print(f"[DEBUG] Created profile path for user '{username}': {path}")
-        return path
-
-    def _load_profile(self, path: Path, username: str) -> Profile:
-        print(f"[DEBUG] Loading profile for user '{username}' from {path}")
-
-        links = [
-            line
-            for line in path.read_text(encoding="utf-8").splitlines()
-            if line.strip()
-        ]
-
-        connections = [ConnectionFactory.from_link(link) for link in links]
-        user = User(username, connections)
-
-        profile = Profile(folder_name=path.parent.name, user=user)
-        print(
-            f"[DEBUG] Loaded profile: folder='{profile.folder_name}', connections={len(connections)}"
-        )
-        return profile
-
-    def load_profile(self, username: str) -> Profile:
-        path = self._get_profile_path(username)
-        profile = self._load_profile(path, username)
-        return profile
-
-    def load_profiles(self) -> list[Profile]:
-        storage_path = self.xrmc.profile_base_path
-
-        profiles = []
-        for profile_path in storage_path.iterdir():
-            if not profile_path.is_dir():
-                continue
-
-            folder_name = profile_path.name
-
-            username = self.xrmc.folder_profile_mapping.get(folder_name)
-            if not username:
-                print(
-                    f"[WARN] Unknown profile folder: {profile_path.name}, skipping"
-                )
-                continue
-
-            profile_file = profile_path / self.xrmc.profile_file_name
-            if not profile_file.exists():
-                print(
-                    f"[WARN] Unknown profile folder: {profile_path.name}, skipping"
-                )
-                continue
-
-            profile = self._load_profile(profile_file, username)
-            profiles.append(profile)
-
-        print(f"[DEBUG] Loaded total {len(profiles)} profiles")
-        return profiles
-
-    def save_profile(self, profile: Profile) -> Path:
-        try:
-            path = self._create_profile_path(
-                profile.user.username, profile.folder_name
-            )
-        except FileExistsError:
-            print(
-                f"[INFO] Profile already exists for user '{profile.user.username}', overwriting"
-            )
-            path = self._get_profile_path(profile.user.username)
-
-        path.write_text(profile.to_text(), encoding="utf-8")
-        print(
-            f"[DEBUG] Saved profile for user '{profile.user.username}' at '{path}'"
-        )
-        return path
-
-    def clear_profile(self, username: str) -> None:
-        path = self._get_profile_path(username)
-        path.write_text("", encoding="utf-8")
-        print(f"[INFO] Cleared profile for user '{username}'")
-
-    def delete_profile(self, username: str) -> None:
-        path = self._get_profile_path(username)
-
-        path.unlink()
-        print(f"[INFO] Deleted profile file for user '{username}'")
-
-        path.parent.rmdir()
-        print(f"[INFO] Deleted profile folder for user '{username}'")
-
-        self.xrmc.delete_profile_folder(username)
-        print(f"[INFO] Removed folder mapping for user '{username}'")

src/xray_manager/core/outbound.py

@@ -0,0 +1,385 @@
+from __future__ import annotations
+
+import secrets
+import uuid
+from abc import ABC, abstractmethod
+from collections import defaultdict
+from dataclasses import dataclass, make_dataclass
+from typing import TYPE_CHECKING, TypeVar
+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, 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]
+        targets = xray_manager_config.targets
+        return outbound_cls.from_link(targets, link)
+
+    @classmethod
+    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]
+        targets = xray_manager_config.targets
+        username, outbound = outbound_cls.from_inbound(
+            targets, 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]
+
+        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):
+    PROTOCOL: str
+    LINK_SCHEME: str
+
+    @classmethod
+    @abstractmethod
+    def from_link(
+        cls, targets: dict[str, dict[str, str]], link: str
+    ) -> Outbound: ...
+
+    @classmethod
+    @abstractmethod
+    def from_inbound(
+        cls,
+        targets: dict[str, dict[str, str]],
+        host: str,
+        client: dict,
+        inbound: dict,
+    ) -> tuple[str, Outbound]: ...
+
+    @classmethod
+    @abstractmethod
+    def from_scratch(
+        cls, target: str, target_pretty_name: str, 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 = 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: ...
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool: ...
+
+
+_ShadowsocksFields = make_dataclass(
+    "_ShadowsocksFields", fields=OUTBOUND_FIELDS["shadowsocks"]
+)
+
+
+@OutboundFactory.register
+class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
+    PROTOCOL = "shadowsocks"
+    LINK_SCHEME = "ss"
+
+    @classmethod
+    def from_link(
+        cls, targets: dict[str, dict[str, str]], 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 = 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,
+        )
+
+    @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
+        )
+
+
+_VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
+
+
+@OutboundFactory.register
+class VlessOutbound(_VlessFields, Outbound):
+    PROTOCOL = "vless"
+    LINK_SCHEME = "vless"

src/xray_manager/core/outbound_fields.py

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

src/xray_manager/core/profile.py

@@ -0,0 +1,173 @@
+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 pathlib import Path
+from typing import TYPE_CHECKING, TypeVar
+from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
+
+from .connection import ConnectionFactory
+from .user import User
+
+if TYPE_CHECKING:
+    from .xray_manager_config import XrayManagerConfig
+
+
+@dataclass
+class Profile:
+    folder_name: str
+    user: User
+
+    def to_text(self):
+        links = [connection.to_link() for connection in self.user.connections]
+        text = "\n".join(links)
+        return text
+
+
+class ProfileFactory:
+    @classmethod
+    def from_user(cls, user: User, folder_name: str | None = None):
+        if not folder_name:
+            folder_name = cls.generate_folder_name()
+
+        return Profile(folder_name=folder_name, user=user)
+
+    @staticmethod
+    def generate_folder_name() -> str:
+        import secrets
+        import string
+
+        folder_name = "".join(
+            secrets.choice(string.ascii_letters + string.digits)
+            for _ in range(12)
+        )
+
+        return folder_name
+
+
+class ProfileStorage:
+    def __init__(self, xray_manager_config: XrayManagerConfig):
+        self.xrmc = xray_manager_config
+
+    def _get_profile_path(self, username: str) -> Path:
+        folder_name = self.xrmc.profile_folder_mapping.get(username)
+
+        if not folder_name:
+            raise ValueError(f"No folder found for username: {username}")
+
+        path = (
+            self.xrmc.profile_base_path
+            / folder_name
+            / self.xrmc.profile_file_name
+        )
+
+        if not path.exists():
+            raise FileNotFoundError(f"Profile path does not exist: {path}")
+
+        print(f"[DEBUG] Found profile path for user '{username}': {path}")
+        return path
+
+    def _create_profile_path(self, username: str, folder_name: str) -> Path:
+        path = (
+            self.xrmc.profile_base_path
+            / folder_name
+            / self.xrmc.profile_file_name
+        )
+        path.parent.mkdir(parents=True, exist_ok=False)
+        path.touch(exist_ok=False)
+
+        self.xrmc.set_profile_folder(username, folder_name)
+
+        print(f"[DEBUG] Created profile path for user '{username}': {path}")
+        return path
+
+    def _load_profile(self, path: Path, username: str) -> Profile:
+        print(f"[DEBUG] Loading profile for user '{username}' from {path}")
+
+        links = [
+            line
+            for line in path.read_text(encoding="utf-8").splitlines()
+            if line.strip()
+        ]
+
+        connections = [ConnectionFactory.from_link(link) for link in links]
+        user = User(username, connections)
+
+        profile = Profile(folder_name=path.parent.name, user=user)
+        print(
+            f"[DEBUG] Loaded profile: folder='{profile.folder_name}', connections={len(connections)}"
+        )
+        return profile
+
+    def load_profile(self, username: str) -> Profile:
+        path = self._get_profile_path(username)
+        profile = self._load_profile(path, username)
+        return profile
+
+    def load_profiles(self) -> list[Profile]:
+        storage_path = self.xrmc.profile_base_path
+
+        profiles = []
+        for profile_path in storage_path.iterdir():
+            if not profile_path.is_dir():
+                continue
+
+            folder_name = profile_path.name
+
+            username = self.xrmc.folder_profile_mapping.get(folder_name)
+            if not username:
+                print(
+                    f"[WARN] Unknown profile folder: {profile_path.name}, skipping"
+                )
+                continue
+
+            profile_file = profile_path / self.xrmc.profile_file_name
+            if not profile_file.exists():
+                print(
+                    f"[WARN] Unknown profile folder: {profile_path.name}, skipping"
+                )
+                continue
+
+            profile = self._load_profile(profile_file, username)
+            profiles.append(profile)
+
+        print(f"[DEBUG] Loaded total {len(profiles)} profiles")
+        return profiles
+
+    def save_profile(self, profile: Profile) -> Path:
+        try:
+            path = self._create_profile_path(
+                profile.user.username, profile.folder_name
+            )
+        except FileExistsError:
+            print(
+                f"[INFO] Profile already exists for user '{profile.user.username}', overwriting"
+            )
+            path = self._get_profile_path(profile.user.username)
+
+        path.write_text(profile.to_text(), encoding="utf-8")
+        print(
+            f"[DEBUG] Saved profile for user '{profile.user.username}' at '{path}'"
+        )
+        return path
+
+    def clear_profile(self, username: str) -> None:
+        path = self._get_profile_path(username)
+        path.write_text("", encoding="utf-8")
+        print(f"[INFO] Cleared profile for user '{username}'")
+
+    def delete_profile(self, username: str) -> None:
+        path = self._get_profile_path(username)
+
+        path.unlink()
+        print(f"[INFO] Deleted profile file for user '{username}'")
+
+        path.parent.rmdir()
+        print(f"[INFO] Deleted profile folder for user '{username}'")
+
+        self.xrmc.delete_profile_folder(username)
+        print(f"[INFO] Removed folder mapping for user '{username}'")

src/xray_manager/core/user.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+from .connection import ConnectionFactory, ConnectionKey
+
+if TYPE_CHECKING:
+    from .connection import Connection
+    from .xray_config import XrayConfig
+
+
+@dataclass
+class User:
+    username: str
+    connections: list[Connection] = field(default_factory=list)
+
+    def modify_connection_by_key(
+        self,
+        xray_config: "XrayConfig",
+        add: list[ConnectionKey] = [],
+        delete: list[ConnectionKey] = [],
+    ):
+        key_map = {
+            ConnectionKey(c.protocol, c.exit_point): c for c in self.connections
+        }
+
+        for key in delete:
+            key_map.pop(key, None)
+
+        for key in add:
+            if key not in key_map:
+                connection = ConnectionFactory.from_spec(key, xray_config)
+                key_map[key] = connection
+
+        self.connections = list(key_map.values())
+
+
+class UserFactory:
+    @staticmethod
+    def from_spec(
+        username: str, keys: list[ConnectionKey], xray_config: "XrayConfig"
+    ) -> User:
+        connections = []
+
+        for key in keys:
+            conn = ConnectionFactory.from_spec(key, xray_config)
+            connections.append(conn)
+
+        user = User(username=username, connections=connections)
+        return user
+
+    @staticmethod
+    def empty(username: str):
+        return User(username, [])

src/xray_manager/core/xray_config.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import json
+from collections import defaultdict
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from .connection import ConnectionFactory
+from .user import User
+
+if TYPE_CHECKING:
+    from .connection import Connection
+
+
+class XrayConfig:
+    def __init__(self, config_folder: Path, host: str):
+        self.path = config_folder
+        self.host = host
+
+        self.inbounds_files = list(self.path.glob("*-in-*.json"))
+        self.outbounds_files = list(self.path.glob("*-out-*.json"))
+
+        self.inbounds_data = {
+            f.name: self._load_json(f) for f in self.inbounds_files
+        }
+
+        self.outbounds_data = {
+            f.name: self._load_json(f) for f in self.outbounds_files
+        }
+
+    def _load_json(self, path: Path) -> dict:
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    def _save_json(self, data: dict, path: Path):
+        path.write_text(
+            json.dumps(data, indent=4, ensure_ascii=False) + "\n",
+            encoding="utf-8",
+        )
+
+    def _find_inbound(self, connection: Connection) -> tuple[str, dict]:
+        for filename, data in self.inbounds_data.items():
+            inbounds = data.get("inbounds", [])
+            for inbound in inbounds:
+                if connection.matches_inbound(inbound):
+                    return filename, inbound
+
+        raise RuntimeError(f"Inbound not found for {connection}")
+
+    def find_managed_inbound_by_protocol(self, protocol: str) -> dict:
+        for data in self.inbounds_data.values():
+            inbounds = data.get("inbounds", [])
+            for inbound in inbounds:
+                if not inbound.get("xrm"):
+                    continue
+
+                if inbound.get("protocol") != protocol:
+                    continue
+
+                return inbound
+
+        raise RuntimeError(
+            f"Managed inbound not found for protocol: {protocol}"
+        )
+
+    # Устаревший метод. Перейти на использование xray_config.get_targets_list()
+    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)
+
+        for data in self.inbounds_data.values():
+            inbounds = data.get("inbounds", [])
+            for inbound in inbounds:
+                protocol = inbound.get("protocol")
+
+                if protocol in ConnectionFactory._registry:
+                    settings = inbound.get("settings", {})
+                    clients = settings.get("clients", [])
+                    for client in clients:
+                        username, connection = ConnectionFactory.from_inbound(
+                            self.host, client, inbound
+                        )
+                        connection_map[username].append(connection)
+
+        users = []
+
+        for username, connections in connection_map.items():
+            users.append(User(username, connections))
+
+        return users
+
+    def get_user(self, username: str) -> User | None:
+        users = self.get_users()
+        for user in users:
+            if user.username == username:
+                return user
+
+    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)
+            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
+
+        deleted = False
+        modified_files = set()
+
+        for connection in user.connections:
+            try:
+                filename, inbound = self._find_inbound(connection)
+            except RuntimeError:
+                continue
+
+            connection.delete_from_inbound(inbound, username)
+            modified_files.add(filename)
+            deleted = True
+
+        for filename in modified_files:
+            path = self.path / filename
+            self._save_json(self.inbounds_data[filename], path)
+
+        return deleted
+
+    def modify_user(self, user: User) -> bool:
+        current_user = self.get_user(user.username)
+
+        if current_user is None:
+            return False
+
+        old_connections = set(current_user.connections)
+        new_connections = set(user.connections)
+
+        to_add = new_connections - old_connections
+        to_remove = old_connections - new_connections
+
+        modified_files = set()
+
+        for connection in to_remove:
+            filename, inbound = self._find_inbound(connection)
+            connection.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)
+            modified_files.add(filename)
+
+        for filename in modified_files:
+            path = self.path / filename
+            self._save_json(self.inbounds_data[filename], path)
+
+        return True

src/xray_manager/core/xray_manager_config.py

@@ -0,0 +1,89 @@
+import json
+from pathlib import Path
+
+
+class XrayManagerConfig:
+    def __init__(self, config_path: str):
+        self.path = Path(config_path)
+        self.config = json.loads(self.path.read_text(encoding="utf-8"))
+
+    def _save_json(self):
+        self.path.write_text(
+            json.dumps(self.config, indent=4, ensure_ascii=False) + "\n",
+            encoding="utf-8",
+        )
+
+    def _update(self, path: list[str], value):
+        ref = self.config
+        for key in path[:-1]:
+            ref = ref[key]
+
+        ref[path[-1]] = value
+
+        self._save_json()
+
+    @property
+    def host(self) -> str:
+        return self.config["host"]
+
+    @host.setter
+    def host(self, value: str):
+        self._update(["host"], value)
+
+    @property
+    def xray_config_folder(self) -> Path:
+        return Path(self.config["xray_config_folder"])
+
+    @xray_config_folder.setter
+    def xray_config_folder(self, value: str):
+        self._update(["xray_config_folder"], value)
+
+    @property
+    def profile_host(self) -> str:
+        return self.config["profiles"]["host"]
+
+    @profile_host.setter
+    def profile_host(self, value: str):
+        self._update(["profiles", "host"], value)
+
+    @property
+    def profile_base_path(self) -> Path:
+        return Path(self.config["profiles"]["base_path"])
+
+    @profile_base_path.setter
+    def profile_base_path(self, value: str):
+        self._update(["profiles", "base_path"], value)
+
+    @property
+    def profile_file_name(self) -> str:
+        return self.config["profiles"]["file_name"]
+
+    @profile_file_name.setter
+    def profile_file_name(self, value: str):
+        self._update(["profiles", "file_name"], value)
+
+    @property
+    def profile_folder_mapping(self) -> dict[str, str]:
+        return self.config["profiles"]["folder_mapping"]
+
+    @property
+    def folder_profile_mapping(self) -> dict[str, str]:
+        return {
+            folder_name: username
+            for username, folder_name in self.profile_folder_mapping.items()
+        }
+
+    @property
+    def targets(self) -> dict[str, dict[str, str]]:
+        return self.config.get("target", {})
+
+    def get_targets_list(self) -> list[str]:
+        return list(self.targets.keys())
+
+    def set_profile_folder(self, username: str, folder_name: str):
+        self.config["profiles"]["folder_mapping"][username] = folder_name
+        self._save_json()
+
+    def delete_profile_folder(self, username: str):
+        self.config["profiles"]["folder_mapping"].pop(username, None)
+        self._save_json()

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

@@ -5,6 +5,27 @@
         "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"
+        }
     }
 }

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)