Initial commit: xray-manager v0.2 with Debian 13 support
This commit is contained in:
@@ -0,0 +1,123 @@
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from . import handlers
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(prog="xray-manager")
|
||||
|
||||
subparsers = parser.add_subparsers(dest="group", required=True)
|
||||
|
||||
# ---------- user ----------
|
||||
user_parser = subparsers.add_parser("user")
|
||||
user_subparser = user_parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
# user add
|
||||
user_add_parser = user_subparser.add_parser("add")
|
||||
user_add_parser.add_argument("username")
|
||||
user_add_parser.add_argument(
|
||||
"-c",
|
||||
"--connection",
|
||||
nargs="+",
|
||||
default=[],
|
||||
help="Connections to add, e.g. shadowsocks:ru1 vless:us1",
|
||||
)
|
||||
user_add_parser.set_defaults(handler=handlers.user_add)
|
||||
|
||||
# user show
|
||||
user_show_parser = user_subparser.add_parser("show")
|
||||
user_show_parser.add_argument("username", nargs="?")
|
||||
user_show_parser.add_argument("-a", "--all", action="store_true")
|
||||
user_show_parser.set_defaults(handler=handlers.user_show)
|
||||
|
||||
# user modify
|
||||
user_modify_parser = user_subparser.add_parser("modify")
|
||||
user_modify_parser.add_argument("username")
|
||||
user_modify_parser.add_argument(
|
||||
"-a",
|
||||
"--add",
|
||||
dest="to_add",
|
||||
nargs="+",
|
||||
default=[],
|
||||
metavar="SPEC",
|
||||
help="Connections to add",
|
||||
)
|
||||
user_modify_parser.add_argument(
|
||||
"-d",
|
||||
"--delete",
|
||||
dest="to_delete",
|
||||
nargs="+",
|
||||
default=[],
|
||||
metavar="SPEC",
|
||||
help="Connections to delete",
|
||||
)
|
||||
user_modify_parser.set_defaults(handler=handlers.user_modify)
|
||||
|
||||
# user delete
|
||||
user_delete_parser = user_subparser.add_parser("delete")
|
||||
group = user_delete_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
user_delete_parser.set_defaults(handler=handlers.user_delete)
|
||||
|
||||
# -------- profile ---------
|
||||
profile_parser = subparsers.add_parser("profile")
|
||||
profile_subparser = profile_parser.add_subparsers(
|
||||
dest="command", required=True
|
||||
)
|
||||
|
||||
# profile build
|
||||
profile_build_parser = profile_subparser.add_parser("build")
|
||||
group = profile_build_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
profile_build_parser.set_defaults(handler=handlers.profile_build)
|
||||
|
||||
# profile show
|
||||
profile_show_parser = profile_subparser.add_parser("show")
|
||||
group = profile_show_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
profile_show_parser.set_defaults(handler=handlers.profile_show)
|
||||
|
||||
# profile clear
|
||||
profile_clear_parser = profile_subparser.add_parser("clear")
|
||||
group = profile_clear_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
profile_clear_parser.set_defaults(handler=handlers.profile_clear)
|
||||
|
||||
# profile delete
|
||||
profile_delete_parser = profile_subparser.add_parser("delete")
|
||||
group = profile_delete_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
profile_delete_parser.set_defaults(handler=handlers.profile_delete)
|
||||
|
||||
# profile qrcode
|
||||
profile_qrcode_parser = profile_subparser.add_parser("qrcode")
|
||||
group = profile_qrcode_parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("username", nargs="?")
|
||||
group.add_argument("-a", "--all", action="store_true")
|
||||
profile_qrcode_parser.set_defaults(handler=handlers.profile_qrcode)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main():
|
||||
parser = build_parser()
|
||||
|
||||
try:
|
||||
args = parser.parse_args()
|
||||
func = args.handler
|
||||
kwargs = vars(args)
|
||||
kwargs.pop("handler")
|
||||
func(**kwargs)
|
||||
except ValueError as e:
|
||||
print(e, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,202 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from .core import (
|
||||
ConnectionFactory,
|
||||
ConnectionKey,
|
||||
Profile,
|
||||
ProfileFactory,
|
||||
ProfileStorage,
|
||||
User,
|
||||
UserFactory,
|
||||
XrayConfig,
|
||||
XrayManagerConfig,
|
||||
)
|
||||
from .utils import print_qrcode, print_table
|
||||
|
||||
DEFAULT_CONFIG_PATH = "/etc/xray-manager/config.json"
|
||||
DEV_CONFIG_PATH = "tests/mock_data/etc/xray-manager/config.json"
|
||||
|
||||
|
||||
def _resolve_config_path() -> str:
|
||||
env_path = os.getenv("XRAY_MANAGER_CONFIG")
|
||||
if env_path:
|
||||
return env_path
|
||||
|
||||
if os.getenv("XRAY_ENV") == "development":
|
||||
return DEV_CONFIG_PATH
|
||||
|
||||
if os.path.exists(DEFAULT_CONFIG_PATH):
|
||||
return DEFAULT_CONFIG_PATH
|
||||
|
||||
return DEV_CONFIG_PATH
|
||||
|
||||
|
||||
xray_manager_config = XrayManagerConfig(_resolve_config_path())
|
||||
|
||||
xray_config = XrayConfig(
|
||||
xray_manager_config.xray_config_folder, xray_manager_config.host
|
||||
)
|
||||
|
||||
storage = ProfileStorage(xray_manager_config)
|
||||
|
||||
|
||||
def _resolve_users(username: str | None, all: bool) -> list[User]:
|
||||
if all:
|
||||
return xray_config.get_users()
|
||||
if username:
|
||||
user = xray_config.get_user(username)
|
||||
if not isinstance(user, User):
|
||||
raise ValueError(f"User not found: {username}")
|
||||
return [user]
|
||||
raise ValueError("Either username or --all must be specified")
|
||||
|
||||
|
||||
def _resolve_profiles(username: str | None, all: bool) -> list[Profile]:
|
||||
if all:
|
||||
return storage.load_profiles()
|
||||
if username:
|
||||
try:
|
||||
profile = storage.load_profile(username)
|
||||
except (ValueError, FileNotFoundError) as e:
|
||||
raise ValueError(f"Cannot load profile for {username}: {e}")
|
||||
return [profile]
|
||||
raise ValueError("Either username or --all must be specified")
|
||||
|
||||
|
||||
def _parse_connection_spec(spec: str) -> ConnectionKey:
|
||||
protocol, exit_point = spec.split(":", 1)
|
||||
if protocol not in ConnectionFactory.get_protocols():
|
||||
raise ValueError(f"Unsupported protocol: {protocol}")
|
||||
if exit_point not in xray_config.get_exit_points():
|
||||
raise ValueError(f"Unsupported exit point: {exit_point}")
|
||||
return ConnectionKey(protocol, exit_point)
|
||||
|
||||
|
||||
def _build_user_rows(users: list[User]) -> list[list[str]]:
|
||||
rows = []
|
||||
for index, user in enumerate(users, start=1):
|
||||
first = True
|
||||
for conn in user.connections:
|
||||
if first:
|
||||
row = [index, user.username, conn.exit_point, conn.protocol]
|
||||
first = False
|
||||
else:
|
||||
row = ["", "", conn.exit_point, conn.protocol]
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
def _build_profile_rows(profiles: list[Profile]) -> list[list[str]]:
|
||||
rows = []
|
||||
for index, profile in enumerate(profiles, start=1):
|
||||
first = True
|
||||
for conn in profile.user.connections:
|
||||
if first:
|
||||
row = [
|
||||
index,
|
||||
profile.user.username,
|
||||
profile.folder_name,
|
||||
conn.exit_point,
|
||||
conn.protocol,
|
||||
]
|
||||
first = False
|
||||
else:
|
||||
row = ["", "", "", conn.exit_point, conn.protocol]
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
def user_add(username: str, connection: list[str], **_):
|
||||
user = xray_config.get_user(username)
|
||||
|
||||
if isinstance(user, User):
|
||||
raise ValueError(f"User already exists: {username}")
|
||||
|
||||
keys: list[ConnectionKey] = [
|
||||
_parse_connection_spec(spec) for spec in connection
|
||||
]
|
||||
|
||||
user = UserFactory.from_spec(
|
||||
username=username, keys=keys, xray_config=xray_config
|
||||
)
|
||||
|
||||
xray_config.add_user(user)
|
||||
|
||||
|
||||
def user_show(username: str | None, all: bool, **_):
|
||||
users = _resolve_users(username, all)
|
||||
|
||||
headers = ["#", "Username", "Exit Point", "Protocol"]
|
||||
rows = _build_user_rows(users)
|
||||
|
||||
print_table(headers, rows)
|
||||
|
||||
|
||||
def user_modify(username: str, to_add: list[str], to_delete: list[str], **_):
|
||||
user = xray_config.get_user(username)
|
||||
|
||||
if not isinstance(user, User):
|
||||
raise ValueError(f"User not found: {username}")
|
||||
|
||||
add_key = [_parse_connection_spec(spec) for spec in to_add]
|
||||
del_key = [_parse_connection_spec(spec) for spec in to_delete]
|
||||
|
||||
user.modify_connection_by_key(xray_config, add_key, del_key)
|
||||
xray_config.modify_user(user)
|
||||
|
||||
|
||||
def user_delete(username: str | None, all: bool, **_):
|
||||
users = _resolve_users(username, all)
|
||||
|
||||
for user in users:
|
||||
xray_config.delete_user(user.username)
|
||||
|
||||
|
||||
def profile_build(username: str | None, all: bool, **_):
|
||||
users = _resolve_users(username, all)
|
||||
|
||||
for user in users:
|
||||
try:
|
||||
profile = storage.load_profile(user.username)
|
||||
profile = ProfileFactory.from_user(user, profile.folder_name)
|
||||
except (ValueError, FileNotFoundError) as e:
|
||||
print(f"Cannot load profile for {user.username}: {e}")
|
||||
profile = ProfileFactory.from_user(user)
|
||||
|
||||
storage.save_profile(profile)
|
||||
|
||||
|
||||
def profile_show(username: str | None, all: bool, **_):
|
||||
profiles = _resolve_profiles(username, all)
|
||||
|
||||
headers = ["#", "Username", "Profile", "Exit Point", "Protocol"]
|
||||
rows = _build_profile_rows(profiles)
|
||||
|
||||
print_table(headers, rows)
|
||||
|
||||
|
||||
def profile_clear(username: str | None, all: bool, **_):
|
||||
profiles = _resolve_profiles(username, all)
|
||||
|
||||
for profile in profiles:
|
||||
storage.clear_profile(profile.user.username)
|
||||
|
||||
|
||||
def profile_delete(username: str | None, all: bool, **_):
|
||||
profiles = _resolve_profiles(username, all)
|
||||
|
||||
for profile in profiles:
|
||||
storage.delete_profile(profile.user.username)
|
||||
|
||||
|
||||
def profile_qrcode(username: str | None, all: bool, **_):
|
||||
profiles = _resolve_profiles(username, all)
|
||||
|
||||
for profile in profiles:
|
||||
host = xray_manager_config.profile_host
|
||||
folder_name = profile.folder_name
|
||||
filename = xray_manager_config.profile_file_name
|
||||
|
||||
url = f"https://{host}/{folder_name}/{filename}"
|
||||
print_qrcode(url)
|
||||
@@ -0,0 +1,44 @@
|
||||
import base64
|
||||
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import x25519
|
||||
from qrcode import ERROR_CORRECT_L, QRCode
|
||||
from tabulate import tabulate
|
||||
|
||||
|
||||
def print_table(headers: list[str], rows: list[list[str]]) -> None:
|
||||
table = tabulate(rows, headers=headers, tablefmt="rounded_grid")
|
||||
print(table)
|
||||
|
||||
|
||||
def print_qrcode(url: str) -> None:
|
||||
qr = QRCode(version=1, error_correction=ERROR_CORRECT_L)
|
||||
qr.add_data(url)
|
||||
qr.make()
|
||||
|
||||
print(f"{url}\n")
|
||||
qr.print_ascii()
|
||||
|
||||
|
||||
def private_to_public(private_key_b64: str) -> str:
|
||||
def b64decode(s: str) -> bytes:
|
||||
padding = "=" * (-len(s) % 4)
|
||||
return base64.urlsafe_b64decode(s + padding)
|
||||
|
||||
def b64encode(b: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(b).decode().rstrip("=")
|
||||
|
||||
private_bytes = b64decode(private_key_b64)
|
||||
|
||||
if len(private_bytes) != 32:
|
||||
raise ValueError("Invalid private key length (expected 32 bytes)")
|
||||
|
||||
private_key = x25519.X25519PrivateKey.from_private_bytes(private_bytes)
|
||||
public_key = private_key.public_key()
|
||||
|
||||
public_bytes = public_key.public_bytes(
|
||||
encoding=serialization.Encoding.Raw,
|
||||
format=serialization.PublicFormat.Raw,
|
||||
)
|
||||
|
||||
return b64encode(public_bytes)
|
||||
Reference in New Issue
Block a user