"""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}" def _resolve_settings(settings: dict, panel_id: Optional[str] = None) -> dict: """Scope settings to a specific xui panel when panel_id / multi-server registry is used.""" try: from managers.xui_servers import resolve_panel_settings, ensure_xui_servers ensure_xui_servers(settings) return resolve_panel_settings(settings, panel_id) except Exception: return settings 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, }, ) # Share URL is resolved by the caller via resolve_share_link (panel assigns link) return { 'client_id': email, 'uuid': client_uuid, 'sub_id': sub_id, 'email': email, 'config': '', 'links': [], 'expiry_time': expiry_time, 'inbound_id': int(inbound_id), } async def get_client_links(self, email: str) -> list: email = (email or '').strip() if not email: return [] for path in ( f'/panel/api/clients/links/{quote(email, safe="")}', f'/panel/api/inbounds/clientLinks/0/{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 %s failed: %s', path, e) return [] async def get_sub_links(self, sub_id: str) -> list: """Ask 3x-ui for protocol URLs produced for this subscription id.""" sub_id = (sub_id or '').strip() if not sub_id: return [] path = f'/panel/api/inbounds/getSubLinks/{quote(sub_id, 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('getSubLinks failed: %s', e) return [] async def get_panel_sub_base(self) -> str: """Try to read public subscription base URL from 3x-ui panel settings.""" for path, method in ( ('/panel/api/setting', 'GET'), ('/panel/setting/all', 'POST'), ('/panel/api/settings', 'GET'), ): try: payload = await self._request(method, path) except XuiApiError: continue obj = payload.get('obj') if isinstance(payload, dict) else payload if not isinstance(obj, dict): continue # Common field names across 3x-ui versions sub_uri = (obj.get('subURI') or obj.get('subUri') or obj.get('sub_uri') or '').strip() sub_path = (obj.get('subPath') or obj.get('sub_path') or '/sub').strip() or '/sub' sub_port = obj.get('subPort') or obj.get('sub_port') sub_listen = (obj.get('subListen') or '').strip() if sub_uri: return sub_uri.rstrip('/') # Build from panel host + sub path if only path is configured if sub_path and self.base_url: from urllib.parse import urlparse, urlunparse parsed = urlparse(self.base_url) host = sub_listen or parsed.hostname or '' if not host: continue scheme = 'https' if parsed.scheme == 'https' else 'http' netloc = f'{host}:{sub_port}' if sub_port else host path_part = sub_path if sub_path.startswith('/') else f'/{sub_path}' return urlunparse((scheme, netloc, path_part.rstrip('/'), '', '', '')) return '' async def resolve_share_link(self, *, email: str, sub_id: str = '', configured_sub_base: str = '') -> dict: """Ask the panel for the shareable subscription / protocol link (do not invent config).""" email = (email or '').strip() sub_id = (sub_id or '').strip() base = (configured_sub_base or '').strip().rstrip('/') if not base: try: base = (await self.get_panel_sub_base() or '').rstrip('/') except Exception as e: logger.warning('get_panel_sub_base failed: %s', e) base = '' links = [] if sub_id: links = await self.get_sub_links(sub_id) if not links and email: links = await self.get_client_links(email) # Prefer HTTP(S) link returned by 3x-ui; else build from panel/settings sub base + subId panel_http = next( (u for u in links if isinstance(u, str) and u.startswith(('http://', 'https://'))), '', ) subscription_url = panel_http or (f'{base}/{sub_id}' if base and sub_id else '') share = subscription_url or next( (u for u in links if isinstance(u, str) and u.startswith(('vless://', 'vmess://', 'trojan://', 'ss://'))), '', ) return { 'subscription_url': subscription_url, 'links': links, 'share': share, 'sub_base': base, } 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, panel_id: Optional[str] = None, ) -> dict: """Create a client via 3x-ui API; inbound + share link come from the panel response.""" scoped = _resolve_settings(settings, panel_id) creds = _settings_creds(scoped) inbound = inbound_id if inbound_id not in (None, 0, '0') else creds.get('inbound_id') try: inbound = int(inbound or 0) except (TypeError, ValueError): inbound = 0 async with XuiApi.from_panel_settings(scoped) as api: # Panel assigns inbound: use configured default, else first VLESS on that panel 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']) 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 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), ) # Ask the panel for the subscription / protocol link (do not invent config here) share = await api.resolve_share_link( email=created.get('email') or email, sub_id=created.get('sub_id') or '', configured_sub_base=creds.get('sub_url') or '', ) subscription_url = share.get('subscription_url') or '' panel_share = share.get('share') or '' panel_links = share.get('links') or created.get('links') or [] created['subscription_url'] = subscription_url created['links'] = panel_links # What we show the user: panel subscription URL, else panel protocol link created['config'] = subscription_url or panel_share or created.get('config') or '' if not created['config'] and panel_links: created['config'] = panel_links[0] created['inbound_id'] = int(created.get('inbound_id') or inbound) created['panel_id'] = panel_id or '' return created async def xui_get_config(settings: dict, email: str, panel_id: Optional[str] = None) -> str: scoped = _resolve_settings(settings, panel_id) creds = _settings_creds(scoped) async with XuiApi.from_panel_settings(scoped) as api: sub_id = '' try: for inbound in await api.list_inbounds(): if not isinstance(inbound, dict): continue settings_raw = inbound.get('settings') or '{}' if isinstance(settings_raw, str): try: settings_obj = json.loads(settings_raw) except Exception: settings_obj = {} else: settings_obj = settings_raw if isinstance(settings_raw, dict) else {} for c in settings_obj.get('clients') or []: if isinstance(c, dict) and (c.get('email') or '') == email: sub_id = (c.get('subId') or c.get('sub_id') or '').strip() break if sub_id: break except Exception as e: logger.warning('Could not resolve subId for %s: %s', email, e) share = await api.resolve_share_link( email=email, sub_id=sub_id, configured_sub_base=creds.get('sub_url') or '', ) if share.get('share'): return share['share'] links = share.get('links') or await api.get_client_links(email) vless = next((u for u in links if isinstance(u, str) and 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, panel_id: Optional[str] = None) -> None: scoped = _resolve_settings(settings, panel_id) async with XuiApi.from_panel_settings(scoped) as api: await api.delete_client(email) async def xui_toggle_client(settings: dict, email: str, enable: bool, panel_id: Optional[str] = None) -> None: scoped = _resolve_settings(settings, panel_id) async with XuiApi.from_panel_settings(scoped) as api: await api.set_client_enabled(email, enable) async def xui_list_vless_inbounds(settings: dict, panel_id: Optional[str] = None) -> list: scoped = _resolve_settings(settings, panel_id) async with XuiApi.from_panel_settings(scoped) as api: return await api.list_vless_inbounds()