diff --git a/src/xray_manager/core/__init__.py b/src/xray_manager/core/__init__.py new file mode 100644 index 0000000..91bf7e9 --- /dev/null +++ b/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 diff --git a/src/xray_manager/core.py b/src/xray_manager/core/connection.py similarity index 57% rename from src/xray_manager/core.py rename to src/xray_manager/core/connection.py index 4ee81d6..1a5efe1 100644 --- a/src/xray_manager/core.py +++ b/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 pathlib import Path -from typing import TypeVar +from dataclasses import dataclass +from typing import TYPE_CHECKING, TypeVar from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse -from .utils import private_to_public +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}'") diff --git a/src/xray_manager/core/profile.py b/src/xray_manager/core/profile.py new file mode 100644 index 0000000..b7582fc --- /dev/null +++ b/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}'") diff --git a/src/xray_manager/core/user.py b/src/xray_manager/core/user.py new file mode 100644 index 0000000..6fc9091 --- /dev/null +++ b/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, []) diff --git a/src/xray_manager/core/xray_config.py b/src/xray_manager/core/xray_config.py new file mode 100644 index 0000000..821104f --- /dev/null +++ b/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 diff --git a/src/xray_manager/core/xray_manager_config.py b/src/xray_manager/core/xray_manager_config.py new file mode 100644 index 0000000..87e45ad --- /dev/null +++ b/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()