test2/vpn
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

3x-ui: create VLESS/WireGuard clients and reload available inbounds

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
test2
2026-07-25 22:09:59 +03:00
co-authored by Cursor
parent b67e7b80ba
commit 1626c16df0
4 changed files with 525 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/xui.py
+210 -51
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import logging
import secrets
import string
from typing import Any
from urllib.parse import urljoin
@@ -10,6 +12,8 @@ import httpx
logger = logging.getLogger(__name__)
SUPPORTED_CLIENT_PROTOCOLS = ("vless", "wireguard")
class XuiApiError(Exception):
def __init__(self, message: str, status_code: int | None = None):
@@ -17,13 +21,18 @@ class XuiApiError(Exception):
self.status_code = status_code
def _random_sub_id(length: int = 16) -> str:
alphabet = string.ascii_lowercase + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
class XuiClient:
"""
Client for MHSanaei/3x-ui REST API (v3.5.0).
Auth options:
1) API Token — Settings → Security → API Token → Authorization: Bearer …
2) Username/password session (with CSRF when required)
Auth:
1) API Token → Authorization: Bearer …
2) Username/password session (+ CSRF when required)
"""
def __init__(
@@ -75,9 +84,7 @@ class XuiClient:
return self._client
async def login(self) -> None:
"""Session login for cookie auth (when no API token)."""
assert self.username and self.password
# Warm-up cookies / CSRF from login page if present
csrf: str | None = None
try:
warm = await self.client.get("/")
@@ -94,12 +101,11 @@ class XuiClient:
except Exception: # noqa: BLE001
pass
headers = {}
headers: dict[str, str] = {}
if csrf:
headers["X-CSRF-Token"] = csrf
payload = {"username": self.username, "password": self.password}
# Try JSON first (v3), then form (older)
resp = await self.client.post("/login", json=payload, headers=headers)
if resp.status_code == 403 and csrf:
resp = await self.client.post(
@@ -144,58 +150,214 @@ class XuiClient:
data = await self._request("GET", "/panel/api/server/status")
return data.get("obj") if isinstance(data, dict) else data
async def get_panel_update_info(self) -> dict:
data = await self._request("GET", "/panel/api/server/getPanelUpdateInfo")
return data.get("obj") if isinstance(data, dict) else data
async def get_xray_version(self) -> Any:
data = await self._request("GET", "/panel/api/server/getXrayVersion")
return data.get("obj") if isinstance(data, dict) else data
async def get_new_uuid(self) -> str:
data = await self._request("GET", "/panel/api/server/getNewUUID")
obj = data.get("obj") if isinstance(data, dict) else data
if isinstance(obj, str):
return obj
if isinstance(obj, dict) and obj.get("uuid"):
return str(obj["uuid"])
raise XuiApiError("Не удалось получить UUID")
async def get_new_x25519(self) -> dict[str, str]:
data = await self._request("GET", "/panel/api/server/getNewX25519Cert")
obj = data.get("obj") if isinstance(data, dict) else data
if not isinstance(obj, dict) or not obj.get("privateKey") or not obj.get("publicKey"):
raise XuiApiError("Не удалось получить X25519 ключи")
return {"privateKey": str(obj["privateKey"]), "publicKey": str(obj["publicKey"])}
async def list_inbounds(self) -> list[dict]:
data = await self._request("GET", "/panel/api/inbounds/list")
obj = data.get("obj") if isinstance(data, dict) else data
return obj if isinstance(obj, list) else []
async def list_inbound_options(self) -> list[dict]:
"""Lightweight inbound list for dropdowns (v3.5)."""
try:
data = await self._request("GET", "/panel/api/inbounds/options")
obj = data.get("obj") if isinstance(data, dict) else data
if isinstance(obj, list):
return obj
except XuiApiError:
pass
# Fallback: slim projection from full list
inbounds = await self.list_inbounds()
return [
{
"id": i.get("id"),
"remark": i.get("remark"),
"tag": i.get("tag"),
"protocol": i.get("protocol"),
"port": i.get("port"),
"enable": i.get("enable", True),
"tlsFlowCapable": False,
}
for i in inbounds
]
async def list_available_inbounds(
self,
*,
protocols: tuple[str, ...] | list[str] | None = None,
enabled_only: bool = True,
) -> list[dict]:
options = await self.list_inbound_options()
result: list[dict] = []
wanted = {p.lower() for p in protocols} if protocols else None
for item in options:
proto = str(item.get("protocol") or "").lower()
if wanted and proto not in wanted:
continue
if enabled_only and item.get("enable") is False:
continue
result.append(item)
return result
async def get_inbound(self, inbound_id: int) -> dict:
data = await self._request("GET", f"/panel/api/inbounds/get/{inbound_id}")
obj = data.get("obj") if isinstance(data, dict) else data
if not isinstance(obj, dict):
raise XuiApiError(f"Inbound {inbound_id} not found")
return obj
async def list_clients(self) -> list[dict]:
data = await self._request("GET", "/panel/api/clients/list")
obj = data.get("obj") if isinstance(data, dict) else data
return obj if isinstance(obj, list) else []
async def add_client(
self,
*,
email: str,
inbound_ids: list[int],
total_gb: int = 0,
expiry_time: int = 0,
limit_ip: int = 0,
enable: bool = True,
remark: str | None = None,
) -> dict:
body = {
"client": {
"email": email,
"totalGB": total_gb,
"expiryTime": expiry_time,
"limitIp": limit_ip,
"enable": enable,
"tgId": 0,
"comment": remark or "",
},
"inboundIds": inbound_ids,
}
return await self._request("POST", "/panel/api/clients/add", json=body)
async def delete_client(self, email: str) -> dict:
return await self._request("POST", f"/panel/api/clients/del/{email}")
async def get_client(self, email: str) -> dict:
data = await self._request("GET", f"/panel/api/clients/get/{email}")
return data.get("obj") if isinstance(data, dict) else data
async def get_client_links(self, email: str) -> list[str]:
try:
data = await self._request("GET", f"/panel/api/clients/links/{email}")
obj = data.get("obj") if isinstance(data, dict) else data
if isinstance(obj, list):
return [str(x) for x in obj]
except XuiApiError:
pass
return []
async def add_client_raw(self, client: dict, inbound_ids: list[int]) -> dict:
body = {"client": client, "inboundIds": inbound_ids}
return await self._request("POST", "/panel/api/clients/add", json=body)
async def add_vless_client(
self,
*,
email: str,
inbound_id: int,
total_gb: int = 0,
expiry_time: int = 0,
limit_ip: int = 0,
flow: str = "",
comment: str = "",
) -> dict:
inbound = await self.get_inbound(inbound_id)
proto = str(inbound.get("protocol") or "").lower()
if proto != "vless":
raise XuiApiError(f"Inbound #{inbound_id} — не VLESS (protocol={proto})")
uuid = await self.get_new_uuid()
# Auto vision flow for Reality TCP if caller left flow empty
if not flow:
try:
stream = inbound.get("streamSettings") or {}
if isinstance(stream, str):
import json
stream = json.loads(stream)
network = str(stream.get("network") or "tcp").lower()
security = str(stream.get("security") or "").lower()
if network == "tcp" and security in {"reality", "tls"}:
flow = "xtls-rprx-vision"
except Exception: # noqa: BLE001
flow = ""
client = {
"id": uuid,
"email": email,
"enable": True,
"flow": flow,
"totalGB": total_gb,
"expiryTime": expiry_time,
"limitIp": limit_ip,
"tgId": 0,
"subId": _random_sub_id(),
"comment": comment,
}
await self.add_client_raw(client, [inbound_id])
details = await self.get_client(email)
links = await self.get_client_links(email)
return {
"protocol": "vless",
"email": email,
"inbound_id": inbound_id,
"uuid": uuid,
"flow": flow,
"details": details,
"links": links,
}
async def add_wireguard_client(
self,
*,
email: str,
inbound_id: int,
total_gb: int = 0,
expiry_time: int = 0,
limit_ip: int = 0,
comment: str = "",
) -> dict:
inbound = await self.get_inbound(inbound_id)
proto = str(inbound.get("protocol") or "").lower()
if proto != "wireguard":
raise XuiApiError(f"Inbound #{inbound_id} — не WireGuard (protocol={proto})")
keys = await self.get_new_x25519()
client = {
"email": email,
"enable": True,
"privateKey": keys["privateKey"],
"publicKey": keys["publicKey"],
"preSharedKey": "",
"allowedIPs": ["0.0.0.0/0", "::/0"],
"keepAlive": 0,
"totalGB": total_gb,
"expiryTime": expiry_time,
"limitIp": limit_ip,
"tgId": 0,
"subId": _random_sub_id(),
"comment": comment,
}
await self.add_client_raw(client, [inbound_id])
details = await self.get_client(email)
# WG has no vless:// link — expose keys for conf building
return {
"protocol": "wireguard",
"email": email,
"inbound_id": inbound_id,
"privateKey": keys["privateKey"],
"publicKey": keys["publicKey"],
"details": details,
"inbound": {
"id": inbound.get("id"),
"remark": inbound.get("remark"),
"port": inbound.get("port"),
"listen": inbound.get("listen"),
},
"links": [],
}
async def delete_client(self, email: str) -> dict:
return await self._request("POST", f"/panel/api/clients/del/{email}")
async def test_connection(self) -> dict[str, Any]:
"""Verify credentials and return summary for UI."""
status = await self.get_status()
info: dict[str, Any] = {"status": status}
try:
@@ -203,17 +365,14 @@ class XuiClient:
except Exception as exc: # noqa: BLE001
info["xray_version_error"] = str(exc)
try:
inbounds = await self.list_inbounds()
info["inbounds_count"] = len(inbounds)
info["inbounds"] = [
{
"id": i.get("id"),
"remark": i.get("remark"),
"protocol": i.get("protocol"),
"port": i.get("port"),
"enable": i.get("enable"),
}
for i in inbounds[:50]
available = await self.list_available_inbounds(
protocols=SUPPORTED_CLIENT_PROTOCOLS, enabled_only=False
)
info["inbounds_count"] = len(available)
info["inbounds"] = available[:100]
info["vless_inbounds"] = [i for i in available if str(i.get("protocol")).lower() == "vless"]
info["wireguard_inbounds"] = [
i for i in available if str(i.get("protocol")).lower() == "wireguard"
]
except Exception as exc: # noqa: BLE001
info["inbounds_error"] = str(exc)