chore: создание каталога sandbox для хранения проверочных скриптов

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

README.md

@@ -0,0 +1,35 @@
+# 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 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)
+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,14 @@
+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,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,650 @@
+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 ..utils import private_to_public
+from .outbound_fields import OUTBOUND_FIELDS
+
+if TYPE_CHECKING:
+    from .target import Target, TargetStorage
+    from .xray_config import XrayConfig
+    from .xray_manager_config import XrayManagerConfig
+
+
+@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_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):
+    PROTOCOL: str
+    LINK_SCHEME: str
+
+    @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_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 = email.rsplit("-", 1)
+        return username, target
+
+    @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, 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("_VlessFields", fields=OUTBOUND_FIELDS["vless"])
+
+
+@OutboundFactory.register
+class VlessOutbound(_VlessFields, Outbound):
+    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

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

@@ -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,97 @@
+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()
+        }
+
+    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()
+
+    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()

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)