298 lines
9.5 KiB
Python
298 lines
9.5 KiB
Python
"""Клиент Amnezia Web Panel (AWG / AWG 2.0) — Bearer API token."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DEFAULT_PROTOCOL = "awg2"
|
||
|
||
|
||
class AwgError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class AwgPanelServer:
|
||
id: int
|
||
name: str
|
||
host: str
|
||
protocols: list[str]
|
||
|
||
@property
|
||
def has_awg2(self) -> bool:
|
||
return any(
|
||
str(p).split("__", 1)[0] in {"awg2", "awg"} for p in self.protocols
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class AwgConnectionInfo:
|
||
ok: bool
|
||
alive: bool
|
||
ping_ms: int | None
|
||
protocols: list[str]
|
||
message: str = ""
|
||
|
||
|
||
def _normalize_base_url(url: str) -> str:
|
||
raw = (url or "").strip().rstrip("/")
|
||
if not raw:
|
||
raise AwgError("API URL пуст")
|
||
if not raw.startswith(("http://", "https://")):
|
||
raw = "https://" + raw
|
||
parsed = urlparse(raw)
|
||
if not parsed.netloc:
|
||
raise AwgError("Некорректный API URL")
|
||
return raw
|
||
|
||
|
||
class AwgClient:
|
||
"""HTTP-клиент к Amnezia Web Panel (panel5 / PRVTPRO)."""
|
||
|
||
def __init__(
|
||
self,
|
||
api_url: str,
|
||
*,
|
||
api_token: str,
|
||
verify_ssl: bool = True,
|
||
timeout: float = 60.0,
|
||
) -> None:
|
||
self.base_url = _normalize_base_url(api_url)
|
||
self.api_token = (api_token or "").strip()
|
||
if not self.api_token:
|
||
raise AwgError("Нужен API Token (Settings → API Tokens)")
|
||
self.verify_ssl = verify_ssl
|
||
self._client = httpx.AsyncClient(
|
||
base_url=self.base_url,
|
||
timeout=timeout,
|
||
verify=verify_ssl,
|
||
headers={
|
||
"Authorization": f"Bearer {self.api_token}",
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
},
|
||
)
|
||
|
||
async def __aenter__(self) -> "AwgClient":
|
||
return self
|
||
|
||
async def __aexit__(self, *args: Any) -> None:
|
||
await self.aclose()
|
||
|
||
async def aclose(self) -> None:
|
||
await self._client.aclose()
|
||
|
||
async def _request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
json_body: dict[str, Any] | None = None,
|
||
params: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
try:
|
||
res = await self._client.request(
|
||
method, path, json=json_body, params=params
|
||
)
|
||
except httpx.HTTPError as exc:
|
||
raise AwgError(f"Сеть: {exc}") from exc
|
||
|
||
if res.status_code in (401, 403):
|
||
raise AwgError("Доступ запрещён — проверьте API Token")
|
||
if res.status_code == 404:
|
||
raise AwgError("Не найдено (проверьте server_id на панели)")
|
||
|
||
data: Any
|
||
ctype = res.headers.get("content-type", "")
|
||
if "application/json" in ctype or res.text[:1] in "{[":
|
||
try:
|
||
data = res.json()
|
||
except Exception:
|
||
data = {"raw": res.text[:500]}
|
||
else:
|
||
data = {"raw": res.text[:500]}
|
||
|
||
if res.status_code >= 400:
|
||
err = ""
|
||
if isinstance(data, dict):
|
||
err = str(data.get("error") or data.get("detail") or "")
|
||
raise AwgError(err or f"HTTP {res.status_code}")
|
||
|
||
return data
|
||
|
||
async def ping(self, panel_server_id: int) -> dict[str, Any]:
|
||
return await self._request("GET", f"/api/servers/{int(panel_server_id)}/ping")
|
||
|
||
async def list_panel_servers(self) -> list[AwgPanelServer]:
|
||
"""Список нод панели через backup download (единственный полный inventory API)."""
|
||
try:
|
||
res = await self._client.get("/api/settings/backup/download")
|
||
except httpx.HTTPError as exc:
|
||
raise AwgError(f"Сеть: {exc}") from exc
|
||
if res.status_code in (401, 403):
|
||
raise AwgError("Доступ запрещён — проверьте API Token")
|
||
if res.status_code >= 400:
|
||
raise AwgError(f"Не удалось загрузить список серверов (HTTP {res.status_code})")
|
||
try:
|
||
data = res.json()
|
||
except Exception as exc:
|
||
raise AwgError("Ответ панели не JSON") from exc
|
||
servers = data.get("servers") if isinstance(data, dict) else None
|
||
if not isinstance(servers, list):
|
||
raise AwgError("В backup нет списка servers")
|
||
out: list[AwgPanelServer] = []
|
||
for i, s in enumerate(servers):
|
||
if not isinstance(s, dict):
|
||
continue
|
||
protos = s.get("protocols") or {}
|
||
keys = list(protos.keys()) if isinstance(protos, dict) else []
|
||
out.append(
|
||
AwgPanelServer(
|
||
id=i,
|
||
name=str(s.get("name") or s.get("host") or f"Server #{i}"),
|
||
host=str(s.get("host") or ""),
|
||
protocols=keys,
|
||
)
|
||
)
|
||
return out
|
||
|
||
async def probe(
|
||
self, panel_server_id: int, *, protocol: str = DEFAULT_PROTOCOL
|
||
) -> AwgConnectionInfo:
|
||
try:
|
||
ping = await self.ping(panel_server_id)
|
||
except AwgError as exc:
|
||
return AwgConnectionInfo(
|
||
ok=False, alive=False, ping_ms=None, protocols=[], message=str(exc)
|
||
)
|
||
alive = bool(ping.get("alive"))
|
||
ms = ping.get("ms")
|
||
try:
|
||
ms_int = int(ms) if ms is not None else None
|
||
except (TypeError, ValueError):
|
||
ms_int = None
|
||
protocols: list[str] = []
|
||
try:
|
||
servers = await self.list_panel_servers()
|
||
for s in servers:
|
||
if s.id == int(panel_server_id):
|
||
protocols = s.protocols
|
||
break
|
||
except AwgError:
|
||
protocols = []
|
||
base = (protocol or DEFAULT_PROTOCOL).split("__", 1)[0]
|
||
has_proto = any(str(p).split("__", 1)[0] == base for p in protocols) if protocols else True
|
||
if not alive:
|
||
return AwgConnectionInfo(
|
||
ok=False,
|
||
alive=False,
|
||
ping_ms=ms_int,
|
||
protocols=protocols,
|
||
message=str(ping.get("error") or "SSH недоступен"),
|
||
)
|
||
if protocols and not has_proto:
|
||
return AwgConnectionInfo(
|
||
ok=False,
|
||
alive=True,
|
||
ping_ms=ms_int,
|
||
protocols=protocols,
|
||
message=f"Протокол {base} не найден на сервере",
|
||
)
|
||
return AwgConnectionInfo(
|
||
ok=True,
|
||
alive=True,
|
||
ping_ms=ms_int,
|
||
protocols=protocols,
|
||
message=f"OK · ping {ms_int or '—'} ms",
|
||
)
|
||
|
||
async def list_connections(
|
||
self, panel_server_id: int, *, protocol: str = DEFAULT_PROTOCOL
|
||
) -> list[dict[str, Any]]:
|
||
data = await self._request(
|
||
"GET",
|
||
f"/api/servers/{int(panel_server_id)}/connections",
|
||
params={"protocol": protocol or DEFAULT_PROTOCOL},
|
||
)
|
||
clients = data.get("clients") if isinstance(data, dict) else None
|
||
return clients if isinstance(clients, list) else []
|
||
|
||
async def add_connection(
|
||
self,
|
||
panel_server_id: int,
|
||
*,
|
||
name: str,
|
||
protocol: str = DEFAULT_PROTOCOL,
|
||
) -> dict[str, Any]:
|
||
data = await self._request(
|
||
"POST",
|
||
f"/api/servers/{int(panel_server_id)}/connections/add",
|
||
json_body={
|
||
"protocol": protocol or DEFAULT_PROTOCOL,
|
||
"name": (name or "Client").strip()[:80] or "Client",
|
||
},
|
||
)
|
||
if not isinstance(data, dict):
|
||
raise AwgError("Пустой ответ при создании клиента")
|
||
if not data.get("client_id") and not data.get("config"):
|
||
raise AwgError("Панель не вернула client_id/config")
|
||
return data
|
||
|
||
async def remove_connection(
|
||
self,
|
||
panel_server_id: int,
|
||
*,
|
||
client_id: str,
|
||
protocol: str = DEFAULT_PROTOCOL,
|
||
) -> None:
|
||
await self._request(
|
||
"POST",
|
||
f"/api/servers/{int(panel_server_id)}/connections/remove",
|
||
json_body={
|
||
"protocol": protocol or DEFAULT_PROTOCOL,
|
||
"client_id": client_id,
|
||
},
|
||
)
|
||
|
||
async def get_connection_config(
|
||
self,
|
||
panel_server_id: int,
|
||
*,
|
||
client_id: str,
|
||
protocol: str = DEFAULT_PROTOCOL,
|
||
) -> dict[str, Any]:
|
||
data = await self._request(
|
||
"POST",
|
||
f"/api/servers/{int(panel_server_id)}/connections/config",
|
||
json_body={
|
||
"protocol": protocol or DEFAULT_PROTOCOL,
|
||
"client_id": client_id,
|
||
},
|
||
)
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
async def toggle_connection(
|
||
self,
|
||
panel_server_id: int,
|
||
*,
|
||
client_id: str,
|
||
enable: bool,
|
||
protocol: str = DEFAULT_PROTOCOL,
|
||
) -> None:
|
||
await self._request(
|
||
"POST",
|
||
f"/api/servers/{int(panel_server_id)}/connections/toggle",
|
||
json_body={
|
||
"protocol": protocol or DEFAULT_PROTOCOL,
|
||
"client_id": client_id,
|
||
"enable": bool(enable),
|
||
},
|
||
)
|