Amnezia-Web-Panel-main/managers/xui_api.py

"""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()