docs: обновление списка команд

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,33 @@
+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
+
+
+class Outbound(ABC):
+    pass
+
+
+_ShadowsocksFields = make_dataclass(
+    "_ShadowsocksFields", fields=OUTBOUND_FIELDS["shadowsocks"]
+)
+
+
+class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
+    pass
+
+
+_VlessFields = make_dataclass(
+    "_VlessFieldss", fields=OUTBOUND_FIELDS["vless"]
+)
+
+
+class VlessOutbound(_VlessFields, Outbound):
+    pass

src/xray_manager/core/outbound_fields.py

@@ -0,0 +1,29 @@
+from dataclasses import field
+
+
+_BASE_FIELDS = [
+    ("host", str),
+    ("port", str),
+    ("target", 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,171 @@
+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}"
+        )
+
+    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,82 @@
+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()