Compare commits
4
Commits
v0.2
...
9da0596f6d
| Author | SHA1 | Date | |
|---|---|---|---|
 vladimir
|
9da0596f6d | ||
|
vladimir
|
e98b340393 | ||
|
vladimir
|
797ceb5c4b | ||
|
vladimir
|
274235b957 |
@@ -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>)
|
||||
```
|
||||
|
||||
@@ -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
|
||||
@@ -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}'")
|
||||
@@ -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
|
||||
@@ -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"))
|
||||
],
|
||||
}
|
||||
@@ -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}'")
|
||||
@@ -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, [])
|
||||
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user