"""3x-ui panel API client (compatible with v3.5.0).""" from __future__ import annotations import logging from typing import Any from urllib.parse import urljoin import httpx logger = logging.getLogger(__name__) class XuiApiError(Exception): def __init__(self, message: str, status_code: int | None = None): super().__init__(message) self.status_code = status_code 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) """ def __init__( self, base_url: str, *, username: str | None = None, password: str | None = None, api_token: str | None = None, verify_ssl: bool = True, timeout: float = 30.0, ): self.base_url = base_url.rstrip("/") self.username = (username or "").strip() or None self.password = (password or "").strip() or None self.api_token = (api_token or "").strip() or None self.verify_ssl = verify_ssl self.timeout = timeout self._client: httpx.AsyncClient | None = None self._logged_in = False if not self.api_token and not (self.username and self.password): raise ValueError("Укажите API Token или логин/пароль 3x-ui") async def __aenter__(self) -> XuiClient: headers: dict[str, str] = {"Accept": "application/json"} if self.api_token: headers["Authorization"] = f"Bearer {self.api_token}" self._client = httpx.AsyncClient( base_url=self.base_url, headers=headers, verify=self.verify_ssl, timeout=self.timeout, follow_redirects=True, ) if not self.api_token: await self.login() return self async def __aexit__(self, *args) -> None: if self._client: await self._client.aclose() self._client = None @property def client(self) -> httpx.AsyncClient: if not self._client: raise RuntimeError("XuiClient is not started") 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("/") csrf = warm.headers.get("X-CSRF-Token") or warm.cookies.get("csrf_token") except Exception: # noqa: BLE001 pass try: csrf_resp = await self.client.get("/panel/csrf-token") if csrf_resp.status_code == 200: data = csrf_resp.json() if isinstance(data, dict) and data.get("obj"): csrf = str(data["obj"]) except Exception: # noqa: BLE001 pass headers = {} 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( "/login", data=payload, headers={**headers, "Content-Type": "application/x-www-form-urlencoded"}, ) elif resp.status_code >= 400: resp = await self.client.post("/login", data=payload, headers=headers) if resp.status_code >= 400: raise XuiApiError(f"Login failed HTTP {resp.status_code}: {resp.text[:300]}", resp.status_code) try: body = resp.json() except Exception: # noqa: BLE001 body = {} if isinstance(body, dict) and body.get("success") is False: raise XuiApiError(body.get("msg") or "Wrong username or password") self._logged_in = True async def _request(self, method: str, path: str, **kwargs) -> Any: path = path if path.startswith("/") else f"/{path}" resp = await self.client.request(method, path, **kwargs) if resp.status_code == 401 and not self.api_token and not self._logged_in: await self.login() resp = await self.client.request(method, path, **kwargs) if resp.status_code >= 400: raise XuiApiError(f"HTTP {resp.status_code}: {resp.text[:400]}", resp.status_code) try: data = resp.json() except Exception as exc: # noqa: BLE001 raise XuiApiError(f"Invalid JSON response: {exc}") from exc if isinstance(data, dict) and data.get("success") is False: raise XuiApiError(data.get("msg") or "API error") return data async def get_status(self) -> dict: 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 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_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 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: info["xray_version"] = await self.get_xray_version() 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] ] except Exception as exc: # noqa: BLE001 info["inbounds_error"] = str(exc) try: clients = await self.list_clients() info["clients_count"] = len(clients) except Exception as exc: # noqa: BLE001 info["clients_error"] = str(exc) return info def normalize_base_url(url: str) -> str: url = url.strip().rstrip("/") if not url.startswith(("http://", "https://")): url = "https://" + url return url def public_panel_link(base_url: str) -> str: return urljoin(normalize_base_url(base_url) + "/", "")