Compare commits
14
Commits
v0.2
...
cb7f25918b
| Author | SHA1 | Date | |
|---|---|---|---|
 vladimir
|
cb7f25918b | ||
|
vladimir
|
3b2ae419b6 | ||
|
vladimir
|
a3b3fdf33b | ||
|
vladimir
|
6f19a59713 | ||
|
vladimir
|
4acaf817e4 | ||
|
vladimir
|
c3c9a1927e | ||
|
vladimir
|
0a3659c852 | ||
|
vladimir
|
d00081e632 | ||
|
vladimir
|
8063160e38 | ||
|
vladimir
|
89e53a4a48 | ||
|
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,385 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
import uuid
|
||||
from abc import ABC, abstractmethod
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, make_dataclass
|
||||
from typing import TYPE_CHECKING, TypeVar
|
||||
from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse
|
||||
|
||||
from .outbound_fields import OUTBOUND_FIELDS
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .xray_config import XrayConfig
|
||||
from .xray_manager_config import XrayManagerConfig
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OutboundSpec:
|
||||
protocol: str
|
||||
target: str
|
||||
|
||||
|
||||
O = TypeVar("O", bound="Outbound")
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
class OutboundFactory:
|
||||
_registry: dict[str, type[Outbound]] = {}
|
||||
_scheme_registry: dict[str, type[Outbound]] = {}
|
||||
|
||||
@classmethod
|
||||
def register(cls, outbound_cls: type[O]) -> type[O]:
|
||||
protocol = outbound_cls.PROTOCOL
|
||||
scheme = outbound_cls.LINK_SCHEME
|
||||
|
||||
if protocol in cls._registry:
|
||||
raise RuntimeError(f"Protocol already registered: {protocol}")
|
||||
|
||||
if scheme in cls._scheme_registry:
|
||||
raise RuntimeError(f"Scheme already registered: {scheme}")
|
||||
|
||||
cls._registry[protocol] = outbound_cls
|
||||
cls._scheme_registry[scheme] = outbound_cls
|
||||
|
||||
return outbound_cls
|
||||
|
||||
@classmethod
|
||||
def get_protocols(cls) -> list[str]:
|
||||
return list(cls._registry)
|
||||
|
||||
@classmethod
|
||||
def from_link(
|
||||
cls, xray_manager_config: XrayManagerConfig, link: str
|
||||
) -> Outbound:
|
||||
scheme = link.split("://", 1)[0]
|
||||
|
||||
if scheme not in cls._scheme_registry:
|
||||
raise ValueError(f"Unsupported link scheme: {scheme}")
|
||||
|
||||
outbound_cls = cls._scheme_registry[scheme]
|
||||
targets = xray_manager_config.targets
|
||||
return outbound_cls.from_link(targets, link)
|
||||
|
||||
@classmethod
|
||||
def from_inbound(
|
||||
cls,
|
||||
xray_manager_config: XrayManagerConfig,
|
||||
host: str,
|
||||
client: dict,
|
||||
inbound: dict,
|
||||
):
|
||||
protocol = inbound.get("protocol")
|
||||
|
||||
if protocol not in cls._registry:
|
||||
raise ValueError(f"Unsupported protocol: {protocol}")
|
||||
|
||||
outbound_cls = cls._registry[protocol]
|
||||
targets = xray_manager_config.targets
|
||||
username, outbound = outbound_cls.from_inbound(
|
||||
targets, host, client, inbound
|
||||
)
|
||||
return username, outbound
|
||||
|
||||
@classmethod
|
||||
def from_spec(
|
||||
cls,
|
||||
xray_config: XrayConfig,
|
||||
xray_manager_config: XrayManagerConfig,
|
||||
spec: OutboundSpec,
|
||||
) -> Outbound:
|
||||
if spec.protocol not in cls._registry:
|
||||
raise ValueError(f"Unsupported protocol: {spec.protocol}")
|
||||
|
||||
if spec.target not in xray_manager_config.get_targets_list():
|
||||
raise ValueError(f"Target {spec.target} not found")
|
||||
|
||||
inbound = xray_config.find_managed_inbound_by_protocol(spec.protocol)
|
||||
outbound_cls = cls._registry[spec.protocol]
|
||||
|
||||
targets = xray_manager_config.targets
|
||||
target_pretty_name = targets.get(spec.target, {}).get(
|
||||
"pretty_name", spec.target
|
||||
)
|
||||
|
||||
return outbound_cls.from_scratch(
|
||||
spec.target, target_pretty_name, xray_config.host, inbound
|
||||
)
|
||||
|
||||
|
||||
class Outbound(ABC):
|
||||
PROTOCOL: str
|
||||
LINK_SCHEME: str
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def from_link(
|
||||
cls, targets: dict[str, dict[str, str]], link: str
|
||||
) -> Outbound: ...
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def from_inbound(
|
||||
cls,
|
||||
targets: dict[str, dict[str, str]],
|
||||
host: str,
|
||||
client: dict,
|
||||
inbound: dict,
|
||||
) -> tuple[str, Outbound]: ...
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def from_scratch(
|
||||
cls, target: str, target_pretty_name: str, host: str, inbound: dict
|
||||
) -> Outbound: ...
|
||||
|
||||
@abstractmethod
|
||||
def to_link(self) -> str: ...
|
||||
|
||||
@abstractmethod
|
||||
def matches_inbound(self, inbound: dict) -> bool: ...
|
||||
|
||||
@abstractmethod
|
||||
def add_to_inbound(self, inbound: dict, username: str) -> None: ...
|
||||
|
||||
@abstractmethod
|
||||
def delete_from_inbound(self, inbound: dict, username: str) -> None: ...
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def protocol(self) -> str: ...
|
||||
|
||||
@staticmethod
|
||||
def split_client_email(email: str) -> tuple[str, str]:
|
||||
username, target = email.rsplit("-", 1)
|
||||
return username, target
|
||||
|
||||
@staticmethod
|
||||
def find_target_by_pretty_name(
|
||||
targets: dict[str, dict[str, str]], pretty_name: str
|
||||
) -> str | None:
|
||||
search_name = pretty_name.strip()
|
||||
for target, target_data in targets.items():
|
||||
if target_data.get("pretty_name", "").strip() == search_name:
|
||||
return target
|
||||
return None
|
||||
|
||||
@abstractmethod
|
||||
def __hash__(self) -> int: ...
|
||||
|
||||
@abstractmethod
|
||||
def __eq__(self, other: object) -> bool: ...
|
||||
|
||||
|
||||
_ShadowsocksFields = make_dataclass(
|
||||
"_ShadowsocksFields", fields=OUTBOUND_FIELDS["shadowsocks"]
|
||||
)
|
||||
|
||||
|
||||
@OutboundFactory.register
|
||||
class ShadowsocksOutbound(_ShadowsocksFields, Outbound):
|
||||
PROTOCOL = "shadowsocks"
|
||||
LINK_SCHEME = "ss"
|
||||
|
||||
@classmethod
|
||||
def from_link(
|
||||
cls, targets: dict[str, dict[str, str]], link: str
|
||||
) -> ShadowsocksOutbound:
|
||||
import base64
|
||||
|
||||
prefix_b64, rest = link[5:].split("@", 1)
|
||||
host_port, quoted_target_pretty_name = rest.split("#", 1)
|
||||
host, port_str = host_port.split(":")
|
||||
port = int(port_str)
|
||||
|
||||
prefix_bytes = base64.urlsafe_b64decode(prefix_b64)
|
||||
prefix_str = prefix_bytes.decode()
|
||||
method, server_password, client_password = prefix_str.split(":")
|
||||
|
||||
target_pretty_name = unquote(quoted_target_pretty_name)
|
||||
target = cls.find_target_by_pretty_name(targets, target_pretty_name)
|
||||
if target is None:
|
||||
target = "unknown"
|
||||
|
||||
return cls(
|
||||
host=host,
|
||||
port=port,
|
||||
target=target,
|
||||
target_pretty_name=target_pretty_name,
|
||||
method=method,
|
||||
server_password=server_password,
|
||||
client_password=client_password,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_inbound(
|
||||
cls,
|
||||
targets: dict[str, dict[str, str]],
|
||||
host: str,
|
||||
client: dict,
|
||||
inbound: dict,
|
||||
) -> tuple[str, ShadowsocksOutbound]:
|
||||
port = inbound["port"]
|
||||
method = inbound["settings"]["method"]
|
||||
server_password = inbound["settings"]["password"]
|
||||
client_password = client["password"]
|
||||
email = client["email"]
|
||||
username, target = cls.split_client_email(email)
|
||||
|
||||
target_pretty_name = targets.get(target, {}).get("pretty_name", target)
|
||||
|
||||
outbound = cls(
|
||||
host=host,
|
||||
port=port,
|
||||
target=target,
|
||||
target_pretty_name=target_pretty_name,
|
||||
method=method,
|
||||
server_password=server_password,
|
||||
client_password=client_password,
|
||||
)
|
||||
return username, outbound
|
||||
|
||||
@classmethod
|
||||
def from_scratch(
|
||||
cls, target: str, target_pretty_name: str, host: str, inbound: dict
|
||||
) -> ShadowsocksOutbound:
|
||||
port = inbound["port"]
|
||||
method = inbound["settings"]["method"]
|
||||
server_password = inbound["settings"]["password"]
|
||||
client_password = cls.generate_password(method)
|
||||
|
||||
return cls(
|
||||
host=host,
|
||||
port=port,
|
||||
target=target,
|
||||
target_pretty_name=target_pretty_name,
|
||||
method=method,
|
||||
server_password=server_password,
|
||||
client_password=client_password,
|
||||
)
|
||||
|
||||
def to_link(self) -> str:
|
||||
import base64
|
||||
|
||||
prefix = f"{self.method}:{self.server_password}:{self.client_password}"
|
||||
prefix_b64 = base64.urlsafe_b64encode(prefix.encode()).decode()
|
||||
tag = quote(self.target_pretty_name)
|
||||
link = f"ss://{prefix_b64}@{self.host}:{self.port}#{tag}"
|
||||
return link
|
||||
|
||||
@staticmethod
|
||||
def generate_password(method: str) -> str:
|
||||
import base64
|
||||
import secrets
|
||||
|
||||
key_lengths = {
|
||||
"2022-blake3-aes-128-gcm": 16,
|
||||
"2022-blake3-aes-256-gcm": 32,
|
||||
"2022-blake3-chacha20-poly1305": 32,
|
||||
}
|
||||
|
||||
try:
|
||||
length = key_lengths[method]
|
||||
except KeyError:
|
||||
raise ValueError(f"Unsupported shadowsocks method: {method}")
|
||||
|
||||
key = secrets.token_bytes(length)
|
||||
|
||||
return base64.b64encode(key).decode()
|
||||
|
||||
def matches_inbound(self, inbound: dict):
|
||||
if inbound.get("protocol") != self.PROTOCOL:
|
||||
return False
|
||||
|
||||
if inbound.get("port") != self.port:
|
||||
return False
|
||||
|
||||
settings = inbound.get("settings", {})
|
||||
|
||||
if settings.get("method") != self.method:
|
||||
return False
|
||||
|
||||
if settings.get("password") != self.server_password:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def add_to_inbound(self, inbound: dict, username: str) -> None:
|
||||
client = {
|
||||
"email": f"{username}-{self.target}",
|
||||
"password": self.client_password,
|
||||
}
|
||||
|
||||
clients = inbound["settings"].setdefault("clients", [])
|
||||
clients.append(client)
|
||||
|
||||
clients_by_user = defaultdict(list)
|
||||
for client in clients:
|
||||
username, _ = self.split_client_email(client["email"])
|
||||
clients_by_user[username].append(client)
|
||||
|
||||
priority = {"default": 0, "managed": 1, "relay": 2}
|
||||
|
||||
def sort_key(c):
|
||||
_, target = self.split_client_email(c["email"])
|
||||
return (priority.get(target, 100), c["email"])
|
||||
|
||||
for user_clients in clients_by_user.values():
|
||||
user_clients.sort(key=sort_key)
|
||||
|
||||
sorted_clients = []
|
||||
for username in sorted(clients_by_user):
|
||||
sorted_clients.extend(clients_by_user[username])
|
||||
|
||||
inbound["settings"]["clients"] = sorted_clients
|
||||
|
||||
def delete_from_inbound(self, inbound: dict, username: str) -> None:
|
||||
target_email = f"{username}-{self.target}"
|
||||
settings = inbound["settings"]
|
||||
clients = settings["clients"]
|
||||
|
||||
filtered_clients = []
|
||||
for client in clients:
|
||||
if client["email"] != target_email:
|
||||
filtered_clients.append(client)
|
||||
|
||||
inbound["settings"]["clients"] = filtered_clients
|
||||
|
||||
@property
|
||||
def protocol(self) -> str:
|
||||
return self.PROTOCOL
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(
|
||||
(
|
||||
self.host,
|
||||
self.port,
|
||||
self.target,
|
||||
self.method,
|
||||
self.server_password,
|
||||
self.client_password,
|
||||
)
|
||||
)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, ShadowsocksOutbound):
|
||||
return NotImplemented
|
||||
|
||||
return (
|
||||
self.host == other.host
|
||||
and self.port == other.port
|
||||
and self.target == other.target
|
||||
and self.method == other.method
|
||||
and self.server_password == other.server_password
|
||||
and self.client_password == other.client_password
|
||||
)
|
||||
|
||||
|
||||
_VlessFields = make_dataclass("_VlessFieldss", fields=OUTBOUND_FIELDS["vless"])
|
||||
|
||||
|
||||
@OutboundFactory.register
|
||||
class VlessOutbound(_VlessFields, Outbound):
|
||||
PROTOCOL = "vless"
|
||||
LINK_SCHEME = "vless"
|
||||
@@ -0,0 +1,31 @@
|
||||
from dataclasses import field
|
||||
|
||||
_BASE_FIELDS = [
|
||||
("host", str),
|
||||
("port", str),
|
||||
("target", str),
|
||||
("target_pretty_name", str),
|
||||
]
|
||||
|
||||
|
||||
OUTBOUND_FIELDS = {
|
||||
"shadowsocks": _BASE_FIELDS
|
||||
+ [
|
||||
("method", str),
|
||||
("server_password", str),
|
||||
("client_password", str),
|
||||
],
|
||||
"vless": _BASE_FIELDS
|
||||
+ [
|
||||
("id", str),
|
||||
("security", str),
|
||||
("encryption", str),
|
||||
("public_key", str),
|
||||
("network_type", str),
|
||||
("sni", str),
|
||||
("short_id", str),
|
||||
("flow", str, field(default="xtls-rprx-vision")),
|
||||
("header_type", str, field(default="none")),
|
||||
("finger_print", str, field(default="chrome")),
|
||||
],
|
||||
}
|
||||
@@ -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,61 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .xray_manager_config import XrayManagerConfig
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Target:
|
||||
id: str
|
||||
pretty: str
|
||||
route: str
|
||||
|
||||
|
||||
class TargetStorage:
|
||||
def __init__(self, xray_manager_config: XrayManagerConfig):
|
||||
self.xrmc = xray_manager_config
|
||||
|
||||
@property
|
||||
def target_dict(self) -> dict[str, dict[str, str]]:
|
||||
return self.xrmc.config.get("target", {})
|
||||
|
||||
def load_by_id(self, target_id: str) -> Target:
|
||||
if target_id not in self.target_dict:
|
||||
raise ValueError(f"Target not found: {target_id}")
|
||||
|
||||
target_data = self.target_dict[target_id]
|
||||
|
||||
if "route" not in target_data:
|
||||
raise ValueError(
|
||||
f"Missing required 'route' field for target: {target_id}"
|
||||
)
|
||||
|
||||
return Target(
|
||||
id=target_id,
|
||||
pretty=target_data.get("pretty_name", target_id),
|
||||
route=target_data["route"]
|
||||
)
|
||||
|
||||
def load_by_pretty(self, target_pretty_name: str) -> Target:
|
||||
for target_id, target_data in self.target_dict.items():
|
||||
if target_data.get("pretty_name", target_id) == target_pretty_name:
|
||||
return self.load_by_id(target_id)
|
||||
|
||||
raise ValueError(f"Target pretty name not found: {target_pretty_name}")
|
||||
|
||||
def load_all_targets(self) -> list[Target]:
|
||||
targets = [
|
||||
self.load_by_id(target_id) for target_id in self.target_dict.keys()
|
||||
]
|
||||
return targets
|
||||
|
||||
def save_target(self, target: Target):
|
||||
self.xrmc.set_target(target.id, target.pretty, target.route)
|
||||
|
||||
def delete_target(self, target: Target):
|
||||
if target not in self.load_all_targets():
|
||||
raise ValueError(f"Target not found: {target.id}")
|
||||
|
||||
self.xrmc.delete_target(target.id)
|
||||
@@ -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,172 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .connection import ConnectionFactory
|
||||
from .user import User
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .connection import Connection
|
||||
|
||||
|
||||
class XrayConfig:
|
||||
def __init__(self, config_folder: Path, host: str):
|
||||
self.path = config_folder
|
||||
self.host = host
|
||||
|
||||
self.inbounds_files = list(self.path.glob("*-in-*.json"))
|
||||
self.outbounds_files = list(self.path.glob("*-out-*.json"))
|
||||
|
||||
self.inbounds_data = {
|
||||
f.name: self._load_json(f) for f in self.inbounds_files
|
||||
}
|
||||
|
||||
self.outbounds_data = {
|
||||
f.name: self._load_json(f) for f in self.outbounds_files
|
||||
}
|
||||
|
||||
def _load_json(self, path: Path) -> dict:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
def _save_json(self, data: dict, path: Path):
|
||||
path.write_text(
|
||||
json.dumps(data, indent=4, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def _find_inbound(self, connection: Connection) -> tuple[str, dict]:
|
||||
for filename, data in self.inbounds_data.items():
|
||||
inbounds = data.get("inbounds", [])
|
||||
for inbound in inbounds:
|
||||
if connection.matches_inbound(inbound):
|
||||
return filename, inbound
|
||||
|
||||
raise RuntimeError(f"Inbound not found for {connection}")
|
||||
|
||||
def find_managed_inbound_by_protocol(self, protocol: str) -> dict:
|
||||
for data in self.inbounds_data.values():
|
||||
inbounds = data.get("inbounds", [])
|
||||
for inbound in inbounds:
|
||||
if not inbound.get("xrm"):
|
||||
continue
|
||||
|
||||
if inbound.get("protocol") != protocol:
|
||||
continue
|
||||
|
||||
return inbound
|
||||
|
||||
raise RuntimeError(
|
||||
f"Managed inbound not found for protocol: {protocol}"
|
||||
)
|
||||
|
||||
# Устаревший метод. Перейти на использование xray_config.get_targets_list()
|
||||
def get_exit_points(self) -> set[str]:
|
||||
exit_points = set()
|
||||
for data in self.outbounds_data.values():
|
||||
outbounds = data.get("outbounds", [])
|
||||
for outbound in outbounds:
|
||||
tag: str = outbound.get("tag")
|
||||
if tag:
|
||||
exit_point = tag.rsplit("-", 1)[1]
|
||||
exit_points.add(exit_point)
|
||||
exit_points.add("default")
|
||||
return exit_points
|
||||
|
||||
def get_users(self) -> list[User]:
|
||||
connection_map: defaultdict[str, list[Connection]] = defaultdict(list)
|
||||
|
||||
for data in self.inbounds_data.values():
|
||||
inbounds = data.get("inbounds", [])
|
||||
for inbound in inbounds:
|
||||
protocol = inbound.get("protocol")
|
||||
|
||||
if protocol in ConnectionFactory._registry:
|
||||
settings = inbound.get("settings", {})
|
||||
clients = settings.get("clients", [])
|
||||
for client in clients:
|
||||
username, connection = ConnectionFactory.from_inbound(
|
||||
self.host, client, inbound
|
||||
)
|
||||
connection_map[username].append(connection)
|
||||
|
||||
users = []
|
||||
|
||||
for username, connections in connection_map.items():
|
||||
users.append(User(username, connections))
|
||||
|
||||
return users
|
||||
|
||||
def get_user(self, username: str) -> User | None:
|
||||
users = self.get_users()
|
||||
for user in users:
|
||||
if user.username == username:
|
||||
return user
|
||||
|
||||
def add_user(self, user: User) -> None:
|
||||
modified_files = set()
|
||||
|
||||
for connection in user.connections:
|
||||
filename, inbound = self._find_inbound(connection)
|
||||
connection.add_to_inbound(inbound, user.username)
|
||||
modified_files.add(filename)
|
||||
|
||||
for filename in modified_files:
|
||||
path = self.path / filename
|
||||
self._save_json(self.inbounds_data[filename], path)
|
||||
|
||||
def delete_user(self, username: str) -> bool:
|
||||
user = self.get_user(username)
|
||||
if not user:
|
||||
return False
|
||||
|
||||
deleted = False
|
||||
modified_files = set()
|
||||
|
||||
for connection in user.connections:
|
||||
try:
|
||||
filename, inbound = self._find_inbound(connection)
|
||||
except RuntimeError:
|
||||
continue
|
||||
|
||||
connection.delete_from_inbound(inbound, username)
|
||||
modified_files.add(filename)
|
||||
deleted = True
|
||||
|
||||
for filename in modified_files:
|
||||
path = self.path / filename
|
||||
self._save_json(self.inbounds_data[filename], path)
|
||||
|
||||
return deleted
|
||||
|
||||
def modify_user(self, user: User) -> bool:
|
||||
current_user = self.get_user(user.username)
|
||||
|
||||
if current_user is None:
|
||||
return False
|
||||
|
||||
old_connections = set(current_user.connections)
|
||||
new_connections = set(user.connections)
|
||||
|
||||
to_add = new_connections - old_connections
|
||||
to_remove = old_connections - new_connections
|
||||
|
||||
modified_files = set()
|
||||
|
||||
for connection in to_remove:
|
||||
filename, inbound = self._find_inbound(connection)
|
||||
connection.delete_from_inbound(inbound, user.username)
|
||||
modified_files.add(filename)
|
||||
|
||||
for connection in to_add:
|
||||
filename, inbound = self._find_inbound(connection)
|
||||
connection.add_to_inbound(inbound, user.username)
|
||||
modified_files.add(filename)
|
||||
|
||||
for filename in modified_files:
|
||||
path = self.path / filename
|
||||
self._save_json(self.inbounds_data[filename], path)
|
||||
|
||||
return True
|
||||
@@ -0,0 +1,104 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class XrayManagerConfig:
|
||||
def __init__(self, config_path: str):
|
||||
self.path = Path(config_path)
|
||||
self.config = json.loads(self.path.read_text(encoding="utf-8"))
|
||||
|
||||
def _save_json(self):
|
||||
self.path.write_text(
|
||||
json.dumps(self.config, indent=4, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def _update(self, path: list[str], value):
|
||||
ref = self.config
|
||||
for key in path[:-1]:
|
||||
ref = ref[key]
|
||||
|
||||
ref[path[-1]] = value
|
||||
|
||||
self._save_json()
|
||||
|
||||
@property
|
||||
def host(self) -> str:
|
||||
return self.config["host"]
|
||||
|
||||
@host.setter
|
||||
def host(self, value: str):
|
||||
self._update(["host"], value)
|
||||
|
||||
@property
|
||||
def xray_config_folder(self) -> Path:
|
||||
return Path(self.config["xray_config_folder"])
|
||||
|
||||
@xray_config_folder.setter
|
||||
def xray_config_folder(self, value: str):
|
||||
self._update(["xray_config_folder"], value)
|
||||
|
||||
@property
|
||||
def profile_host(self) -> str:
|
||||
return self.config["profiles"]["host"]
|
||||
|
||||
@profile_host.setter
|
||||
def profile_host(self, value: str):
|
||||
self._update(["profiles", "host"], value)
|
||||
|
||||
@property
|
||||
def profile_base_path(self) -> Path:
|
||||
return Path(self.config["profiles"]["base_path"])
|
||||
|
||||
@profile_base_path.setter
|
||||
def profile_base_path(self, value: str):
|
||||
self._update(["profiles", "base_path"], value)
|
||||
|
||||
@property
|
||||
def profile_file_name(self) -> str:
|
||||
return self.config["profiles"]["file_name"]
|
||||
|
||||
@profile_file_name.setter
|
||||
def profile_file_name(self, value: str):
|
||||
self._update(["profiles", "file_name"], value)
|
||||
|
||||
@property
|
||||
def profile_folder_mapping(self) -> dict[str, str]:
|
||||
return self.config["profiles"]["folder_mapping"]
|
||||
|
||||
@property
|
||||
def folder_profile_mapping(self) -> dict[str, str]:
|
||||
return {
|
||||
folder_name: username
|
||||
for username, folder_name in self.profile_folder_mapping.items()
|
||||
}
|
||||
|
||||
@property
|
||||
def targets(self) -> dict[str, dict[str, str]]:
|
||||
return self.config.get("target", {})
|
||||
|
||||
def get_targets_list(self) -> list[str]:
|
||||
return list(self.targets.keys())
|
||||
|
||||
def set_profile_folder(self, username: str, folder_name: str):
|
||||
self.config["profiles"]["folder_mapping"][username] = folder_name
|
||||
self._save_json()
|
||||
|
||||
def delete_profile_folder(self, username: str):
|
||||
self.config["profiles"]["folder_mapping"].pop(username, None)
|
||||
self._save_json()
|
||||
|
||||
def set_target(
|
||||
self, target_id: str, target_pretty_name: str, target_route: str
|
||||
):
|
||||
self.config["target"][target_id] = {
|
||||
"pretty_name": target_pretty_name,
|
||||
"route": target_route
|
||||
}
|
||||
self.config["target"][target_id]["pretty_name"] = target_pretty_name
|
||||
self.config["target"][target_id]["route"] = target_route
|
||||
self._save_json()
|
||||
|
||||
def delete_target(self, target_id: str):
|
||||
self.config["target"].pop(target_id, None)
|
||||
self._save_json()
|
||||
@@ -5,6 +5,27 @@
|
||||
"host": "localhost.internal",
|
||||
"base_path": "tests/mock_data/var/lib/xray-manager/profiles",
|
||||
"file_name": "list.txt",
|
||||
"folder_mapping": {}
|
||||
"folder_mapping": {
|
||||
"test": "OiVqwjq0V9kI",
|
||||
"test2": "P8TC4FUicrvC"
|
||||
}
|
||||
},
|
||||
"target": {
|
||||
"default": {
|
||||
"route": "out-fr1",
|
||||
"pretty_name": "🚩 Default"
|
||||
},
|
||||
"fr1": {
|
||||
"route": "out-fr1",
|
||||
"pretty_name": "🇫🇷 France #1"
|
||||
},
|
||||
"us1": {
|
||||
"route": "out-us1",
|
||||
"pretty_name": "🇺🇸 USA #1"
|
||||
},
|
||||
"us2": {
|
||||
"route": "out-us2",
|
||||
"pretty_name": "🇺🇸 USA #2"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
from xray_manager.core.outbound import (
|
||||
OutboundFactory,
|
||||
OutboundSpec,
|
||||
ShadowsocksOutbound,
|
||||
)
|
||||
from xray_manager.core.xray_config import XrayConfig
|
||||
from xray_manager.core.xray_manager_config import XrayManagerConfig
|
||||
|
||||
inbound = {
|
||||
"tag": "shadowsocks",
|
||||
"xrm": True,
|
||||
"listen": "0.0.0.0",
|
||||
"port": 8443,
|
||||
"protocol": "shadowsocks",
|
||||
"settings": {
|
||||
"method": "2022-blake3-aes-256-gcm",
|
||||
"password": "Z8WtMeWzZVh1F6/5URGrs8vWdB3CLN5y7A9D1U0Q65E=",
|
||||
"clients": [
|
||||
{
|
||||
"email": "test-fr1",
|
||||
"password": "zV3qwuj5TUng+mBMfqdInp0ih7i9ykSIzUig3v5d1Bg=",
|
||||
},
|
||||
{
|
||||
"email": "test2-us1",
|
||||
"password": "ZNtMDJ5WuaGa2ko2jMbbZguddcMY2TTirsb9cNYHffw=",
|
||||
},
|
||||
],
|
||||
"network": "tcp,udp",
|
||||
},
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
|
||||
xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
|
||||
|
||||
target = "us1"
|
||||
target_pretty_name = xray_manager_config.targets.get(target, {}).get(
|
||||
"pretty_name", target
|
||||
)
|
||||
|
||||
outbound = ShadowsocksOutbound.from_scratch(
|
||||
target, target_pretty_name, "127.0.0.1", inbound
|
||||
)
|
||||
|
||||
link = outbound.to_link()
|
||||
o = ShadowsocksOutbound.from_link(xray_manager_config.targets, link)
|
||||
print(o)
|
||||
@@ -0,0 +1,13 @@
|
||||
from xray_manager.core.target import Target, TargetStorage
|
||||
from xray_manager.core.xray_manager_config import XrayManagerConfig
|
||||
|
||||
DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
xray_manager_config = XrayManagerConfig(DEV_CONFIG_PATH)
|
||||
storage = TargetStorage(xray_manager_config)
|
||||
|
||||
target = Target("de1", "pretty", "out-de1")
|
||||
|
||||
storage.delete_target(target)
|
||||
Reference in New Issue
Block a user