"""Маршруты веб-админки и кодов пополнения.""" from __future__ import annotations import json import logging import secrets from datetime import datetime, timedelta, timezone from typing import Any from aiohttp import web from config import Settings from services.db import Database from services.openrouter import ( DEFAULT_MODELS, DEFAULT_SYSTEM_PROMPT, chat_completion, config_public_dict, list_models, load_openrouter_config, parse_ai_decision, ) from services.payments_runtime import ( apply_payment_runtime, apply_payments_body, payments_public_dict, ) from services.smtp_runtime import apply_smtp_body, smtp_public_dict from services.rate_limit import allow from services.tariffs import ( Tariff, TariffError, parse_tariff_payload, replace_shared_tariffs, save_tariffs, ) from services.tickets import ( BODY_MAX, clamp_body, message_to_dict, notify_user_ticket_reply, ticket_to_dict, ) from services.topup_codes import generate_code from services.xui_client import XuiClient, XuiError, build_subscription_link from services.xui_wg import load_wg_purchase_settings, save_wg_purchase_settings from services.awg_client import DEFAULT_PROTOCOL, AwgClient, AwgError from services.awg_configs import load_awg_purchase_settings, save_awg_purchase_settings logger = logging.getLogger(__name__) ADMIN_COOKIE = "vpn_admin_session" ADMIN_SESSION_HOURS = 12 def register_topup_admin_routes(app: web.Application, *, webapp_dir) -> None: if "admin_sessions" not in app: app["admin_sessions"] = {} async def handle_admin_index(_: web.Request) -> web.Response: return web.FileResponse(webapp_dir / "admin.html") app.router.add_get("/admin", handle_admin_index) app.router.add_post("/api/admin/login", handle_admin_login) app.router.add_post("/api/admin/logout", handle_admin_logout) app.router.add_get("/api/admin/me", handle_admin_me) app.router.add_get("/api/admin/stats", handle_admin_stats) app.router.add_get("/api/admin/codes", handle_admin_list_codes) app.router.add_post("/api/admin/codes", handle_admin_create_codes) app.router.add_get("/api/admin/users", handle_admin_list_users) app.router.add_post("/api/admin/users/balance", handle_admin_user_balance) app.router.add_get("/api/admin/tickets", handle_admin_list_tickets) app.router.add_get("/api/admin/tickets/{ticket_id}", handle_admin_get_ticket) app.router.add_post("/api/admin/tickets/{ticket_id}/messages", handle_admin_reply_ticket) app.router.add_post("/api/admin/tickets/{ticket_id}/close", handle_admin_close_ticket) app.router.add_post("/api/admin/tickets/{ticket_id}/reopen", handle_admin_reopen_ticket) app.router.add_get("/api/admin/ai", handle_admin_ai_get) app.router.add_post("/api/admin/ai", handle_admin_ai_save) app.router.add_get("/api/admin/ai/models", handle_admin_ai_models) app.router.add_post("/api/admin/ai/test", handle_admin_ai_test) app.router.add_get("/api/admin/xui/servers", handle_admin_xui_list) app.router.add_post("/api/admin/xui/servers", handle_admin_xui_create) app.router.add_post("/api/admin/xui/test", handle_admin_xui_test) app.router.add_get("/api/admin/xui/servers/{server_id}", handle_admin_xui_get) app.router.add_put("/api/admin/xui/servers/{server_id}", handle_admin_xui_update) app.router.add_delete("/api/admin/xui/servers/{server_id}", handle_admin_xui_delete) app.router.add_post( "/api/admin/xui/servers/{server_id}/test", handle_admin_xui_test_saved, ) app.router.add_get( "/api/admin/xui/servers/{server_id}/inbounds", handle_admin_xui_inbounds, ) app.router.add_get("/api/admin/xui/wg-settings", handle_admin_wg_settings_get) app.router.add_post("/api/admin/xui/wg-settings", handle_admin_wg_settings_save) app.router.add_get("/api/admin/awg/servers", handle_admin_awg_list) app.router.add_post("/api/admin/awg/servers", handle_admin_awg_create) app.router.add_post("/api/admin/awg/test", handle_admin_awg_test) app.router.add_post("/api/admin/awg/discover", handle_admin_awg_discover) app.router.add_get("/api/admin/awg/servers/{server_id}", handle_admin_awg_get) app.router.add_put("/api/admin/awg/servers/{server_id}", handle_admin_awg_update) app.router.add_delete("/api/admin/awg/servers/{server_id}", handle_admin_awg_delete) app.router.add_post( "/api/admin/awg/servers/{server_id}/test", handle_admin_awg_test_saved, ) app.router.add_get("/api/admin/awg/settings", handle_admin_awg_settings_get) app.router.add_post("/api/admin/awg/settings", handle_admin_awg_settings_save) app.router.add_get("/api/admin/tariffs", handle_admin_tariffs_list) app.router.add_post("/api/admin/tariffs", handle_admin_tariffs_create) app.router.add_put("/api/admin/tariffs/{tariff_id}", handle_admin_tariffs_update) app.router.add_delete("/api/admin/tariffs/{tariff_id}", handle_admin_tariffs_delete) app.router.add_get("/api/admin/payments", handle_admin_payments_get) app.router.add_post("/api/admin/payments", handle_admin_payments_save) app.router.add_get("/api/admin/smtp", handle_admin_smtp_get) app.router.add_post("/api/admin/smtp", handle_admin_smtp_save) app.router.add_post("/api/admin/smtp/test", handle_admin_smtp_test) def _utc_now() -> datetime: return datetime.now(timezone.utc) def _admin_enabled(settings: Settings) -> bool: return settings.web_admin_enabled def _safe_secret_compare(got: str, expected: str) -> bool: got_b = got.encode("utf-8") exp_b = expected.encode("utf-8") if len(got_b) != len(exp_b): return False return secrets.compare_digest(got_b, exp_b) def _check_admin_password(settings: Settings, password: str) -> bool: expected = settings.web_admin_password if not expected: return False return _safe_secret_compare((password or "").strip(), expected) def _check_admin_login(settings: Settings, login: str) -> bool: expected = settings.web_admin_login.strip().lower() if not expected: return False return _safe_secret_compare((login or "").strip().lower(), expected) def _admin_token(request: web.Request) -> str | None: auth = request.headers.get("Authorization") or "" if auth.lower().startswith("bearer "): return auth[7:].strip() or None return request.cookies.get(ADMIN_COOKIE) def _admin_authenticated(request: web.Request) -> bool: settings: Settings = request.app["settings"] if not _admin_enabled(settings): return False token = _admin_token(request) if not token: return False sessions: dict[str, datetime] = request.app.get("admin_sessions") or {} expires = sessions.get(token) if not expires or expires <= _utc_now(): sessions.pop(token, None) return False return True def _require_admin(request: web.Request) -> web.Response | None: if not _admin_enabled(request.app["settings"]): return web.json_response({"ok": False, "error": "admin disabled"}, status=503) if not _admin_authenticated(request): return web.json_response({"ok": False, "error": "unauthorized"}, status=401) return None def _set_admin_cookie(response: web.Response, request: web.Request, token: str) -> None: secure = request.headers.get("X-Forwarded-Proto", "").lower() == "https" or ( request.url.scheme == "https" ) response.set_cookie( ADMIN_COOKIE, token, max_age=ADMIN_SESSION_HOURS * 3600, httponly=True, samesite="Lax", secure=secure, path="/", ) async def handle_admin_login(request: web.Request) -> web.Response: settings: Settings = request.app["settings"] if not _admin_enabled(settings): return web.json_response({"ok": False, "error": "admin disabled"}, status=503) if not allow(request, "admin_login", limit=8, window_sec=300): return web.json_response({"ok": False, "error": "Слишком часто"}, status=429) try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) login = str(body.get("login") or "").strip() password = str(body.get("password") or "").strip() login_ok = _check_admin_login(settings, login) pass_ok = _check_admin_password(settings, password) if not login_ok or not pass_ok: logger.warning( "Web admin login failed (login_ok=%s pass_ok=%s, configured_login=%r)", login_ok, pass_ok, settings.web_admin_login, ) return web.json_response({"ok": False, "error": "Неверный логин или пароль"}, status=403) token = secrets.token_urlsafe(32) sessions: dict[str, datetime] = request.app["admin_sessions"] sessions[token] = _utc_now() + timedelta(hours=ADMIN_SESSION_HOURS) resp = web.json_response({"ok": True, "token": token}) _set_admin_cookie(resp, request, token) return resp async def handle_admin_logout(request: web.Request) -> web.Response: token = _admin_token(request) if token: sessions: dict[str, datetime] = request.app.get("admin_sessions") or {} sessions.pop(token, None) resp = web.json_response({"ok": True}) resp.del_cookie(ADMIN_COOKIE, path="/") return resp async def handle_admin_me(request: web.Request) -> web.Response: settings: Settings = request.app["settings"] return web.json_response( { "ok": True, "enabled": _admin_enabled(settings), "authenticated": _admin_authenticated(request), } ) async def handle_admin_stats(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] return web.json_response( { "ok": True, "stats": db.topup_code_stats(), "tickets": db.ticket_stats(), } ) async def handle_admin_list_codes(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied status = request.rel_url.query.get("status") or "all" try: limit = int(request.rel_url.query.get("limit") or "50") offset = int(request.rel_url.query.get("offset") or "0") except ValueError: limit, offset = 50, 0 db: Database = request.app["db"] codes = db.list_topup_codes(status=status, limit=limit, offset=offset) return web.json_response({"ok": True, "codes": _serialize_codes(codes)}) async def handle_admin_create_codes(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) try: amount = float(body.get("amount") or 0) count = int(body.get("count") or 1) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad amount/count"}, status=400) if amount <= 0: return web.json_response({"ok": False, "error": "amount must be > 0"}, status=400) count = max(1, min(count, 100)) note = str(body.get("note") or "web_admin")[:120] codes: list[str] = [] seen: set[str] = set() while len(codes) < count: code = generate_code() if code in seen: continue seen.add(code) codes.append(code) db: Database = request.app["db"] inserted = db.create_topup_codes( amount_rub=amount, count=count, created_by=note, codes=codes, ) return web.json_response( { "ok": True, "inserted": inserted, "codes": codes[:inserted], "amount": amount, } ) async def handle_admin_list_users(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied q = request.rel_url.query.get("q") or "" try: limit = int(request.rel_url.query.get("limit") or "50") offset = int(request.rel_url.query.get("offset") or "0") except ValueError: limit, offset = 50, 0 db: Database = request.app["db"] rows, total = db.list_users_admin(query=q, limit=limit, offset=offset) users = [] for row in rows: tid = int(row["telegram_id"]) name = row.get("full_name") or row.get("login") or row.get("username") if row.get("username") and not str(name or "").startswith("@"): uname = row["username"] else: uname = row.get("username") users.append( { "telegram_id": tid, "login": row.get("login"), "username": uname, "full_name": row.get("full_name"), "display_name": name or f"ID {tid}", "balance": float(row["balance"]), "created_at": _iso(row.get("created_at")), "is_telegram": tid > 0, "banned": bool(row.get("banned")), } ) return web.json_response( {"ok": True, "users": users, "total": total, "limit": limit, "offset": offset} ) async def handle_admin_user_balance(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) try: telegram_id = int(body.get("telegram_id")) amount = float(body.get("amount") or 0) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad telegram_id/amount"}, status=400) comment = str(body.get("comment") or "").strip() if amount <= 0: return web.json_response({"ok": False, "error": "amount must be > 0"}, status=400) if not comment: return web.json_response( {"ok": False, "error": "Укажите комментарий для пользователя"}, status=400, ) db: Database = request.app["db"] user = db.get_user(telegram_id) if user is None: db.ensure_user(telegram_id) user = db.get_user(telegram_id) if user is None: return web.json_response({"ok": False, "error": "user not found"}, status=404) try: balance = db.admin_credit_balance( telegram_id, amount, comment=comment[:200], ) except ValueError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) bot = request.app.get("bot") if bot is not None and telegram_id > 0: try: await bot.send_message( telegram_id, ( "💰 Баланс пополнен\n\n" f"Зачислено: +{amount:.0f} ₽\n" f"Комментарий: {comment[:200]}\n" f"Текущий баланс: {balance:.0f} ₽" ), parse_mode="HTML", ) except Exception: # noqa: BLE001 logger.warning("admin credit notify failed tg=%s", telegram_id) logger.info( "Admin credit tg=%s amount=%s comment=%r", telegram_id, amount, comment[:80], ) return web.json_response( { "ok": True, "telegram_id": telegram_id, "balance": balance, "amount": amount, "comment": comment[:200], } ) async def handle_admin_list_tickets(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied status = request.rel_url.query.get("status") or "open" try: limit = int(request.rel_url.query.get("limit") or "50") offset = int(request.rel_url.query.get("offset") or "0") except ValueError: limit, offset = 50, 0 db: Database = request.app["db"] rows, total = db.list_tickets_admin(status=status, limit=limit, offset=offset) return web.json_response( { "ok": True, "tickets": [ticket_to_dict(r, include_user=True) for r in rows], "total": total, "stats": db.ticket_stats(), "body_max": BODY_MAX, } ) async def handle_admin_get_ticket(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: tid = int(request.match_info["ticket_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] ticket = db.get_ticket(tid) if not ticket: return web.json_response({"ok": False, "error": "not found"}, status=404) messages = db.list_ticket_messages(tid) return web.json_response( { "ok": True, "ticket": ticket_to_dict( {**ticket, "message_count": len(messages)}, include_user=True, ), "messages": [message_to_dict(m) for m in messages], } ) async def handle_admin_reply_ticket(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: tid = int(request.match_info["ticket_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) text = clamp_body(str(body.get("body") or body.get("message") or "")) if len(text) < 1: return web.json_response({"ok": False, "error": "Пустое сообщение"}, status=400) db: Database = request.app["db"] ticket = db.get_ticket(tid) if not ticket: return web.json_response({"ok": False, "error": "not found"}, status=404) try: ticket = db.add_ticket_message( tid, author_role="admin", author_id=None, body=text, ) except ValueError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) bot = request.app.get("bot") if bot is not None: await notify_user_ticket_reply( bot, telegram_id=int(ticket["telegram_id"]), ticket_id=tid, subject=str(ticket.get("subject") or ""), body=text, ) messages = db.list_ticket_messages(tid) return web.json_response( { "ok": True, "ticket": ticket_to_dict( {**ticket, "message_count": len(messages)}, include_user=True, ), "messages": [message_to_dict(m) for m in messages], } ) async def handle_admin_close_ticket(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: tid = int(request.match_info["ticket_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] ticket = db.set_ticket_status(tid, "closed") if not ticket: return web.json_response({"ok": False, "error": "not found"}, status=404) messages = db.list_ticket_messages(tid) return web.json_response( { "ok": True, "ticket": ticket_to_dict( {**ticket, "message_count": len(messages)}, include_user=True, ), "messages": [message_to_dict(m) for m in messages], } ) async def handle_admin_reopen_ticket(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: tid = int(request.match_info["ticket_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] ticket = db.set_ticket_status(tid, "open") if not ticket: return web.json_response({"ok": False, "error": "not found"}, status=404) messages = db.list_ticket_messages(tid) return web.json_response( { "ok": True, "ticket": ticket_to_dict( {**ticket, "message_count": len(messages)}, include_user=True, ), "messages": [message_to_dict(m) for m in messages], } ) async def handle_admin_ai_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] cfg = load_openrouter_config(db) return web.json_response({"ok": True, "ai": config_public_dict(cfg)}) async def handle_admin_ai_save(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) db: Database = request.app["db"] data: dict[str, str] = {} if "enabled" in body: data["openrouter_enabled"] = "true" if bool(body.get("enabled")) else "false" if "auto_reply" in body: data["openrouter_auto_reply"] = ( "true" if bool(body.get("auto_reply")) else "false" ) if "api_key" in body: key = str(body.get("api_key") or "").strip() if key == "__clear__": data["openrouter_api_key"] = "" elif key and "…" not in key and "•" not in key and "..." not in key: data["openrouter_api_key"] = key[:200] if "model" in body: model = str(body.get("model") or "").strip() if model: data["openrouter_model"] = model[:120] if "models" in body: models_raw = body.get("models") models: list[str] = [] if isinstance(models_raw, list): models = [str(x).strip() for x in models_raw if str(x).strip()] elif isinstance(models_raw, str): models = [x.strip() for x in models_raw.split(",") if x.strip()] # unique keep order seen: set[str] = set() uniq: list[str] = [] for m in models: if m not in seen: seen.add(m) uniq.append(m[:120]) data["openrouter_models"] = json.dumps(uniq[:20], ensure_ascii=False) if "system_prompt" in body: prompt = str(body.get("system_prompt") or "").strip() data["openrouter_system_prompt"] = prompt[:8000] if prompt else DEFAULT_SYSTEM_PROMPT if data: db.set_settings_map(data) cfg = load_openrouter_config(db) # если активная модель не в списке — добавим if cfg.model and cfg.model not in cfg.models: models = [cfg.model, *cfg.models][:20] db.set_setting("openrouter_models", json.dumps(models, ensure_ascii=False)) cfg = load_openrouter_config(db) return web.json_response({"ok": True, "ai": config_public_dict(cfg)}) async def handle_admin_ai_models(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] cfg = load_openrouter_config(db) settings: Settings = request.app["settings"] if not cfg.api_key: return web.json_response( { "ok": True, "models": [{"id": m, "name": m} for m in DEFAULT_MODELS], "source": "defaults", "error": "Сначала сохраните API-ключ OpenRouter", } ) try: models = await list_models( cfg.api_key, referer=(settings.webapp_url or "").rstrip("/") or "https://bot.example.com", ) return web.json_response( {"ok": True, "models": models, "source": "openrouter", "total": len(models)} ) except Exception as exc: # noqa: BLE001 logger.exception("OpenRouter models list failed") return web.json_response( { "ok": False, "error": str(exc), "models": [{"id": m, "name": m} for m in DEFAULT_MODELS], "source": "defaults", }, status=400, ) async def handle_admin_ai_test(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] cfg = load_openrouter_config(db) if not cfg.api_key: return web.json_response( {"ok": False, "error": "Нет API-ключа"}, status=400, ) try: body = await request.json() except json.JSONDecodeError: body = {} prompt = str(body.get("prompt") or "Как подключить VPN? Ответь коротко.").strip() settings: Settings = request.app["settings"] try: raw, used = await chat_completion( api_key=cfg.api_key, model=cfg.model, models=cfg.models_for_request(), messages=[ {"role": "system", "content": cfg.system_prompt}, {"role": "user", "content": prompt}, ], referer=(settings.webapp_url or "").rstrip("/") or "https://bot.example.com", max_tokens=400, ) decision = parse_ai_decision(raw, used_model=used) return web.json_response( { "ok": True, "model": used, "raw": raw[:2000], "action": decision.action, "reply": decision.reply or decision.reason, } ) except Exception as exc: # noqa: BLE001 logger.exception("OpenRouter test failed") return web.json_response({"ok": False, "error": str(exc)}, status=400) # ─── 3x-ui servers ─────────────────────────────────────────────────────────── def _serialize_xui_server(row: dict[str, Any], *, hide_password: bool = True) -> dict[str, Any]: sub = str(row.get("subscription_url") or "").strip() return { "id": int(row["id"]), "name": row.get("name") or "", "api_url": row.get("api_url") or "", "username": row.get("username") or "", "password_set": bool(row.get("password")), "password": "" if hide_password else (row.get("password") or ""), "api_key_set": bool(str(row.get("api_key") or "").strip()), "api_key": "" if hide_password else (row.get("api_key") or ""), "inbound_id": row.get("inbound_id"), "inbound_remark": row.get("inbound_remark") or "", "subscription_url": sub, "subscription_preview": build_subscription_link(sub, "exampleSubId") if sub else "", "enabled": bool(row.get("enabled", True)), "verify_ssl": bool(row.get("verify_ssl", True)), "last_check_at": _iso(row.get("last_check_at")), "last_check_ok": row.get("last_check_ok"), "last_check_msg": row.get("last_check_msg") or "", "created_at": _iso(row.get("created_at")), "updated_at": _iso(row.get("updated_at")), "panel": "3x-ui", "panel_version": "3.5.0", "protocol": "wireguard", } def _parse_inbound_id(value: Any) -> int | None: if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None def _xui_client_from_fields( *, api_url: str, username: str = "", password: str = "", api_key: str = "", verify_ssl: bool = True, ) -> XuiClient: return XuiClient( api_url, username=username, password=password, api_key=api_key, verify_ssl=verify_ssl, ) async def _xui_probe( *, api_url: str, username: str = "", password: str = "", api_key: str = "", verify_ssl: bool = True, ) -> dict[str, Any]: async with _xui_client_from_fields( api_url=api_url, username=username, password=password, api_key=api_key, verify_ssl=verify_ssl, ) as client: info = await client.test_connection() return { "ok": info.ok, "message": info.message, "version": info.version, "inbounds_total": info.inbounds_total, "wireguard": [ { "id": i.id, "remark": i.remark, "protocol": i.protocol, "port": i.port, "enable": i.enable, "clients_count": i.clients_count, } for i in info.wireguard ], } async def handle_admin_xui_list(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] rows = db.list_xui_servers() return web.json_response( { "ok": True, "servers": [_serialize_xui_server(r) for r in rows], "panel_version": "3.5.0", } ) async def handle_admin_xui_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] row = db.get_xui_server(server_id) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) return web.json_response({"ok": True, "server": _serialize_xui_server(row)}) async def handle_admin_xui_create(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) name = str(body.get("name") or "").strip() api_url = str(body.get("api_url") or "").strip() username = str(body.get("username") or "").strip() password = str(body.get("password") or "") api_key = str(body.get("api_key") or "").strip() subscription_url = str(body.get("subscription_url") or "").strip() if not name or not api_url: return web.json_response( {"ok": False, "error": "Нужны name и api_url"}, status=400, ) if not api_key and (not username or not password): return web.json_response( {"ok": False, "error": "Нужен API Token или логин+пароль"}, status=400, ) inbound_id = _parse_inbound_id(body.get("inbound_id")) inbound_remark = str(body.get("inbound_remark") or "").strip() or None enabled = bool(body.get("enabled", True)) verify_ssl = bool(body.get("verify_ssl", True)) db: Database = request.app["db"] row = db.create_xui_server( name=name, api_url=api_url, username=username, password=password, api_key=api_key, subscription_url=subscription_url, inbound_id=inbound_id, inbound_remark=inbound_remark, enabled=enabled, verify_ssl=verify_ssl, ) return web.json_response({"ok": True, "server": _serialize_xui_server(row)}) async def handle_admin_xui_update(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) db: Database = request.app["db"] existing = db.get_xui_server(server_id) if not existing: return web.json_response({"ok": False, "error": "not found"}, status=404) fields: dict[str, Any] = {} if "name" in body: name = str(body.get("name") or "").strip() if not name: return web.json_response({"ok": False, "error": "name пуст"}, status=400) fields["name"] = name if "api_url" in body: api_url = str(body.get("api_url") or "").strip() if not api_url: return web.json_response({"ok": False, "error": "api_url пуст"}, status=400) fields["api_url"] = api_url if "username" in body: fields["username"] = str(body.get("username") or "").strip() if "password" in body: password = str(body.get("password") or "") if password: fields["password"] = password if "api_key" in body: api_key = str(body.get("api_key") or "").strip() if api_key: fields["api_key"] = api_key if "subscription_url" in body: fields["subscription_url"] = str(body.get("subscription_url") or "").strip() if "inbound_id" in body: fields["inbound_id"] = _parse_inbound_id(body.get("inbound_id")) if "inbound_remark" in body: fields["inbound_remark"] = str(body.get("inbound_remark") or "").strip() or None if "enabled" in body: fields["enabled"] = bool(body.get("enabled")) if "verify_ssl" in body: fields["verify_ssl"] = bool(body.get("verify_ssl")) row = db.update_xui_server(server_id, **fields) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) return web.json_response({"ok": True, "server": _serialize_xui_server(row)}) async def handle_admin_xui_delete(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] deleted = db.delete_xui_server(server_id) if not deleted: return web.json_response({"ok": False, "error": "not found"}, status=404) return web.json_response({"ok": True}) async def handle_admin_xui_test(request: web.Request) -> web.Response: """Проверка панели по полям формы (ещё не сохранено).""" denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) api_url = str(body.get("api_url") or "").strip() username = str(body.get("username") or "").strip() password = str(body.get("password") or "") api_key = str(body.get("api_key") or "").strip() verify_ssl = bool(body.get("verify_ssl", True)) if not api_url: return web.json_response( {"ok": False, "error": "Нужен api_url"}, status=400, ) if not api_key and (not username or not password): return web.json_response( {"ok": False, "error": "Нужен API Token или логин+пароль"}, status=400, ) try: result = await _xui_probe( api_url=api_url, username=username, password=password, api_key=api_key, verify_ssl=verify_ssl, ) return web.json_response({"ok": True, **result}) except XuiError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) except Exception as exc: # noqa: BLE001 logger.exception("3x-ui test failed") return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_xui_test_saved(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] row = db.get_xui_server(server_id) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) try: result = await _xui_probe( api_url=str(row["api_url"]), username=str(row.get("username") or ""), password=str(row.get("password") or ""), api_key=str(row.get("api_key") or ""), verify_ssl=bool(row.get("verify_ssl", True)), ) db.set_xui_server_check( server_id, ok=bool(result.get("ok")), message=str(result.get("message") or ""), ) refreshed = db.get_xui_server(server_id) return web.json_response( { "ok": True, **result, "server": _serialize_xui_server(refreshed) if refreshed else None, } ) except Exception as exc: # noqa: BLE001 logger.exception("3x-ui saved test failed") db.set_xui_server_check(server_id, ok=False, message=str(exc)) return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_xui_inbounds(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] row = db.get_xui_server(server_id) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) protocol = (request.rel_url.query.get("protocol") or "wireguard").strip().lower() try: async with _xui_client_from_fields( api_url=str(row["api_url"]), username=str(row.get("username") or ""), password=str(row.get("password") or ""), api_key=str(row.get("api_key") or ""), verify_ssl=bool(row.get("verify_ssl", True)), ) as client: await client.ensure_auth() if protocol in {"wireguard", "wg"}: items = await client.list_wireguard_inbounds() else: items = await client.list_inbounds() return web.json_response( { "ok": True, "inbounds": [ { "id": i.id, "remark": i.remark, "protocol": i.protocol, "port": i.port, "enable": i.enable, "clients_count": i.clients_count, "is_wireguard": i.is_wireguard, } for i in items ], } ) except XuiError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) except Exception as exc: # noqa: BLE001 logger.exception("3x-ui inbounds failed") return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_wg_settings_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] settings = load_wg_purchase_settings(db) return web.json_response({"ok": True, **settings.as_dict()}) async def handle_admin_wg_settings_save(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) enabled = body.get("enabled") price = body.get("price") kwargs: dict[str, Any] = {} if enabled is not None: kwargs["enabled"] = bool(enabled) if price is not None: try: kwargs["price"] = float(price) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad price"}, status=400) db: Database = request.app["db"] settings = save_wg_purchase_settings(db, **kwargs) return web.json_response({"ok": True, **settings.as_dict()}) def _serialize_awg_server(row: dict[str, Any], *, hide_token: bool = True) -> dict[str, Any]: token = str(row.get("api_token") or "") return { "id": int(row["id"]), "name": row.get("name") or "", "api_url": row.get("api_url") or "", "api_token_set": bool(token.strip()), "api_token_masked": (token[:8] + "…") if hide_token and len(token) > 8 else ("" if hide_token else token), "panel_server_id": int(row.get("panel_server_id") or 0), "protocol": row.get("protocol") or DEFAULT_PROTOCOL, "enabled": bool(row.get("enabled", True)), "verify_ssl": bool(row.get("verify_ssl", True)), "last_check_at": _iso(row.get("last_check_at")), "last_check_ok": row.get("last_check_ok"), "last_check_msg": row.get("last_check_msg") or "", "created_at": _iso(row.get("created_at")), "updated_at": _iso(row.get("updated_at")), } async def handle_admin_awg_list(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] rows = db.list_awg_servers() return web.json_response( {"ok": True, "servers": [_serialize_awg_server(r) for r in rows]} ) async def handle_admin_awg_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] row = db.get_awg_server(server_id) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) return web.json_response({"ok": True, "server": _serialize_awg_server(row)}) async def handle_admin_awg_create(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) name = str(body.get("name") or "").strip() api_url = str(body.get("api_url") or "").strip() api_token = str(body.get("api_token") or "").strip() if not name or not api_url or not api_token: return web.json_response( {"ok": False, "error": "Нужны название, API URL и API Token"}, status=400, ) try: panel_server_id = int(body.get("panel_server_id") or 0) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad panel_server_id"}, status=400) protocol = str(body.get("protocol") or DEFAULT_PROTOCOL).strip() or DEFAULT_PROTOCOL db: Database = request.app["db"] row = db.create_awg_server( name=name, api_url=api_url, api_token=api_token, panel_server_id=panel_server_id, protocol=protocol, enabled=bool(body.get("enabled", True)), verify_ssl=bool(body.get("verify_ssl", True)), ) return web.json_response({"ok": True, "server": _serialize_awg_server(row)}) async def handle_admin_awg_update(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) db: Database = request.app["db"] existing = db.get_awg_server(server_id) if not existing: return web.json_response({"ok": False, "error": "not found"}, status=404) fields: dict[str, Any] = {} if "name" in body: fields["name"] = str(body.get("name") or "").strip() if "api_url" in body: fields["api_url"] = str(body.get("api_url") or "").strip() if "api_token" in body: token = str(body.get("api_token") or "").strip() if token: fields["api_token"] = token if "panel_server_id" in body: try: fields["panel_server_id"] = int(body.get("panel_server_id") or 0) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad panel_server_id"}, status=400) if "protocol" in body: fields["protocol"] = str(body.get("protocol") or DEFAULT_PROTOCOL).strip() or DEFAULT_PROTOCOL if "enabled" in body: fields["enabled"] = bool(body.get("enabled")) if "verify_ssl" in body: fields["verify_ssl"] = bool(body.get("verify_ssl")) row = db.update_awg_server(server_id, **fields) return web.json_response({"ok": True, "server": _serialize_awg_server(row) if row else None}) async def handle_admin_awg_delete(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] deleted = db.delete_awg_server(server_id) if not deleted: return web.json_response({"ok": False, "error": "not found"}, status=404) return web.json_response({"ok": True}) async def handle_admin_awg_discover(request: web.Request) -> web.Response: """Список нод с панели (backup inventory).""" denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) api_url = str(body.get("api_url") or "").strip() api_token = str(body.get("api_token") or "").strip() verify_ssl = bool(body.get("verify_ssl", True)) if not api_url or not api_token: return web.json_response( {"ok": False, "error": "Укажите API URL и Token"}, status=400 ) try: async with AwgClient(api_url, api_token=api_token, verify_ssl=verify_ssl) as client: servers = await client.list_panel_servers() return web.json_response( { "ok": True, "servers": [ { "id": s.id, "name": s.name, "host": s.host, "protocols": s.protocols, "has_awg2": s.has_awg2, } for s in servers ], } ) except AwgError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) except Exception as exc: # noqa: BLE001 logger.exception("AWG discover failed") return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_awg_test(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) api_url = str(body.get("api_url") or "").strip() api_token = str(body.get("api_token") or "").strip() try: panel_server_id = int(body.get("panel_server_id") or 0) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad panel_server_id"}, status=400) protocol = str(body.get("protocol") or DEFAULT_PROTOCOL).strip() or DEFAULT_PROTOCOL verify_ssl = bool(body.get("verify_ssl", True)) try: async with AwgClient(api_url, api_token=api_token, verify_ssl=verify_ssl) as client: result = await client.probe(panel_server_id, protocol=protocol) return web.json_response( { "ok": result.ok, "alive": result.alive, "ping_ms": result.ping_ms, "protocols": result.protocols, "message": result.message, } ) except AwgError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) except Exception as exc: # noqa: BLE001 logger.exception("AWG test failed") return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_awg_test_saved(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: server_id = int(request.match_info["server_id"]) except (KeyError, ValueError): return web.json_response({"ok": False, "error": "bad id"}, status=400) db: Database = request.app["db"] row = db.get_awg_server(server_id) if not row: return web.json_response({"ok": False, "error": "not found"}, status=404) try: async with AwgClient( str(row["api_url"]), api_token=str(row.get("api_token") or ""), verify_ssl=bool(row.get("verify_ssl", True)), ) as client: result = await client.probe( int(row.get("panel_server_id") or 0), protocol=str(row.get("protocol") or DEFAULT_PROTOCOL), ) db.set_awg_server_check(server_id, ok=result.ok, message=result.message) refreshed = db.get_awg_server(server_id) return web.json_response( { "ok": result.ok, "alive": result.alive, "ping_ms": result.ping_ms, "protocols": result.protocols, "message": result.message, "server": _serialize_awg_server(refreshed) if refreshed else None, } ) except Exception as exc: # noqa: BLE001 db.set_awg_server_check(server_id, ok=False, message=str(exc)) return web.json_response({"ok": False, "error": str(exc)}, status=400) async def handle_admin_awg_settings_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied db: Database = request.app["db"] settings = load_awg_purchase_settings(db) return web.json_response({"ok": True, **settings.as_dict()}) async def handle_admin_awg_settings_save(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) enabled = body.get("enabled") price = body.get("price") kwargs: dict[str, Any] = {} if enabled is not None: kwargs["enabled"] = bool(enabled) if price is not None: try: kwargs["price"] = float(price) except (TypeError, ValueError): return web.json_response({"ok": False, "error": "bad price"}, status=400) db: Database = request.app["db"] settings = save_awg_purchase_settings(db, **kwargs) return web.json_response({"ok": True, **settings.as_dict()}) def _serialize_codes(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for row in rows: out.append( { "code": row["code"], "amount": float(row["amount_rub"]), "created_at": _iso(row.get("created_at")), "created_by": row.get("created_by"), "used_at": _iso(row.get("used_at")), "used_by": row.get("used_by"), } ) return out def _iso(value: Any) -> str | None: if value is None: return None if isinstance(value, datetime): return value.isoformat() return str(value) def _tariffs_path(request: web.Request) -> str: settings: Settings = request.app["settings"] return settings.tariffs_path def _persist_tariffs(request: web.Request, tariffs: list[Tariff]) -> None: path = _tariffs_path(request) save_tariffs(path, tariffs) shared: list[Tariff] = request.app["tariffs"] replace_shared_tariffs(shared, tariffs) async def handle_admin_tariffs_list(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied tariffs: list[Tariff] = request.app["tariffs"] return web.json_response( {"ok": True, "tariffs": [t.to_admin_dict() for t in tariffs]} ) async def handle_admin_tariffs_create(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) shared: list[Tariff] = list(request.app["tariffs"]) default_squads = shared[0].squad_uuids if shared else () try: tariff = parse_tariff_payload(body, default_squads=default_squads) except TariffError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) if any(t.id == tariff.id for t in shared): return web.json_response( {"ok": False, "error": f"Тариф «{tariff.id}» уже есть"}, status=400, ) shared.append(tariff) try: _persist_tariffs(request, shared) except OSError as exc: logger.exception("tariffs save failed") return web.json_response( {"ok": False, "error": f"Не удалось записать tariffs.json: {exc}"}, status=500, ) return web.json_response({"ok": True, "tariff": tariff.to_admin_dict()}) async def handle_admin_tariffs_update(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied tariff_id = str(request.match_info.get("tariff_id") or "").strip() try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) shared: list[Tariff] = list(request.app["tariffs"]) existing = next((t for t in shared if t.id == tariff_id), None) if existing is None: return web.json_response({"ok": False, "error": "Тариф не найден"}, status=404) payload = dict(body) payload["id"] = tariff_id try: updated = parse_tariff_payload(payload, existing=existing) except TariffError as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) shared = [updated if t.id == tariff_id else t for t in shared] try: _persist_tariffs(request, shared) except OSError as exc: logger.exception("tariffs save failed") return web.json_response( {"ok": False, "error": f"Не удалось записать tariffs.json: {exc}"}, status=500, ) return web.json_response({"ok": True, "tariff": updated.to_admin_dict()}) async def handle_admin_tariffs_delete(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied tariff_id = str(request.match_info.get("tariff_id") or "").strip() shared: list[Tariff] = list(request.app["tariffs"]) if not any(t.id == tariff_id for t in shared): return web.json_response({"ok": False, "error": "Тариф не найден"}, status=404) if len(shared) <= 1: return web.json_response( {"ok": False, "error": "Нельзя удалить последний тариф"}, status=400, ) shared = [t for t in shared if t.id != tariff_id] try: _persist_tariffs(request, shared) except OSError as exc: logger.exception("tariffs save failed") return web.json_response( {"ok": False, "error": f"Не удалось записать tariffs.json: {exc}"}, status=500, ) return web.json_response({"ok": True}) async def handle_admin_payments_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied settings: Settings = request.app["settings"] return web.json_response({"ok": True, "payments": payments_public_dict(settings)}) async def handle_admin_payments_save(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) try: settings = apply_payments_body(body if isinstance(body, dict) else {}) apply_payment_runtime(request.app, settings) except OSError as exc: logger.exception("payments .env write failed") return web.json_response( {"ok": False, "error": f"Не удалось записать .env: {exc}"}, status=500, ) except SystemExit as exc: return web.json_response( {"ok": False, "error": str(exc) or "Ошибка настроек"}, status=400, ) except Exception as exc: # noqa: BLE001 logger.exception("payments save failed") return web.json_response({"ok": False, "error": str(exc)}, status=500) return web.json_response( {"ok": True, "payments": payments_public_dict(settings)} ) async def handle_admin_smtp_get(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied settings: Settings = request.app["settings"] return web.json_response({"ok": True, "smtp": smtp_public_dict(settings)}) async def handle_admin_smtp_save(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied try: body = await request.json() except json.JSONDecodeError: return web.json_response({"ok": False, "error": "invalid json"}, status=400) try: settings = apply_smtp_body(body if isinstance(body, dict) else {}) request.app["settings"] = settings inject = request.app.get("inject") if inject is not None: inject.settings = settings except OSError as exc: logger.exception("smtp .env write failed") return web.json_response( {"ok": False, "error": f"Не удалось записать .env: {exc}"}, status=500, ) except SystemExit as exc: return web.json_response( {"ok": False, "error": str(exc) or "Ошибка настроек"}, status=400, ) except Exception as exc: # noqa: BLE001 logger.exception("smtp save failed") return web.json_response({"ok": False, "error": str(exc)}, status=500) return web.json_response({"ok": True, "smtp": smtp_public_dict(settings)}) async def handle_admin_smtp_test(request: web.Request) -> web.Response: denied = _require_admin(request) if denied: return denied settings: Settings = request.app["settings"] try: body = await request.json() except json.JSONDecodeError: body = {} to = str((body or {}).get("to") or settings.smtp_from or "").strip() if not to: return web.json_response( {"ok": False, "error": "Укажите email получателя"}, status=400 ) from services.mailer import render_smtp_test_email, send_email, smtp_configured if not smtp_configured(settings): return web.json_response( {"ok": False, "error": "SMTP не настроен"}, status=400 ) try: await send_email( settings, to=to, subject="Тест SMTP — VPN Service", html=render_smtp_test_email(webapp_url=settings.webapp_url or ""), ) except Exception as exc: # noqa: BLE001 logger.exception("smtp test failed") return web.json_response({"ok": False, "error": str(exc)}, status=500) return web.json_response({"ok": True})