Add 3x-ui VLESS configs and guest access without registration.
Guests can open a public link to view or create VPN configs; panel creates VLESS clients via the 3x-ui API. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
orohi
@@ -0,0 +1,350 @@
|
||||
"""3x-ui HTTP API client — create VLESS clients and fetch share links.
|
||||
|
||||
Works with modern panels (/panel/api/clients/*) and falls back to legacy
|
||||
/panel/api/inbounds/addClient when needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import secrets
|
||||
import string
|
||||
import uuid
|
||||
from typing import Any, Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _new_sub_id(length: int = 16) -> str:
|
||||
alphabet = string.ascii_lowercase + string.digits
|
||||
return ''.join(secrets.choice(alphabet) for _ in range(length))
|
||||
|
||||
|
||||
def _settings_creds(settings: dict) -> dict:
|
||||
sync = (settings or {}).get('sync') or {}
|
||||
return {
|
||||
'url': (sync.get('xui_url') or '').strip().rstrip('/'),
|
||||
'api_token': (sync.get('xui_api_token') or '').strip(),
|
||||
'username': (sync.get('xui_username') or '').strip(),
|
||||
'password': sync.get('xui_password') or '',
|
||||
'inbound_id': sync.get('xui_inbound_id'),
|
||||
}
|
||||
|
||||
|
||||
class XuiApiError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class XuiApi:
|
||||
def __init__(self, base_url: str, *, api_token: str = '', username: str = '', password: str = ''):
|
||||
self.base_url = (base_url or '').rstrip('/')
|
||||
if not self.base_url:
|
||||
raise XuiApiError('3x-ui URL is not configured')
|
||||
self.api_token = (api_token or '').strip()
|
||||
self.username = (username or '').strip()
|
||||
self.password = password or ''
|
||||
self._client: Optional[httpx.AsyncClient] = None
|
||||
|
||||
@classmethod
|
||||
def from_panel_settings(cls, settings: dict) -> 'XuiApi':
|
||||
c = _settings_creds(settings)
|
||||
return cls(c['url'], api_token=c['api_token'], username=c['username'], password=c['password'])
|
||||
|
||||
async def __aenter__(self) -> 'XuiApi':
|
||||
headers = {'Accept': 'application/json'}
|
||||
if self.api_token:
|
||||
headers['Authorization'] = f'Bearer {self.api_token}'
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=self.base_url,
|
||||
timeout=30.0,
|
||||
follow_redirects=True,
|
||||
headers=headers,
|
||||
)
|
||||
if not self.api_token:
|
||||
await self._login()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *args):
|
||||
if self._client:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
|
||||
async def _login(self):
|
||||
if not self.username or not self.password:
|
||||
raise XuiApiError('Provide 3x-ui API token or username/password')
|
||||
resp = await self._client.post('/login', json={
|
||||
'username': self.username,
|
||||
'password': self.password,
|
||||
})
|
||||
if resp.status_code != 200:
|
||||
raise XuiApiError(f'3x-ui login failed: HTTP {resp.status_code}')
|
||||
try:
|
||||
body = resp.json()
|
||||
except Exception:
|
||||
body = {}
|
||||
if body.get('success') is False:
|
||||
raise XuiApiError(f"3x-ui login failed: {body.get('msg', 'unknown error')}")
|
||||
|
||||
async def _request(self, method: str, path: str, **kwargs) -> Any:
|
||||
assert self._client is not None
|
||||
resp = await self._client.request(method, path, **kwargs)
|
||||
try:
|
||||
payload = resp.json()
|
||||
except Exception:
|
||||
raise XuiApiError(f'3x-ui {method} {path}: HTTP {resp.status_code} non-JSON')
|
||||
if resp.status_code >= 400:
|
||||
raise XuiApiError(
|
||||
f"3x-ui {method} {path}: HTTP {resp.status_code} {payload.get('msg') or payload}"
|
||||
)
|
||||
if isinstance(payload, dict) and payload.get('success') is False:
|
||||
raise XuiApiError(payload.get('msg') or f'3x-ui {method} {path} failed')
|
||||
return payload
|
||||
|
||||
async def list_inbounds(self) -> list:
|
||||
for path, method in (
|
||||
('/panel/api/inbounds/list', 'GET'),
|
||||
('/panel/api/inbounds/list', 'POST'),
|
||||
('/panel/inbound/list', 'POST'),
|
||||
):
|
||||
try:
|
||||
payload = await self._request(method, path)
|
||||
except XuiApiError:
|
||||
continue
|
||||
obj = payload.get('obj') if isinstance(payload, dict) else None
|
||||
if isinstance(obj, list):
|
||||
return obj
|
||||
return []
|
||||
|
||||
async def list_vless_inbounds(self) -> list:
|
||||
result = []
|
||||
for inbound in await self.list_inbounds():
|
||||
if not isinstance(inbound, dict):
|
||||
continue
|
||||
proto = (inbound.get('protocol') or '').lower()
|
||||
if proto != 'vless':
|
||||
continue
|
||||
result.append({
|
||||
'id': inbound.get('id'),
|
||||
'remark': inbound.get('remark') or f"VLESS:{inbound.get('port')}",
|
||||
'port': inbound.get('port'),
|
||||
'protocol': proto,
|
||||
'enable': bool(inbound.get('enable', True)),
|
||||
})
|
||||
return result
|
||||
|
||||
async def add_vless_client(
|
||||
self,
|
||||
*,
|
||||
email: str,
|
||||
inbound_id: int,
|
||||
comment: str = '',
|
||||
enable: bool = True,
|
||||
total_gb: int = 0,
|
||||
expiry_time: int = 0,
|
||||
limit_ip: int = 0,
|
||||
flow: str = '',
|
||||
) -> dict:
|
||||
email = (email or '').strip()
|
||||
if not email:
|
||||
raise XuiApiError('Client email is required')
|
||||
if not inbound_id:
|
||||
raise XuiApiError('VLESS inbound id is required')
|
||||
|
||||
client_uuid = str(uuid.uuid4())
|
||||
sub_id = _new_sub_id()
|
||||
client = {
|
||||
'id': client_uuid,
|
||||
'email': email,
|
||||
'enable': enable,
|
||||
'flow': flow or '',
|
||||
'limitIp': limit_ip,
|
||||
'totalGB': total_gb,
|
||||
'expiryTime': expiry_time,
|
||||
'tgId': 0,
|
||||
'subId': sub_id,
|
||||
'comment': comment or '',
|
||||
}
|
||||
|
||||
# Modern API
|
||||
try:
|
||||
await self._request(
|
||||
'POST',
|
||||
'/panel/api/clients/add',
|
||||
json={'client': client, 'inboundIds': [int(inbound_id)]},
|
||||
)
|
||||
except XuiApiError as modern_err:
|
||||
logger.info('Modern clients/add failed (%s), trying legacy addClient', modern_err)
|
||||
# Legacy: settings must be a JSON-encoded string on many builds
|
||||
settings_obj = {'clients': [client]}
|
||||
try:
|
||||
await self._request(
|
||||
'POST',
|
||||
'/panel/api/inbounds/addClient',
|
||||
json={
|
||||
'id': int(inbound_id),
|
||||
'settings': json.dumps(settings_obj),
|
||||
},
|
||||
)
|
||||
except XuiApiError:
|
||||
# Some builds accept nested object
|
||||
await self._request(
|
||||
'POST',
|
||||
'/panel/api/inbounds/addClient',
|
||||
json={
|
||||
'id': int(inbound_id),
|
||||
'settings': settings_obj,
|
||||
},
|
||||
)
|
||||
|
||||
links = await self.get_client_links(email)
|
||||
vless = next((u for u in links if isinstance(u, str) and u.startswith('vless://')), None)
|
||||
if not vless and links:
|
||||
vless = links[0]
|
||||
return {
|
||||
'client_id': email,
|
||||
'uuid': client_uuid,
|
||||
'sub_id': sub_id,
|
||||
'email': email,
|
||||
'config': vless or '',
|
||||
'links': links,
|
||||
}
|
||||
|
||||
async def get_client_links(self, email: str) -> list:
|
||||
email = (email or '').strip()
|
||||
if not email:
|
||||
return []
|
||||
path = f'/panel/api/clients/links/{quote(email, safe="")}'
|
||||
try:
|
||||
payload = await self._request('GET', path)
|
||||
obj = payload.get('obj') if isinstance(payload, dict) else None
|
||||
if isinstance(obj, list):
|
||||
return [x for x in obj if isinstance(x, str) and x.strip()]
|
||||
except XuiApiError as e:
|
||||
logger.warning('get_client_links failed: %s', e)
|
||||
return []
|
||||
|
||||
async def delete_client(self, email: str) -> None:
|
||||
email = (email or '').strip()
|
||||
if not email:
|
||||
return
|
||||
path = f'/panel/api/clients/del/{quote(email, safe="")}'
|
||||
try:
|
||||
await self._request('POST', path)
|
||||
return
|
||||
except XuiApiError as e:
|
||||
logger.warning('clients/del failed (%s), trying legacy', e)
|
||||
# Legacy: need inbound id — try remove by scanning
|
||||
for inbound in await self.list_inbounds():
|
||||
if not isinstance(inbound, dict):
|
||||
continue
|
||||
settings = inbound.get('settings')
|
||||
if isinstance(settings, str):
|
||||
try:
|
||||
settings = json.loads(settings)
|
||||
except Exception:
|
||||
settings = {}
|
||||
clients = (settings or {}).get('clients') if isinstance(settings, dict) else None
|
||||
if not isinstance(clients, list):
|
||||
continue
|
||||
match = next((c for c in clients if isinstance(c, dict) and c.get('email') == email), None)
|
||||
if not match:
|
||||
continue
|
||||
cid = match.get('id') or email
|
||||
inbound_id = inbound.get('id')
|
||||
for path in (
|
||||
f'/panel/api/inbounds/{inbound_id}/delClient/{quote(str(cid), safe="")}',
|
||||
f'/panel/api/inbounds/{inbound_id}/delClientByEmail/{quote(email, safe="")}',
|
||||
):
|
||||
try:
|
||||
await self._request('POST', path)
|
||||
return
|
||||
except XuiApiError:
|
||||
continue
|
||||
raise XuiApiError(f'Failed to delete 3x-ui client {email}')
|
||||
|
||||
async def set_client_enabled(self, email: str, enable: bool) -> None:
|
||||
email = (email or '').strip()
|
||||
if not email:
|
||||
return
|
||||
# Prefer full client get + update
|
||||
try:
|
||||
payload = await self._request('GET', f'/panel/api/clients/get/{quote(email, safe="")}')
|
||||
obj = payload.get('obj') if isinstance(payload, dict) else None
|
||||
client = None
|
||||
if isinstance(obj, dict):
|
||||
client = obj.get('client') if isinstance(obj.get('client'), dict) else obj
|
||||
if isinstance(client, dict):
|
||||
client = dict(client)
|
||||
client['enable'] = bool(enable)
|
||||
client['email'] = email
|
||||
await self._request(
|
||||
'POST',
|
||||
f'/panel/api/clients/update/{quote(email, safe="")}',
|
||||
json=client,
|
||||
)
|
||||
return
|
||||
except XuiApiError as e:
|
||||
logger.warning('toggle via clients/update failed: %s', e)
|
||||
raise XuiApiError(f'Failed to toggle 3x-ui client {email}')
|
||||
|
||||
|
||||
async def xui_create_vless_config(settings: dict, *, name: str, inbound_id: Optional[int] = None) -> dict:
|
||||
"""Create a VLESS client on 3x-ui and return {client_id, config, links}."""
|
||||
creds = _settings_creds(settings)
|
||||
inbound = inbound_id if inbound_id is not None else creds.get('inbound_id')
|
||||
try:
|
||||
inbound = int(inbound)
|
||||
except (TypeError, ValueError):
|
||||
inbound = 0
|
||||
|
||||
async with XuiApi.from_panel_settings(settings) as api:
|
||||
if not inbound:
|
||||
vless_inbounds = await api.list_vless_inbounds()
|
||||
if not vless_inbounds:
|
||||
raise XuiApiError('No VLESS inbound found on 3x-ui — create one in the panel first')
|
||||
inbound = int(vless_inbounds[0]['id'])
|
||||
|
||||
# Sanitize email: 3x-ui emails are unique free-form ids
|
||||
base = ''.join(ch if ch.isalnum() or ch in '._-+@' else '_' for ch in (name or 'user').strip())
|
||||
base = base[:48] or f'user_{secrets.token_hex(4)}'
|
||||
email = base
|
||||
# Avoid collisions
|
||||
for _ in range(5):
|
||||
try:
|
||||
return await api.add_vless_client(email=email, inbound_id=inbound, comment=name or email)
|
||||
except XuiApiError as e:
|
||||
if 'exist' in str(e).lower() or 'duplicate' in str(e).lower() or 'already' in str(e).lower():
|
||||
email = f'{base}_{secrets.token_hex(3)}'
|
||||
continue
|
||||
raise
|
||||
return await api.add_vless_client(email=email, inbound_id=inbound, comment=name or email)
|
||||
|
||||
|
||||
async def xui_get_config(settings: dict, email: str) -> str:
|
||||
async with XuiApi.from_panel_settings(settings) as api:
|
||||
links = await api.get_client_links(email)
|
||||
vless = next((u for u in links if u.startswith('vless://')), None)
|
||||
if vless:
|
||||
return vless
|
||||
if links:
|
||||
return links[0]
|
||||
raise XuiApiError(f'No share links for 3x-ui client {email}')
|
||||
|
||||
|
||||
async def xui_delete_client(settings: dict, email: str) -> None:
|
||||
async with XuiApi.from_panel_settings(settings) as api:
|
||||
await api.delete_client(email)
|
||||
|
||||
|
||||
async def xui_toggle_client(settings: dict, email: str, enable: bool) -> None:
|
||||
async with XuiApi.from_panel_settings(settings) as api:
|
||||
await api.set_client_enabled(email, enable)
|
||||
|
||||
|
||||
async def xui_list_vless_inbounds(settings: dict) -> list:
|
||||
async with XuiApi.from_panel_settings(settings) as api:
|
||||
return await api.list_vless_inbounds()
|
||||
Reference in New Issue
Block a user