"""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'), 'sub_url': (sync.get('xui_sub_url') or '').strip().rstrip('/'), } def build_subscription_url(settings: dict, sub_id: str) -> str: """Build public subscription URL from panel settings + client subId.""" sub_id = (sub_id or '').strip() if not sub_id: return '' base = _settings_creds(settings).get('sub_url') or '' if not base: return '' return f"{base}/{sub_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, 'expiry_time': expiry_time, } 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, expiry_time: int = 0, ) -> dict: """Create a VLESS client on 3x-ui and return {client_id, config, subscription_url, 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 created = None # Avoid collisions for _ in range(5): try: created = await api.add_vless_client( email=email, inbound_id=inbound, comment=name or email, expiry_time=int(expiry_time or 0), ) break 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 if created is None: created = await api.add_vless_client( email=email, inbound_id=inbound, comment=name or email, expiry_time=int(expiry_time or 0), ) sub_url = build_subscription_url(settings, created.get('sub_id') or '') # Prefer subscription URL as the main share string when configured if sub_url: created['subscription_url'] = sub_url created['config'] = sub_url else: created['subscription_url'] = '' created['inbound_id'] = inbound return created 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()