644 lines
22 KiB
Python
644 lines
22 KiB
Python
"""Клиент API панели 3x-ui (v3.5.0+): login, inbounds, WireGuard clients."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import secrets
|
||
import string
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import quote, urljoin, urlparse
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 3x-ui 3.5.0: WireGuard — multi-client через settings.clients + addClient
|
||
SUPPORTED_PANEL = "3.5.0"
|
||
|
||
|
||
class XuiError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class XuiInbound:
|
||
id: int
|
||
remark: str
|
||
protocol: str
|
||
port: int
|
||
enable: bool
|
||
clients_count: int = 0
|
||
listen: str = ""
|
||
|
||
@property
|
||
def is_wireguard(self) -> bool:
|
||
return (self.protocol or "").lower() in {"wireguard", "wg"}
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class XuiConnectionInfo:
|
||
ok: bool
|
||
version: str | None
|
||
inbounds_total: int
|
||
wireguard: list[XuiInbound]
|
||
message: str = ""
|
||
|
||
|
||
def _normalize_base_url(url: str) -> str:
|
||
raw = (url or "").strip().rstrip("/")
|
||
if not raw:
|
||
raise XuiError("API URL пуст")
|
||
if not raw.startswith(("http://", "https://")):
|
||
raw = "https://" + raw
|
||
parsed = urlparse(raw)
|
||
if not parsed.netloc:
|
||
raise XuiError("Некорректный API URL")
|
||
return raw
|
||
|
||
|
||
def _new_sub_id(length: int = 16) -> str:
|
||
alphabet = string.ascii_lowercase + string.digits
|
||
return "".join(secrets.choice(alphabet) for _ in range(length))
|
||
|
||
|
||
def build_subscription_link(subscription_base: str, sub_id: str) -> str:
|
||
"""Склеивает ручной base URL подписки с subId клиента."""
|
||
base = (subscription_base or "").strip().rstrip("/")
|
||
sid = (sub_id or "").strip().lstrip("/")
|
||
if not base or not sid:
|
||
return ""
|
||
return f"{base}/{sid}"
|
||
|
||
|
||
def _parse_settings(raw: Any) -> dict[str, Any]:
|
||
if isinstance(raw, dict):
|
||
return raw
|
||
if isinstance(raw, str) and raw.strip():
|
||
try:
|
||
data = json.loads(raw)
|
||
return data if isinstance(data, dict) else {}
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
return {}
|
||
|
||
|
||
def _clients_count(settings: dict[str, Any]) -> int:
|
||
clients = settings.get("clients")
|
||
if isinstance(clients, list):
|
||
return len(clients)
|
||
peers = settings.get("peers")
|
||
if isinstance(peers, list):
|
||
return len(peers)
|
||
return 0
|
||
|
||
|
||
def _inbound_from_obj(obj: dict[str, Any]) -> XuiInbound:
|
||
settings = _parse_settings(obj.get("settings"))
|
||
return XuiInbound(
|
||
id=int(obj.get("id") or 0),
|
||
remark=str(obj.get("remark") or ""),
|
||
protocol=str(obj.get("protocol") or ""),
|
||
port=int(obj.get("port") or 0),
|
||
enable=bool(obj.get("enable", True)),
|
||
clients_count=_clients_count(settings),
|
||
listen=str(obj.get("listen") or ""),
|
||
)
|
||
|
||
|
||
class XuiClient:
|
||
"""Сессия к одной панели 3x-ui.
|
||
|
||
Предпочтительно: API Token (Settings → Security → API Token) как Bearer.
|
||
Fallback: cookie после POST /login (username + password).
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
api_url: str,
|
||
username: str = "",
|
||
password: str = "",
|
||
*,
|
||
api_key: str = "",
|
||
verify_ssl: bool = True,
|
||
timeout: float = 25.0,
|
||
) -> None:
|
||
self.base_url = _normalize_base_url(api_url)
|
||
self.username = (username or "").strip()
|
||
self.password = password or ""
|
||
self.api_key = (api_key or "").strip()
|
||
self.verify_ssl = verify_ssl
|
||
self.timeout = timeout
|
||
self._client: httpx.AsyncClient | None = None
|
||
self._authed = False
|
||
|
||
async def __aenter__(self) -> XuiClient:
|
||
await self.open()
|
||
return self
|
||
|
||
async def __aexit__(self, *args: Any) -> None:
|
||
await self.close()
|
||
|
||
async def open(self) -> None:
|
||
if self._client is not None:
|
||
return
|
||
headers: dict[str, str] = {"Accept": "application/json"}
|
||
if self.api_key:
|
||
headers["Authorization"] = f"Bearer {self.api_key}"
|
||
self._client = httpx.AsyncClient(
|
||
base_url=self.base_url,
|
||
timeout=self.timeout,
|
||
verify=self.verify_ssl,
|
||
follow_redirects=True,
|
||
headers=headers,
|
||
)
|
||
|
||
async def close(self) -> None:
|
||
if self._client is not None:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
self._authed = False
|
||
|
||
def _http(self) -> httpx.AsyncClient:
|
||
if self._client is None:
|
||
raise XuiError("Клиент не открыт — вызовите open() или используйте async with")
|
||
return self._client
|
||
|
||
async def _request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
json_body: dict[str, Any] | None = None,
|
||
data: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
http = self._http()
|
||
url = path if path.startswith("/") else f"/{path}"
|
||
try:
|
||
resp = await http.request(method, url, json=json_body, data=data)
|
||
except httpx.RequestError as exc:
|
||
raise XuiError(f"Сеть: {exc}") from exc
|
||
|
||
try:
|
||
payload = resp.json()
|
||
except json.JSONDecodeError:
|
||
text = (resp.text or "")[:200].replace("\n", " ")
|
||
hint = ""
|
||
if resp.status_code == 404:
|
||
hint = (
|
||
" (часто: неверный API Token — панель отвечает 404; "
|
||
"или устаревший путь API)"
|
||
)
|
||
raise XuiError(
|
||
f"Не JSON ответ ({resp.status_code}){hint}: {text}"
|
||
) from None
|
||
|
||
if not isinstance(payload, dict):
|
||
raise XuiError("Неожиданный формат ответа панели")
|
||
|
||
if resp.status_code >= 400:
|
||
msg = payload.get("msg") or payload.get("message") or resp.reason_phrase
|
||
raise XuiError(f"HTTP {resp.status_code}: {msg}")
|
||
|
||
# 3x-ui обычно: {success: bool, msg: str, obj: ...}
|
||
if "success" in payload and payload.get("success") is False:
|
||
raise XuiError(str(payload.get("msg") or "Ошибка панели"))
|
||
|
||
return payload
|
||
|
||
async def ensure_auth(self) -> None:
|
||
"""Авторизация: Bearer API key или login cookie."""
|
||
if self._authed:
|
||
return
|
||
await self.open()
|
||
if self.api_key:
|
||
# Bearer уже в headers — login не нужен (обходит CSRF на /panel/api/*)
|
||
self._authed = True
|
||
return
|
||
await self.login()
|
||
self._authed = True
|
||
|
||
async def login(self) -> None:
|
||
if not self.username or not self.password:
|
||
raise XuiError("Нужен API Token или пара логин/пароль")
|
||
# Классический form login; часть сборок принимает JSON
|
||
try:
|
||
await self._request(
|
||
"POST",
|
||
"/login",
|
||
data={"username": self.username, "password": self.password},
|
||
)
|
||
except XuiError:
|
||
await self._request(
|
||
"POST",
|
||
"/login",
|
||
json_body={"username": self.username, "password": self.password},
|
||
)
|
||
|
||
async def get_version(self) -> str | None:
|
||
for path in ("/panel/api/server/status", "/panel/api/server/getConfigJson"):
|
||
try:
|
||
payload = await self._request("GET", path)
|
||
except XuiError:
|
||
continue
|
||
obj = payload.get("obj")
|
||
if isinstance(obj, dict):
|
||
for key in ("version", "xrayVersion", "appVersion"):
|
||
val = obj.get(key)
|
||
if val:
|
||
return str(val)
|
||
# status иногда кладёт version на верхний уровень obj как строку в nested
|
||
if isinstance(obj, dict) and "cpu" in obj:
|
||
# нет версии в status — ок
|
||
return None
|
||
return None
|
||
|
||
async def list_inbounds(self) -> list[XuiInbound]:
|
||
payload = await self._request("GET", "/panel/api/inbounds/list")
|
||
obj = payload.get("obj")
|
||
items: list[Any]
|
||
if isinstance(obj, list):
|
||
items = obj
|
||
elif isinstance(obj, dict):
|
||
items = obj.get("inbounds") or obj.get("list") or []
|
||
else:
|
||
items = []
|
||
out: list[XuiInbound] = []
|
||
for item in items:
|
||
if isinstance(item, dict) and item.get("id") is not None:
|
||
out.append(_inbound_from_obj(item))
|
||
return out
|
||
|
||
async def get_inbound(self, inbound_id: int) -> dict[str, Any]:
|
||
payload = await self._request("GET", f"/panel/api/inbounds/get/{int(inbound_id)}")
|
||
obj = payload.get("obj")
|
||
if not isinstance(obj, dict):
|
||
raise XuiError(f"Инбаунд #{inbound_id} не найден")
|
||
return obj
|
||
|
||
async def list_wireguard_inbounds(self) -> list[XuiInbound]:
|
||
return [i for i in await self.list_inbounds() if i.is_wireguard]
|
||
|
||
async def add_wireguard_client(
|
||
self,
|
||
inbound_id: int,
|
||
*,
|
||
email: str,
|
||
sub_id: str | None = None,
|
||
tg_id: int | str = 0,
|
||
expiry_ms: int = 0,
|
||
total_gb: int = 0,
|
||
limit_ip: int = 0,
|
||
enable: bool = True,
|
||
comment: str = "",
|
||
) -> dict[str, Any]:
|
||
"""Добавляет WireGuard-клиента (3x-ui 3.5.0: POST /panel/api/clients/add)."""
|
||
await self.ensure_auth()
|
||
sid = (sub_id or "").strip() or _new_sub_id()
|
||
try:
|
||
tg_num = int(tg_id) if tg_id not in ("", None) else 0
|
||
except (TypeError, ValueError):
|
||
tg_num = 0
|
||
client: dict[str, Any] = {
|
||
"email": email,
|
||
"enable": bool(enable),
|
||
"expiryTime": int(expiry_ms),
|
||
"totalGB": int(total_gb),
|
||
"limitIp": int(limit_ip),
|
||
"tgId": tg_num,
|
||
"subId": sid,
|
||
"comment": comment or "",
|
||
"reset": 0,
|
||
}
|
||
modern_body = {
|
||
"client": client,
|
||
"inboundIds": [int(inbound_id)],
|
||
}
|
||
try:
|
||
payload = await self._request(
|
||
"POST",
|
||
"/panel/api/clients/add",
|
||
json_body=modern_body,
|
||
)
|
||
except XuiError as modern_exc:
|
||
# Fallback для старых панелей с /inbounds/addClient
|
||
legacy_body = {
|
||
"id": int(inbound_id),
|
||
"settings": json.dumps({"clients": [client]}, ensure_ascii=False),
|
||
}
|
||
try:
|
||
payload = await self._request(
|
||
"POST",
|
||
"/panel/api/inbounds/addClient",
|
||
json_body=legacy_body,
|
||
)
|
||
except XuiError:
|
||
raise modern_exc from None
|
||
|
||
created = await self.get_client(inbound_id, email=email)
|
||
if created and created.get("subId"):
|
||
sid = str(created.get("subId") or sid)
|
||
return {
|
||
"sub_id": sid,
|
||
"email": email,
|
||
"client": created,
|
||
"response": payload,
|
||
}
|
||
|
||
def _find_client_in_inbound(
|
||
self, inbound: dict[str, Any], *, email: str
|
||
) -> dict[str, Any] | None:
|
||
settings = _parse_settings(inbound.get("settings"))
|
||
clients = settings.get("clients")
|
||
if not isinstance(clients, list):
|
||
return None
|
||
email_l = email.strip().lower()
|
||
for client in clients:
|
||
if isinstance(client, dict) and str(client.get("email") or "").strip().lower() == email_l:
|
||
return client
|
||
return None
|
||
|
||
def _path_email(self, email: str) -> str:
|
||
"""Email в path: простые id без кодирования, иначе PathEscape-совместимо."""
|
||
raw = (email or "").strip()
|
||
if re.fullmatch(r"[A-Za-z0-9_.-]+", raw or ""):
|
||
return raw
|
||
return quote(raw, safe="._-~")
|
||
|
||
@staticmethod
|
||
def _sanitize_client_payload(
|
||
client: dict[str, Any],
|
||
*,
|
||
enable: bool | None = None,
|
||
email: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Только поля клиента — без traffic/inboundIds (иначе update ломается)."""
|
||
allowed = {
|
||
"email",
|
||
"enable",
|
||
"expiryTime",
|
||
"totalGB",
|
||
"limitIp",
|
||
"tgId",
|
||
"subId",
|
||
"comment",
|
||
"reset",
|
||
"flow",
|
||
"password",
|
||
"id",
|
||
"security",
|
||
"encryption",
|
||
"privateKey",
|
||
"publicKey",
|
||
"preSharedKey",
|
||
"allowedIPs",
|
||
"keepAlive",
|
||
"method",
|
||
"auth",
|
||
}
|
||
out: dict[str, Any] = {}
|
||
for key, value in client.items():
|
||
if key not in allowed or value is None:
|
||
continue
|
||
if key == "allowedIPs" and not isinstance(value, list):
|
||
continue
|
||
if key != "allowedIPs" and isinstance(value, (dict, list)):
|
||
continue
|
||
out[key] = value
|
||
|
||
# Client.id в 3x-ui — строка (UUID и т.п.), не numeric id из таблицы clients
|
||
raw_id = out.get("id", None)
|
||
if raw_id is None and client.get("uuid"):
|
||
raw_id = client.get("uuid")
|
||
if isinstance(raw_id, int):
|
||
# числовой PK таблицы clients нельзя слать как Client.id
|
||
out.pop("id", None)
|
||
if client.get("uuid"):
|
||
out["id"] = str(client["uuid"])
|
||
elif raw_id is not None:
|
||
out["id"] = str(raw_id)
|
||
|
||
if email:
|
||
out["email"] = email
|
||
elif "email" not in out and client.get("email"):
|
||
out["email"] = client["email"]
|
||
if enable is not None:
|
||
out["enable"] = bool(enable)
|
||
|
||
# tgId панель иногда ждёт число
|
||
if "tgId" in out:
|
||
try:
|
||
out["tgId"] = int(out["tgId"] or 0)
|
||
except (TypeError, ValueError):
|
||
out["tgId"] = 0
|
||
return out
|
||
|
||
async def get_client(
|
||
self, inbound_id: int, *, email: str
|
||
) -> dict[str, Any] | None:
|
||
await self.ensure_auth()
|
||
# Modern API
|
||
try:
|
||
payload = await self._request(
|
||
"GET",
|
||
f"/panel/api/clients/get/{self._path_email(email)}",
|
||
)
|
||
obj = payload.get("obj")
|
||
if isinstance(obj, dict):
|
||
client = obj.get("client") if isinstance(obj.get("client"), dict) else obj
|
||
if isinstance(client, dict) and client.get("email"):
|
||
return client
|
||
except XuiError:
|
||
pass
|
||
inbound = await self.get_inbound(inbound_id)
|
||
return self._find_client_in_inbound(inbound, email=email)
|
||
|
||
async def update_client(
|
||
self,
|
||
inbound_id: int,
|
||
client_id: str,
|
||
client: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
"""Обновление клиента: modern clients/update → legacy updateClient → inbound update."""
|
||
await self.ensure_auth()
|
||
email = str(client.get("email") or client_id).strip()
|
||
clean = self._sanitize_client_payload(client, email=email)
|
||
|
||
# 1) Modern API
|
||
for path in (
|
||
f"/panel/api/clients/update/{self._path_email(email)}",
|
||
f"/panel/api/clients/update/{self._path_email(email)}?inboundIds={int(inbound_id)}",
|
||
):
|
||
try:
|
||
return await self._request("POST", path, json_body=clean)
|
||
except XuiError:
|
||
pass
|
||
|
||
# 2) Legacy inbounds/updateClient
|
||
legacy_body = {
|
||
"id": int(inbound_id),
|
||
"settings": json.dumps({"clients": [clean]}, ensure_ascii=False),
|
||
}
|
||
for client_key in (email, str(clean.get("id") or ""), client_id):
|
||
if not client_key:
|
||
continue
|
||
path = f"/panel/api/inbounds/updateClient/{self._path_email(str(client_key))}"
|
||
try:
|
||
return await self._request("POST", path, json_body=legacy_body)
|
||
except XuiError:
|
||
pass
|
||
try:
|
||
return await self._request(
|
||
"POST",
|
||
path,
|
||
data={
|
||
"id": str(int(inbound_id)),
|
||
"settings": legacy_body["settings"],
|
||
},
|
||
)
|
||
except XuiError:
|
||
pass
|
||
|
||
# 3) Fallback: правим поля прямо в settings inbound
|
||
return await self._update_inbound_client_fields(
|
||
inbound_id,
|
||
email=email,
|
||
patch=clean,
|
||
)
|
||
|
||
async def _update_inbound_client_fields(
|
||
self,
|
||
inbound_id: int,
|
||
*,
|
||
email: str,
|
||
patch: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
inbound = await self.get_inbound(inbound_id)
|
||
settings = _parse_settings(inbound.get("settings"))
|
||
clients = settings.get("clients")
|
||
if not isinstance(clients, list):
|
||
raise XuiError(f"В inbound #{inbound_id} нет clients[]")
|
||
email_l = email.strip().lower()
|
||
found = False
|
||
new_clients: list[Any] = []
|
||
for item in clients:
|
||
if not isinstance(item, dict):
|
||
new_clients.append(item)
|
||
continue
|
||
if str(item.get("email") or "").strip().lower() == email_l:
|
||
merged = dict(item)
|
||
merged.update(patch)
|
||
merged["email"] = item.get("email") or email
|
||
new_clients.append(merged)
|
||
found = True
|
||
else:
|
||
new_clients.append(item)
|
||
if not found:
|
||
raise XuiError(f"Клиент {email} не найден в inbound #{inbound_id}")
|
||
settings["clients"] = new_clients
|
||
|
||
payload = dict(inbound)
|
||
payload["id"] = int(inbound_id)
|
||
payload["settings"] = json.dumps(settings, ensure_ascii=False)
|
||
for key in ("streamSettings", "sniffing", "allocate"):
|
||
val = payload.get(key)
|
||
if isinstance(val, (dict, list)):
|
||
payload[key] = json.dumps(val, ensure_ascii=False)
|
||
# убрать служебные поля ответа get
|
||
for key in ("clientStats", "clients", "obj"):
|
||
payload.pop(key, None)
|
||
return await self._request(
|
||
"POST",
|
||
f"/panel/api/inbounds/update/{int(inbound_id)}",
|
||
json_body=payload,
|
||
)
|
||
|
||
async def set_client_enabled(
|
||
self,
|
||
inbound_id: int,
|
||
*,
|
||
email: str,
|
||
enable: bool,
|
||
) -> dict[str, Any]:
|
||
await self.ensure_auth()
|
||
current = await self.get_client(inbound_id, email=email)
|
||
if not current:
|
||
# всё равно пробуем через inbound settings
|
||
return await self._update_inbound_client_fields(
|
||
inbound_id,
|
||
email=email,
|
||
patch={"enable": bool(enable), "email": email},
|
||
)
|
||
clean = self._sanitize_client_payload(current, enable=enable, email=email)
|
||
try:
|
||
return await self.update_client(inbound_id, email, clean)
|
||
except XuiError:
|
||
return await self._update_inbound_client_fields(
|
||
inbound_id,
|
||
email=email,
|
||
patch=clean,
|
||
)
|
||
|
||
async def delete_client_by_email(self, inbound_id: int, email: str) -> dict[str, Any]:
|
||
await self.ensure_auth()
|
||
try:
|
||
return await self._request(
|
||
"POST",
|
||
f"/panel/api/clients/del/{self._path_email(email)}",
|
||
)
|
||
except XuiError:
|
||
path = (
|
||
f"/panel/api/inbounds/{int(inbound_id)}/"
|
||
f"delClientByEmail/{self._path_email(email)}"
|
||
)
|
||
try:
|
||
return await self._request("POST", path)
|
||
except XuiError:
|
||
# отвязка / disable как мягкое удаление
|
||
return await self._update_inbound_client_fields(
|
||
inbound_id,
|
||
email=email,
|
||
patch={"enable": False, "email": email},
|
||
)
|
||
|
||
async def test_connection(self) -> XuiConnectionInfo:
|
||
await self.open()
|
||
try:
|
||
await self.ensure_auth()
|
||
inbounds = await self.list_inbounds()
|
||
wg = [i for i in inbounds if i.is_wireguard]
|
||
version = None
|
||
try:
|
||
version = await self.get_version()
|
||
except XuiError:
|
||
version = None
|
||
auth_mode = "api_key" if self.api_key else "login"
|
||
msg = (
|
||
f"OK · {auth_mode} · inbounds {len(inbounds)} · WireGuard {len(wg)}"
|
||
+ (f" · panel {version}" if version else f" · target {SUPPORTED_PANEL}+")
|
||
)
|
||
return XuiConnectionInfo(
|
||
ok=True,
|
||
version=version,
|
||
inbounds_total=len(inbounds),
|
||
wireguard=wg,
|
||
message=msg,
|
||
)
|
||
except XuiError as exc:
|
||
return XuiConnectionInfo(
|
||
ok=False,
|
||
version=None,
|
||
inbounds_total=0,
|
||
wireguard=[],
|
||
message=str(exc),
|
||
)
|
||
|
||
|
||
def subscription_preview(subscription_url: str, sample_sub_id: str = "exampleSubId") -> str:
|
||
link = build_subscription_link(subscription_url, sample_sub_id)
|
||
if link:
|
||
return link
|
||
return urljoin(_normalize_base_url(subscription_url or "https://example.invalid") + "/", sample_sub_id)
|