"""HTTP-сервер Telegram Mini App + веб-кабинет (aiohttp)."""
from __future__ import annotations
import base64
import json
import logging
from pathlib import Path
from typing import Any
from aiohttp import web
from webauthn.helpers import bytes_to_base64url
from config import Settings
from services.clients import TOP_CLIENTS
from services.cryptopay import CryptoPay
from services.db import AuthError, BotUser, Database, SESSION_DAYS
from services.digiseller import Digiseller, DigisellerError
from services.digiseller_fulfill import fulfill_digiseller_unique_code
from services.faq import FAQ_ITEMS
from services.heleket import Heleket, verify_webhook_sign
from services.anypay import (
Anypay,
PAID_STATUSES as ANYPAY_PAID,
client_ip_allowed,
verify_notification_sign as verify_anypay_sign,
)
from services.lava import Lava, verify_webhook as verify_lava_webhook
from services.nodes_status import fetch_nodes_status
from services.payments import fulfill_paid_invoice
from services.purchase import (
get_my_subscription_info,
purchase_device_pack,
purchase_tariff,
purchase_traffic_pack,
)
from services.xui_wg import (
delete_wg_config,
enforce_wg_configs_subscription,
list_available_servers,
list_user_wg_configs,
load_wg_purchase_settings,
purchase_wg_config,
set_wg_config_enabled,
)
from services.awg_configs import (
delete_awg_config,
enforce_awg_configs_subscription,
list_available_awg_servers,
list_user_awg_configs,
load_awg_purchase_settings,
purchase_awg_config,
set_awg_config_enabled,
)
from services.qrcode_svg import make_qr_svg
from services.tariffs import Tariff, get_tariff
from services.devices import DEVICE_PACKS, get_device_pack
from services.traffic import TRAFFIC_PACKS, get_traffic_pack
from services.extras_billing import extras_as_dict
from services.stars import create_stars_topup, packs_payload
from services.topup_code_redeem import TopupCodeError, redeem_topup_code
from services.web_topup import (
TopupError,
create_web_topup_invoice,
topup_methods_payload,
)
from services.transactions_ui import transaction_label, transaction_row_dict
from services.webauthn_auth import (
build_authentication_options,
build_registration_options,
challenge_from_client_data,
credential_id_b64,
credential_id_bytes,
parse_transports,
verify_authentication,
verify_registration,
)
from services.tickets import (
BODY_MAX,
MAX_OPEN_PER_USER,
SUBJECT_MAX,
clamp_body,
clamp_subject,
message_to_dict,
notify_admins_new_ticket,
notify_admins_ticket_reply,
ticket_to_dict,
)
from services.ticket_ai import try_ai_ticket_reply
from services.tg_webapp import (
InitDataError,
parse_and_validate_id_token,
parse_and_validate_init_data,
parse_and_validate_login_widget,
)
from webapp_admin_routes import register_topup_admin_routes
from services.webapp_growth_routes import register_growth_routes
from services.rate_limit import allow, client_ip
logger = logging.getLogger(__name__)
WEBAPP_DIR = Path(__file__).resolve().parent / "webapp"
SESSION_COOKIE = "vpn_session"
def create_webapp(
*,
settings: Settings,
db: Database,
sdk: Any,
tariffs: list[Tariff],
cryptopay: CryptoPay,
heleket: Heleket | None = None,
digiseller: Digiseller | None = None,
lava: Lava | None = None,
anypay: Anypay | None = None,
bot: Any | None = None,
bot_username: str = "",
) -> web.Application:
app = web.Application()
app["settings"] = settings
app["db"] = db
app["sdk"] = sdk
app["tariffs"] = tariffs
app["cryptopay"] = cryptopay
app["heleket"] = heleket or Heleket("", "")
app["digiseller"] = digiseller or Digiseller("", "")
app["lava"] = lava or Lava("", "")
app["anypay"] = anypay or Anypay("", "", "")
app["bot"] = bot
app["bot_username"] = bot_username.lstrip("@")
app.router.add_get("/api/health", handle_health)
app.router.add_get("/api/config", handle_config)
app.router.add_post("/api/auth/register", handle_register)
app.router.add_post("/api/auth/login", handle_login)
app.router.add_post("/api/auth/telegram", handle_telegram_auth)
app.router.add_post("/api/auth/logout", handle_logout)
app.router.add_post("/api/auth/qr/start", handle_qr_auth_start)
app.router.add_get("/api/auth/qr/poll", handle_qr_auth_poll)
app.router.add_post("/api/auth/qr/approve", handle_qr_auth_approve)
app.router.add_post("/api/auth/passkey/register/options", handle_passkey_register_options)
app.router.add_post("/api/auth/passkey/register/verify", handle_passkey_register_verify)
app.router.add_post("/api/auth/passkey/login/options", handle_passkey_login_options)
app.router.add_post("/api/auth/passkey/login/verify", handle_passkey_login_verify)
app.router.add_get("/api/auth/passkey/list", handle_passkey_list)
app.router.add_delete("/api/auth/passkey/{credential_id}", handle_passkey_delete)
app.router.add_post("/api/account/email", handle_account_email)
app.router.add_get("/api/me", handle_me)
app.router.add_post("/api/auto-renew", handle_auto_renew)
app.router.add_get("/api/transactions", handle_transactions)
app.router.add_get("/api/tariffs", handle_tariffs)
app.router.add_post("/api/buy", handle_buy)
app.router.add_get("/api/traffic/packs", handle_traffic_packs)
app.router.add_post("/api/traffic/buy", handle_traffic_buy)
app.router.add_get("/api/devices/packs", handle_device_packs)
app.router.add_post("/api/devices/buy", handle_device_buy)
app.router.add_get("/api/clients", handle_clients)
app.router.add_get("/api/servers", handle_servers)
app.router.add_get("/api/faq", handle_faq)
app.router.add_get("/api/tickets", handle_tickets_list)
app.router.add_post("/api/tickets", handle_tickets_create)
app.router.add_get("/api/tickets/{ticket_id}", handle_ticket_get)
app.router.add_post("/api/tickets/{ticket_id}/messages", handle_ticket_reply)
app.router.add_post("/api/tickets/{ticket_id}/close", handle_ticket_close)
app.router.add_post("/api/payments/heleket", handle_heleket_webhook)
app.router.add_get("/api/payments/digiseller", handle_digiseller_callback)
app.router.add_post("/api/payments/digiseller", handle_digiseller_callback)
app.router.add_post("/api/payments/lava", handle_lava_webhook)
app.router.add_post("/api/payments/anypay", handle_anypay_webhook)
app.router.add_get("/api/payments/anypay", handle_anypay_webhook)
app.router.add_get("/pay/lava/success", handle_lava_success_page)
app.router.add_get("/pay/lava/fail", handle_lava_fail_page)
app.router.add_get("/pay/anypay/success", handle_anypay_success_page)
app.router.add_get("/pay/anypay/fail", handle_anypay_fail_page)
app.router.add_post("/api/topup/redeem", handle_topup_redeem)
app.router.add_get("/api/topup/methods", handle_topup_methods)
app.router.add_post("/api/topup/invoice", handle_topup_invoice)
app.router.add_get("/api/stars/packs", handle_stars_packs)
app.router.add_post("/api/stars/invoice", handle_stars_invoice)
app.router.add_get("/api/wg", handle_wg_status)
app.router.add_get("/api/wg/configs", handle_wg_list)
app.router.add_post("/api/wg/configs", handle_wg_buy)
app.router.add_get("/api/wg/configs/{config_id}/qr", handle_wg_qr)
app.router.add_post("/api/wg/configs/{config_id}/enable", handle_wg_enable)
app.router.add_post("/api/wg/configs/{config_id}/disable", handle_wg_disable)
app.router.add_delete("/api/wg/configs/{config_id}", handle_wg_delete)
app.router.add_get("/api/awg", handle_awg_status)
app.router.add_get("/api/awg/configs", handle_awg_list)
app.router.add_post("/api/awg/configs", handle_awg_buy)
app.router.add_get("/api/awg/configs/{config_id}/qr", handle_awg_qr)
app.router.add_post("/api/awg/configs/{config_id}/enable", handle_awg_enable)
app.router.add_post("/api/awg/configs/{config_id}/disable", handle_awg_disable)
app.router.add_delete("/api/awg/configs/{config_id}", handle_awg_delete)
register_topup_admin_routes(app, webapp_dir=WEBAPP_DIR)
register_growth_routes(app)
if WEBAPP_DIR.is_dir():
app.router.add_get("/", handle_index)
app.router.add_static("/static/", WEBAPP_DIR / "static", name="static")
return app
async def handle_index(_: web.Request) -> web.Response:
index = WEBAPP_DIR / "index.html"
resp = web.FileResponse(index)
# Нужно для popup Telegram Login (postMessage)
resp.headers["Cross-Origin-Opener-Policy"] = "same-origin-allow-popups"
return resp
async def handle_health(_: web.Request) -> web.Response:
return web.json_response({"ok": True})
async def handle_config(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
return web.json_response(
{
"ok": True,
"bot_username": request.app["bot_username"],
"telegram_client_id": settings.effective_telegram_client_id,
"webapp_url": settings.webapp_url,
"crypto_enabled": request.app["cryptopay"].enabled,
"heleket_enabled": request.app["heleket"].enabled,
"digiseller_enabled": request.app["digiseller"].enabled,
"lava_enabled": request.app["lava"].enabled,
"anypay_enabled": request.app["anypay"].enabled,
"stars_enabled": settings.stars_enabled,
"stars_rub_per_star": settings.stars_rub_per_star,
"topup_code_pay_url": settings.topup_code_pay_url,
"topup": topup_methods_payload(
cryptopay=request.app["cryptopay"],
heleket=request.app["heleket"],
lava=request.app["lava"],
anypay=request.app["anypay"],
),
}
)
async def handle_topup_methods(request: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
**topup_methods_payload(
cryptopay=request.app["cryptopay"],
heleket=request.app["heleket"],
lava=request.app["lava"],
anypay=request.app["anypay"],
),
}
)
async def handle_topup_invoice(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
db: Database = request.app["db"]
try:
inv = await create_web_topup_invoice(
db=db,
telegram_id=user.telegram_id,
amount_rub=body.get("amount"),
provider=body.get("provider"),
cryptopay=request.app["cryptopay"],
heleket=request.app["heleket"],
lava=request.app["lava"],
anypay=request.app["anypay"],
)
except TopupError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
except Exception as exc: # noqa: BLE001
logger.exception("web topup invoice failed")
return web.json_response(
{"ok": False, "error": f"Не удалось создать счёт: {exc}"},
status=500,
)
return web.json_response(
{
"ok": True,
"provider": inv.provider,
"title": inv.title,
"amount": inv.amount_rub,
"pay_url": inv.pay_url,
"invoice_id": inv.invoice_id,
"order_id": inv.order_id,
"message": (
f"Счёт на {inv.amount_rub:.0f} ₽ создан. "
"После оплаты баланс зачислится автоматически."
),
}
)
async def handle_topup_redeem(request: web.Request) -> web.Response:
if not allow(request, "topup_redeem", limit=10, window_sec=300):
return web.json_response({"ok": False, "error": "Слишком много попыток"}, status=429)
user = _auth_user_optional(request)
if not user:
return web.json_response(
{"ok": False, "error": "Войдите в аккаунт"},
status=401,
)
if user.banned:
return web.json_response({"ok": False, "error": "Аккаунт заблокирован"}, status=403)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
code = str(body.get("code") or "")
db: Database = request.app["db"]
try:
result = redeem_topup_code(db, telegram_id=user.telegram_id, raw_code=code)
db.log_payment_event(
provider="topup_code",
event_type="redeem",
telegram_id=user.telegram_id,
amount=float(getattr(result, "amount", 0) or 0),
status="ok",
meta=code[:32],
)
except TopupCodeError as exc:
db.log_payment_event(
provider="topup_code",
event_type="redeem",
telegram_id=user.telegram_id,
status="error",
meta=str(exc)[:200],
)
return web.json_response({"ok": False, "error": str(exc)}, status=400)
plain = (
result.message.replace("", "")
.replace("", "")
.replace("", "")
.replace("", "")
)
return web.json_response(
{
"ok": result.ok,
"already": result.already,
"amount": result.amount,
"balance": result.balance,
"message": plain,
}
)
async def handle_stars_packs(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.stars_enabled:
return web.json_response({"ok": False, "error": "Stars выключены"}, status=503)
return web.json_response(
{
"ok": True,
"rub_per_star": settings.stars_rub_per_star,
"packs": packs_payload(rub_per_star=settings.stars_rub_per_star),
"note": (
"Оплата через Telegram Stars (Mini App / Telegram Web). "
f"Курс: 1 ⭐ ≈ {settings.stars_rub_per_star:g} ₽"
),
}
)
async def handle_stars_invoice(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.stars_enabled:
return web.json_response({"ok": False, "error": "Stars выключены"}, status=503)
bot = request.app.get("bot")
if bot is None:
return web.json_response(
{"ok": False, "error": "Бот недоступен для создания счёта"},
status=503,
)
# Stars только для Telegram-пользователя (initData)
init_data = request.headers.get("X-Telegram-Init-Data") or ""
db: Database = request.app["db"]
user: BotUser | None = None
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
user = db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
else:
user = _auth_user_optional(request)
if not user:
return web.json_response(
{"ok": False, "error": "Откройте кабинет из Telegram"},
status=401,
)
if user.telegram_id <= 0:
return web.json_response(
{
"ok": False,
"error": "Stars доступны только через Telegram. Войдите через Telegram Login или Mini App.",
},
status=400,
)
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 body.get("amount_rub") or 0)
except (TypeError, ValueError):
return web.json_response({"ok": False, "error": "bad amount"}, status=400)
if amount <= 0:
return web.json_response({"ok": False, "error": "amount must be > 0"}, status=400)
try:
created = await create_stars_topup(
bot=bot,
db=db,
telegram_id=user.telegram_id,
amount_rub=amount,
rub_per_star=settings.stars_rub_per_star,
)
except Exception as exc: # noqa: BLE001
logger.exception("stars invoice create failed")
return web.json_response({"ok": False, "error": str(exc)}, status=500)
return web.json_response({"ok": True, **created})
def _digiseller_html(
*,
title: str,
body: str,
bot_username: str,
ok: bool,
) -> str:
bot_link = f"https://t.me/{bot_username}" if bot_username else "#"
color = "#16a34a" if ok else "#dc2626"
return f"""
{title}
"""
async def handle_digiseller_callback(request: web.Request) -> web.Response:
"""Автопроверка уникального кода (deliver_url Digiseller)."""
digiseller: Digiseller = request.app["digiseller"]
bot_username = str(request.app.get("bot_username") or "")
if not digiseller.enabled:
return web.Response(
text=_digiseller_html(
title="Digiseller выключен",
body="Пополнение через Digiseller не настроено.",
bot_username=bot_username,
ok=False,
),
content_type="text/html",
status=503,
)
code = (
request.rel_url.query.get("uniquecode")
or request.rel_url.query.get("unique_code")
or request.rel_url.query.get("UniqueCode")
or ""
).strip()
tg_hint_raw = (
request.rel_url.query.get("telegram_id")
or request.rel_url.query.get("tg_id")
or ""
).strip()
if request.method == "POST" and not code:
ctype = request.headers.get("Content-Type", "")
try:
if "json" in ctype:
data = await request.json()
if isinstance(data, dict):
code = str(
data.get("uniquecode")
or data.get("unique_code")
or data.get("UniqueCode")
or ""
).strip()
if not tg_hint_raw:
tg_hint_raw = str(
data.get("telegram_id") or data.get("tg_id") or ""
).strip()
else:
form = await request.post()
code = str(
form.get("uniquecode")
or form.get("unique_code")
or form.get("UniqueCode")
or ""
).strip()
if not tg_hint_raw:
tg_hint_raw = str(
form.get("telegram_id") or form.get("tg_id") or ""
).strip()
except Exception: # noqa: BLE001
logger.exception("Digiseller callback: bad POST body")
if not code:
return web.Response(
text=_digiseller_html(
title="Нет кода",
body="В ссылке нет параметра uniquecode. Вернись в бота.",
bot_username=bot_username,
ok=False,
),
content_type="text/html",
status=400,
)
tg_hint: int | None = None
if tg_hint_raw.isdigit():
tg_hint = int(tg_hint_raw)
try:
result = await fulfill_digiseller_unique_code(
db=request.app["db"],
digiseller=digiseller,
unique_code=code,
telegram_id_hint=tg_hint,
bot=request.app.get("bot"),
notify=True,
)
except DigisellerError as exc:
logger.warning("Digiseller callback reject: %s", exc)
return web.Response(
text=_digiseller_html(
title="Не удалось зачислить",
body=str(exc),
bot_username=bot_username,
ok=False,
),
content_type="text/html",
status=400,
)
except Exception as exc: # noqa: BLE001
logger.exception("Digiseller callback error")
return web.Response(
text=_digiseller_html(
title="Ошибка",
body=str(exc),
bot_username=bot_username,
ok=False,
),
content_type="text/html",
status=500,
)
if result.already:
title = "Уже зачислено"
body = (
f"Этот платёж уже обработан.
"
f"Баланс: {result.balance:.0f} ₽"
)
else:
title = "Баланс пополнен"
body = (
f"+{result.amount:.0f} ₽ зачислено автоматически.
"
f"Доступно: {result.balance:.0f} ₽
"
f"Можешь закрыть страницу и вернуться в бота."
)
return web.Response(
text=_digiseller_html(
title=title,
body=body,
bot_username=bot_username,
ok=True,
),
content_type="text/html",
)
async def handle_heleket_webhook(request: web.Request) -> web.Response:
heleket: Heleket = request.app["heleket"]
if not heleket.enabled:
return web.json_response({"ok": False, "error": "disabled"}, status=503)
try:
data = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
if not isinstance(data, dict) or not verify_webhook_sign(data, heleket.api_key):
logger.warning("Heleket webhook: bad signature")
return web.json_response({"ok": False, "error": "bad sign"}, status=403)
status = str(data.get("status") or "")
order_id = str(data.get("order_id") or "")
if not order_id:
return web.json_response({"ok": False, "error": "no order_id"}, status=400)
if status not in ("paid", "paid_over"):
return web.json_response({"ok": True, "ignored": status})
db: Database = request.app["db"]
local = db.get_invoice_by_order_id(order_id)
if not local:
logger.warning("Heleket webhook: unknown order_id=%s", order_id)
return web.json_response({"ok": False, "error": "unknown order"}, status=404)
try:
result = await fulfill_paid_invoice(
db=db,
sdk=request.app["sdk"],
tariffs=request.app["tariffs"],
local=local,
)
except Exception: # noqa: BLE001
logger.exception("Heleket fulfill failed order_id=%s", order_id)
return web.json_response({"ok": False, "error": "fulfill failed"}, status=500)
logger.info(
"Heleket paid order_id=%s already=%s ok=%s",
order_id,
result.already,
result.ok,
)
return web.json_response({"ok": True, "already": result.already})
async def handle_lava_webhook(request: web.Request) -> web.Response:
lava: Lava = request.app["lava"]
if not lava.enabled:
return web.json_response({"ok": False, "error": "disabled"}, status=503)
raw = await request.read()
try:
data = json.loads(raw.decode("utf-8") or "{}")
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
if not isinstance(data, dict):
return web.json_response({"ok": False, "error": "invalid body"}, status=400)
auth = request.headers.get("Authorization")
if not lava.additional_key:
logger.error("Lava webhook rejected: LAVA_ADDITIONAL_KEY not set")
return web.json_response(
{"ok": False, "error": "LAVA_ADDITIONAL_KEY required"},
status=503,
)
if not verify_lava_webhook(
data=data,
raw_body=raw,
authorization=auth,
additional_key=lava.additional_key,
):
logger.warning("Lava webhook: bad signature")
return web.json_response({"ok": False, "error": "bad sign"}, status=403)
status = str(data.get("status") or "").lower()
order_id = str(data.get("order_id") or data.get("orderId") or "").strip()
if not order_id:
return web.json_response({"ok": False, "error": "no order_id"}, status=400)
if status not in ("success", "paid"):
return web.json_response({"ok": True, "ignored": status})
db: Database = request.app["db"]
local = db.get_invoice_by_order_id(order_id)
if not local:
logger.warning("Lava webhook: unknown order_id=%s", order_id)
return web.json_response({"ok": False, "error": "unknown order"}, status=404)
try:
result = await fulfill_paid_invoice(
db=db,
sdk=request.app["sdk"],
tariffs=request.app["tariffs"],
local=local,
)
except Exception: # noqa: BLE001
logger.exception("Lava fulfill failed order_id=%s", order_id)
return web.json_response({"ok": False, "error": "fulfill failed"}, status=500)
bot = request.app.get("bot")
if bot is not None and result.ok and not result.already:
try:
await bot.send_message(
int(local["telegram_id"]),
result.message,
parse_mode="HTML",
)
except Exception as exc: # noqa: BLE001
logger.warning("Lava notify failed: %s", exc)
logger.info(
"Lava paid order_id=%s already=%s ok=%s",
order_id,
result.already,
result.ok,
)
return web.json_response({"ok": True, "already": result.already})
async def _anypay_form_data(request: web.Request) -> dict[str, Any]:
data: dict[str, Any] = {}
if request.method == "GET":
for k, v in request.rel_url.query.items():
data[k] = v
return data
try:
if request.content_type and "json" in request.content_type:
body = await request.json()
if isinstance(body, dict):
return body
except Exception: # noqa: BLE001
pass
try:
form = await request.post()
for k in form.keys():
data[str(k)] = form.get(k)
except Exception: # noqa: BLE001
logger.exception("AnyPay webhook: bad form")
return data
async def handle_anypay_webhook(request: web.Request) -> web.Response:
anypay: Anypay = request.app["anypay"]
if not anypay.enabled:
return web.Response(text="disabled", status=503)
data = await _anypay_form_data(request)
if not data:
return web.Response(text="empty", status=400)
if anypay.check_ip:
peer = request.remote
forwarded = request.headers.get("X-Forwarded-For")
if not client_ip_allowed(peer, forwarded=forwarded):
logger.warning("AnyPay webhook: bad ip peer=%s xff=%s", peer, forwarded)
return web.Response(text="bad ip", status=403)
if not verify_anypay_sign(
data, project_id=anypay.project_id, api_key=anypay.api_key
):
logger.warning("AnyPay webhook: bad sign pay_id=%s", data.get("pay_id"))
return web.Response(text="wrong sign", status=403)
status = str(data.get("status") or "").lower()
pay_id = str(data.get("pay_id") or "").strip()
if not pay_id:
return web.Response(text="no pay_id", status=400)
if status not in ANYPAY_PAID and status != "paid":
# ещё не финал — принимаем без зачисления
return web.Response(text="OK")
# сверить сумму с инвойсом
try:
amount_notify = float(data.get("amount") or 0)
except (TypeError, ValueError):
amount_notify = 0.0
db: Database = request.app["db"]
local = db.get_invoice_by_order_id(pay_id)
if not local:
# иногда pay_id = local invoice_id
try:
local = db.get_invoice(int(pay_id), provider="anypay")
except Exception: # noqa: BLE001
local = None
if not local:
logger.warning("AnyPay webhook: unknown pay_id=%s", pay_id)
return web.Response(text="unknown", status=404)
expected = float(local.get("amount_rub") or 0)
if amount_notify > 0 and abs(amount_notify - expected) > 0.05:
logger.warning(
"AnyPay amount mismatch pay_id=%s got=%s expected=%s",
pay_id,
amount_notify,
expected,
)
return web.Response(text="bad amount", status=400)
try:
result = await fulfill_paid_invoice(
db=db,
sdk=request.app["sdk"],
tariffs=request.app["tariffs"],
local=local,
)
except Exception: # noqa: BLE001
logger.exception("AnyPay fulfill failed pay_id=%s", pay_id)
return web.Response(text="error", status=500)
bot = request.app.get("bot")
if bot is not None and result.ok and not result.already:
try:
await bot.send_message(
int(local["telegram_id"]),
result.message,
parse_mode="HTML",
)
except Exception as exc: # noqa: BLE001
logger.warning("AnyPay notify failed: %s", exc)
logger.info(
"AnyPay paid pay_id=%s already=%s ok=%s",
pay_id,
result.already,
result.ok,
)
# Важно: ответ OK латиницей — иначе AnyPay будет ретраить
return web.Response(text="OK")
async def handle_anypay_success_page(request: web.Request) -> web.Response:
bot_username = str(request.app.get("bot_username") or "")
return web.Response(
text=_digiseller_html(
title="Оплата прошла",
body="Если баланс ещё не обновился — подождите минуту и обновите кабинет.",
bot_username=bot_username,
ok=True,
),
content_type="text/html",
)
async def handle_anypay_fail_page(request: web.Request) -> web.Response:
bot_username = str(request.app.get("bot_username") or "")
return web.Response(
text=_digiseller_html(
title="Оплата не завершена",
body="Платёж отменён или не прошёл. Можно попробовать снова в кабинете.",
bot_username=bot_username,
ok=False,
),
content_type="text/html",
)
async def handle_lava_success_page(request: web.Request) -> web.Response:
bot_username = str(request.app.get("bot_username") or "")
return web.Response(
text=_digiseller_html(
title="Оплата прошла",
body=(
"Платёж Lava принят. Баланс зачислится в течение минуты — "
"проверь уведомление в боте."
),
bot_username=bot_username,
ok=True,
),
content_type="text/html",
)
async def handle_lava_fail_page(request: web.Request) -> web.Response:
bot_username = str(request.app.get("bot_username") or "")
return web.Response(
text=_digiseller_html(
title="Оплата не завершена",
body="Платёж отменён или не прошёл. Можно вернуться в бота и попробовать снова.",
bot_username=bot_username,
ok=False,
),
content_type="text/html",
)
def _secure_cookie(request: web.Request) -> bool:
# За прокси Caddy обычно HTTPS; X-Forwarded-Proto надёжнее для cookie Secure
proto = request.headers.get("X-Forwarded-Proto", "").lower()
if proto == "https":
return True
return request.url.scheme == "https"
def _set_session_cookie(response: web.Response, request: web.Request, token: str) -> None:
response.set_cookie(
SESSION_COOKIE,
token,
max_age=SESSION_DAYS * 24 * 3600,
httponly=True,
samesite="Lax",
secure=_secure_cookie(request),
path="/",
)
def _clear_session_cookie(response: web.Response, request: web.Request) -> None:
response.del_cookie(SESSION_COOKIE, path="/")
def _session_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(SESSION_COOKIE)
def _auth_user_optional(request: web.Request) -> BotUser | None:
"""Сессия cookie/Bearer или Telegram Mini App initData."""
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
init_data = (
request.headers.get("X-Telegram-Init-Data")
or request.query.get("initData")
or ""
)
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
return db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
except InitDataError:
pass
return db.get_user_by_session(_session_token(request))
def _require_user(request: web.Request) -> BotUser:
user = _auth_user_optional(request)
if not user:
raise web.HTTPUnauthorized(
text=json.dumps({"ok": False, "error": "unauthorized"}),
content_type="application/json",
)
if user.banned:
raise web.HTTPForbidden(
text=json.dumps({"ok": False, "error": "Аккаунт заблокирован"}),
content_type="application/json",
)
return user
def _user_payload(user: BotUser) -> dict[str, Any]:
return {
"id": user.telegram_id,
"telegram_id": user.telegram_id,
"name": user.display_name,
"username": user.username,
"login": user.login,
"has_password": user.has_password,
"is_telegram": user.is_telegram,
"auto_renew": bool(user.auto_renew),
"email": user.email,
"banned": bool(user.banned),
"referral_code": user.referral_code,
"trial_used": bool(user.trial_used_at),
"notify_telegram": bool(user.notify_telegram),
"notify_email": bool(user.notify_email),
}
async def handle_register(request: web.Request) -> web.Response:
db: Database = request.app["db"]
if not allow(request, "auth_register", limit=8, window_sec=600):
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 "")
password = str(body.get("password") or "")
full_name = str(body.get("full_name") or "").strip() or None
auto_renew = bool(body.get("auto_renew", True))
ref = str(body.get("ref") or request.cookies.get("vpn_ref") or "").strip()
try:
user, token = db.register_with_password(
login=login,
password=password,
full_name=full_name,
auto_renew=auto_renew,
)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
if ref:
db.attach_referrer(user.telegram_id, ref)
db.ensure_referral_code(user.telegram_id)
db.clear_session(token)
token = db.create_session(
user.telegram_id,
user_agent=request.headers.get("User-Agent"),
ip=client_ip(request),
)
user = db.get_user(user.telegram_id) or user
resp = web.json_response(
{
"ok": True,
"token": token,
"user": _user_payload(user),
"balance": user.balance,
}
)
_set_session_cookie(resp, request, token)
return resp
async def handle_login(request: web.Request) -> web.Response:
db: Database = request.app["db"]
if not allow(request, "auth_login", limit=20, 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)
try:
user, token = db.login_with_password(
str(body.get("login") or ""),
str(body.get("password") or ""),
)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
if user.banned:
return web.json_response(
{"ok": False, "error": "Аккаунт заблокирован"}, status=403
)
db.clear_session(token)
token = db.create_session(
user.telegram_id,
user_agent=request.headers.get("User-Agent"),
ip=client_ip(request),
)
resp = web.json_response(
{
"ok": True,
"token": token,
"user": _user_payload(user),
"balance": user.balance,
}
)
_set_session_cookie(resp, request, token)
return resp
async def handle_telegram_auth(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
try:
id_token = str(body.get("id_token") or "").strip()
if id_token:
tg = parse_and_validate_id_token(
id_token,
settings.effective_telegram_client_id,
)
else:
# legacy Login Widget (HMAC) — на случай старых клиентов
tg = parse_and_validate_login_widget(body, settings.bot_token)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
link_from = None
current = db.get_user_by_session(_session_token(request))
if current and not current.is_telegram:
link_from = current.telegram_id
try:
user, token = db.login_or_register_telegram(
tg.id,
username=tg.username,
full_name=tg.full_name,
link_from=link_from,
)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
resp = web.json_response(
{
"ok": True,
"token": token,
"user": _user_payload(user),
"balance": user.balance,
}
)
_set_session_cookie(resp, request, token)
return resp
async def handle_logout(request: web.Request) -> web.Response:
db: Database = request.app["db"]
token = _session_token(request)
db.clear_session(token)
resp = web.json_response({"ok": True})
_clear_session_cookie(resp, request)
return resp
async def handle_qr_auth_start(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
bot_username = str(request.app.get("bot_username") or "").lstrip("@")
created = db.create_login_qr()
token = created["token"]
deep_link = (
f"https://t.me/{bot_username}?start=qrauth_{token}" if bot_username else ""
)
web_approve = ""
if settings.webapp_url:
web_approve = f"{settings.webapp_url.rstrip('/')}/?qr_approve={token}"
qr_payload = deep_link or web_approve
if not qr_payload:
return web.json_response(
{"ok": False, "error": "Нет username бота и WEBAPP_URL"},
status=400,
)
try:
svg = make_qr_svg(qr_payload, scale=5, border=2)
qr_data_url = "data:image/svg+xml;base64," + base64.b64encode(svg).decode(
"ascii"
)
except Exception as exc: # noqa: BLE001
logger.exception("qr svg failed")
return web.json_response({"ok": False, "error": str(exc)}, status=500)
return web.json_response(
{
"ok": True,
"token": token,
"deep_link": deep_link,
"web_approve_url": web_approve,
"expires_in": created["expires_in"],
"qr_svg": qr_data_url,
}
)
async def handle_qr_auth_poll(request: web.Request) -> web.Response:
db: Database = request.app["db"]
token = (request.rel_url.query.get("token") or "").strip()
try:
result = db.poll_login_qr(token)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
status = result.get("status")
if status == "approved":
user: BotUser = result["user"]
session = str(result["token"])
resp = web.json_response(
{
"ok": True,
"status": "approved",
"token": session,
"user": _user_payload(user),
"balance": user.balance,
}
)
_set_session_cookie(resp, request, session)
return resp
return web.json_response({"ok": True, "status": status})
async def handle_qr_auth_approve(request: web.Request) -> web.Response:
"""Подтверждение QR с уже авторизованного устройства (браузер)."""
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
token = str(body.get("token") or "").strip()
try:
db.approve_login_qr(token, user.telegram_id)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
return web.json_response({"ok": True})
async def handle_account_email(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
email = body.get("email")
try:
updated = db.set_user_email(
user.telegram_id,
None if email is None else str(email),
)
except AuthError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
return web.json_response({"ok": True, "user": _user_payload(updated)})
async def handle_passkey_register_options(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
existing = db.list_webauthn_credentials(user.telegram_id)
existing_ids = [credential_id_bytes(r["credential_id"]) for r in existing]
user_name = user.login or user.username or f"user-{user.telegram_id}"
options, challenge = build_registration_options(
settings,
user_id=user.telegram_id,
user_name=user_name,
user_display_name=user.display_name,
existing_credential_ids=existing_ids,
)
db.save_webauthn_challenge(
challenge_b64=bytes_to_base64url(challenge),
purpose="register",
telegram_id=user.telegram_id,
)
except Exception as exc: # noqa: BLE001
logger.exception("passkey register options")
return web.json_response({"ok": False, "error": str(exc)}, status=400)
return web.json_response({"ok": True, "options": options})
async def handle_passkey_register_verify(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
credential = body.get("credential")
if not isinstance(credential, dict):
return web.json_response({"ok": False, "error": "credential required"}, status=400)
client_data = credential.get("response", {}).get("clientDataJSON") or ""
try:
challenge_b64 = challenge_from_client_data(str(client_data))
except Exception: # noqa: BLE001
challenge_b64 = None
if not challenge_b64:
return web.json_response({"ok": False, "error": "challenge missing"}, status=400)
stored = db.take_webauthn_challenge(str(challenge_b64), purpose="register")
if not stored or stored.get("telegram_id") != user.telegram_id:
return web.json_response({"ok": False, "error": "challenge expired"}, status=400)
try:
verification = verify_registration(
settings,
credential=credential,
expected_challenge=credential_id_bytes(str(challenge_b64)),
)
cid = credential_id_b64(verification.credential_id)
transports = parse_transports(credential.get("transports"))
db.add_webauthn_credential(
telegram_id=user.telegram_id,
credential_id=cid,
public_key=verification.credential_public_key,
sign_count=int(verification.sign_count or 0),
transports=transports,
device_name=str(body.get("device_name") or "").strip() or None,
)
except Exception as exc: # noqa: BLE001
logger.exception("passkey register verify")
return web.json_response({"ok": False, "error": str(exc)}, status=400)
return web.json_response({"ok": True, "credential_id": cid})
async def handle_passkey_login_options(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
try:
body = await request.json()
except json.JSONDecodeError:
body = {}
allow_ids: list[bytes] | None = None
login = str((body or {}).get("login") or "").strip()
telegram_id: int | None = None
if login:
user = db.get_user_by_login(login)
if not user:
return web.json_response(
{"ok": False, "error": "Пользователь не найден"}, status=404
)
creds = db.list_webauthn_credentials(user.telegram_id)
if not creds:
return web.json_response(
{"ok": False, "error": "Passkey не привязан"}, status=400
)
allow_ids = [credential_id_bytes(c["credential_id"]) for c in creds]
telegram_id = user.telegram_id
try:
options, challenge = build_authentication_options(
settings, allow_credential_ids=allow_ids
)
db.save_webauthn_challenge(
challenge_b64=bytes_to_base64url(challenge),
purpose="login",
telegram_id=telegram_id,
)
except Exception as exc: # noqa: BLE001
logger.exception("passkey login options")
return web.json_response({"ok": False, "error": str(exc)}, status=400)
return web.json_response({"ok": True, "options": options})
async def handle_passkey_login_verify(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
credential = body.get("credential")
if not isinstance(credential, dict):
return web.json_response({"ok": False, "error": "credential required"}, status=400)
raw_id = str(credential.get("id") or credential.get("rawId") or "").strip()
if not raw_id:
return web.json_response({"ok": False, "error": "credential id missing"}, status=400)
stored_cred = db.get_webauthn_credential(raw_id)
if not stored_cred:
return web.json_response({"ok": False, "error": "Неизвестный passkey"}, status=400)
client_data = credential.get("response", {}).get("clientDataJSON") or ""
try:
challenge_b64 = challenge_from_client_data(str(client_data))
except Exception: # noqa: BLE001
challenge_b64 = None
if not challenge_b64:
return web.json_response({"ok": False, "error": "challenge missing"}, status=400)
stored = db.take_webauthn_challenge(str(challenge_b64), purpose="login")
if not stored:
return web.json_response({"ok": False, "error": "challenge expired"}, status=400)
try:
verification = verify_authentication(
settings,
credential=credential,
expected_challenge=credential_id_bytes(str(challenge_b64)),
credential_public_key=bytes(stored_cred["public_key"]),
credential_current_sign_count=int(stored_cred["sign_count"] or 0),
)
db.update_webauthn_sign_count(
raw_id, int(verification.new_sign_count or 0)
)
tid = int(stored_cred["telegram_id"])
session = db.create_session(tid)
user = db.get_user(tid)
assert user is not None
except Exception as exc: # noqa: BLE001
logger.exception("passkey login verify")
return web.json_response({"ok": False, "error": str(exc)}, status=400)
resp = web.json_response(
{
"ok": True,
"token": session,
"user": _user_payload(user),
"balance": user.balance,
}
)
_set_session_cookie(resp, request, session)
return resp
async def handle_passkey_list(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
rows = db.list_webauthn_credentials(user.telegram_id)
items = [
{
"id": r["credential_id"],
"device_name": r.get("device_name") or "Passkey",
"created_at": r["created_at"].isoformat()
if hasattr(r["created_at"], "isoformat")
else str(r["created_at"]),
"last_used_at": (
r["last_used_at"].isoformat()
if r.get("last_used_at") and hasattr(r["last_used_at"], "isoformat")
else None
),
}
for r in rows
]
return web.json_response({"ok": True, "passkeys": items})
async def handle_passkey_delete(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
cid = request.match_info.get("credential_id") or ""
ok = db.delete_webauthn_credential(user.telegram_id, cid)
if not ok:
return web.json_response({"ok": False, "error": "not found"}, status=404)
return web.json_response({"ok": True})
async def handle_me(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
sdk = request.app["sdk"]
# Mini App: initData → сессия, чтобы дальше работал и браузерный кабинет
init_data = request.headers.get("X-Telegram-Init-Data") or ""
session_issued: str | None = None
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
user = db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
session_issued = db.create_session(user.telegram_id)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
else:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
sub = await get_my_subscription_info(
sdk,
user.telegram_id,
login=user.login,
db=db,
tariffs=request.app["tariffs"],
renewal_bot=request.app.get("bot"),
)
except Exception as exc: # noqa: BLE001
logger.exception("me subscription error")
return web.json_response({"ok": False, "error": str(exc)}, status=500)
# Реалтайм: если подписка мертва — гасим WG/AWG на панелях
try:
await enforce_wg_configs_subscription(
db=db, sdk=sdk, telegram_id=user.telegram_id, login=user.login
)
await enforce_awg_configs_subscription(
db=db, sdk=sdk, telegram_id=user.telegram_id, login=user.login
)
except Exception: # noqa: BLE001
logger.exception("vpn configs enforce on /api/me failed tg=%s", user.telegram_id)
fresh = db.get_user(user.telegram_id) or user
extras = extras_as_dict(fresh)
resp = web.json_response(
{
"ok": True,
"user": _user_payload(fresh),
"balance": fresh.balance,
"subscription": {
"has": sub.has_subscription,
"expire_at": sub.expire_at.isoformat() if sub.expire_at else None,
"url": sub.subscription_url,
"text": sub.text,
"traffic": sub.traffic.as_dict() if sub.traffic else None,
"devices": sub.devices.as_dict() if sub.devices else None,
"extras": extras,
},
"crypto_enabled": request.app["cryptopay"].enabled,
"token": session_issued,
}
)
if session_issued:
_set_session_cookie(resp, request, session_issued)
return resp
async def handle_auto_renew(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
if "enabled" not in body:
return web.json_response({"ok": False, "error": "enabled required"}, status=400)
updated = db.set_auto_renew(user.telegram_id, bool(body.get("enabled")))
return web.json_response(
{
"ok": True,
"auto_renew": bool(updated.auto_renew),
"user": _user_payload(updated),
}
)
async def handle_transactions(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
q = request.rel_url.query
try:
limit = int(q.get("limit") or "10")
except ValueError:
limit = 10
try:
page = int(q.get("page") or "1")
except ValueError:
page = 1
limit = max(1, min(limit, 50))
page = max(1, page)
offset = (page - 1) * limit
rows, total = db.list_transactions(user.telegram_id, limit=limit, offset=offset)
pages = max(1, (total + limit - 1) // limit) if total else 1
if page > pages and total:
page = pages
offset = (page - 1) * limit
rows, total = db.list_transactions(user.telegram_id, limit=limit, offset=offset)
items = [
transaction_row_dict(
float(r["amount"]),
str(r["kind"]),
r.get("meta"),
r["created_at"],
)
for r in rows
]
return web.json_response(
{
"ok": True,
"items": items,
"total": total,
"limit": limit,
"offset": offset,
"page": page,
"pages": pages,
}
)
async def handle_tariffs(request: web.Request) -> web.Response:
tariffs: list[Tariff] = request.app["tariffs"]
return web.json_response(
{
"ok": True,
"tariffs": [
{
"id": t.id,
"title": t.title,
"price": t.price,
"price_label": t.format_price(),
"days": t.days,
"traffic_gb": t.traffic_gb,
"device_limit": t.device_limit,
"description": t.description,
}
for t in tariffs
],
}
)
async def handle_buy(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
user: BotUser | None = None
init_data = request.headers.get("X-Telegram-Init-Data") or ""
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
user = db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
else:
user = _auth_user_optional(request)
if not user:
return web.json_response(
{"ok": False, "error": "Войдите или откройте из Telegram"},
status=401,
)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
tariff_id = str(body.get("tariff_id") or "")
renew = bool(body.get("renew"))
promo_code = str(body.get("promo_code") or "").strip() or None
tariff = get_tariff(request.app["tariffs"], tariff_id)
if not tariff:
return web.json_response({"ok": False, "error": "tariff not found"}, status=404)
result = await purchase_tariff(
db=db,
sdk=request.app["sdk"],
telegram_id=user.telegram_id,
tariff=tariff,
charge_balance=True,
renew=renew,
promo_code=promo_code,
payment_meta=f"webapp:{'renew' if renew else 'buy'}:{tariff.id}",
)
return web.json_response(
{
"ok": result.ok,
"message": result.message,
"balance": result.balance,
"subscription_url": result.subscription_url,
"expire_at": result.expire_at.isoformat() if result.expire_at else None,
},
status=200 if result.ok else 400,
)
async def handle_device_packs(_: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"packs": [
{
"id": p.id,
"slots": p.slots,
"title": p.title,
"price": p.price,
"price_label": p.format_price(),
"billing": "monthly",
}
for p in DEVICE_PACKS
],
"monthly_note": (
"Первое списание при покупке. Далее — каждые 30 дней с баланса "
"за все доп. устройства."
),
}
)
async def handle_device_buy(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
user: BotUser | None = None
init_data = request.headers.get("X-Telegram-Init-Data") or ""
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
user = db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
else:
user = _auth_user_optional(request)
if not user:
return web.json_response(
{"ok": False, "error": "Войдите или откройте из Telegram"},
status=401,
)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
slots = int(body.get("slots") or 0)
pack = get_device_pack(str(body.get("pack_id") or slots or ""))
if not pack and slots > 0:
from services.devices import DevicePack, EXTRA_DEVICE_PRICE_RUB
pack = DevicePack(slots=slots, price=EXTRA_DEVICE_PRICE_RUB * slots)
if not pack:
return web.json_response({"ok": False, "error": "pack not found"}, status=404)
result = await purchase_device_pack(
db=db,
sdk=request.app["sdk"],
telegram_id=user.telegram_id,
pack=pack,
charge_balance=True,
login=user.login,
payment_meta=f"webapp:devices:{pack.id}",
)
return web.json_response(
{
"ok": result.ok,
"message": result.message,
"balance": result.balance,
"devices": result.devices.as_dict() if result.devices else None,
},
status=200 if result.ok else 400,
)
async def handle_traffic_packs(_: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"packs": [
{
"id": p.id,
"packs": p.packs,
"gb": p.gb,
"title": p.title,
"price": p.price,
"price_label": p.format_price(),
"billing": "monthly",
}
for p in TRAFFIC_PACKS
],
"monthly_note": (
"Первое списание при покупке. Далее — каждые 30 дней с баланса "
"за все подключённые пакеты."
),
}
)
async def handle_traffic_buy(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
db: Database = request.app["db"]
user: BotUser | None = None
init_data = request.headers.get("X-Telegram-Init-Data") or ""
if init_data.strip():
try:
tg = parse_and_validate_init_data(init_data, settings.bot_token)
user = db.ensure_user(
tg.id,
username=tg.username,
full_name=tg.full_name,
)
except InitDataError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=401)
else:
user = _auth_user_optional(request)
if not user:
return web.json_response(
{"ok": False, "error": "Войдите или откройте из Telegram"},
status=401,
)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
pack = get_traffic_pack(str(body.get("pack_id") or body.get("packs") or ""))
if not pack:
return web.json_response({"ok": False, "error": "pack not found"}, status=404)
result = await purchase_traffic_pack(
db=db,
sdk=request.app["sdk"],
telegram_id=user.telegram_id,
pack=pack,
charge_balance=True,
login=user.login,
payment_meta=f"webapp:traffic:{pack.id}",
)
return web.json_response(
{
"ok": result.ok,
"message": result.message,
"balance": result.balance,
"traffic": result.traffic.as_dict() if result.traffic else None,
},
status=200 if result.ok else 400,
)
async def handle_faq(_: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"items": [
{
"id": item.id,
"question": item.question,
"answer": item.answer,
}
for item in FAQ_ITEMS
],
}
)
async def handle_tickets_list(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
db: Database = request.app["db"]
include_closed = (request.rel_url.query.get("closed") or "1") != "0"
tickets = db.list_tickets_for_user(
user.telegram_id,
limit=50,
include_closed=include_closed,
)
return web.json_response(
{
"ok": True,
"tickets": [ticket_to_dict(t) for t in tickets],
"open_count": db.count_open_tickets(user.telegram_id),
"max_open": MAX_OPEN_PER_USER,
"subject_max": SUBJECT_MAX,
"body_max": BODY_MAX,
}
)
async def handle_tickets_create(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
db: Database = request.app["db"]
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
subject = clamp_subject(str(body.get("subject") or ""))
text = clamp_body(str(body.get("body") or body.get("message") or ""))
if len(subject) < 3:
return web.json_response(
{"ok": False, "error": "Тема слишком короткая"},
status=400,
)
if len(text) < 5:
return web.json_response(
{"ok": False, "error": "Сообщение слишком короткое"},
status=400,
)
if db.count_open_tickets(user.telegram_id) >= MAX_OPEN_PER_USER:
return web.json_response(
{
"ok": False,
"error": f"Лимит: не больше {MAX_OPEN_PER_USER} открытых тикетов",
},
status=400,
)
try:
ticket = db.create_ticket(user.telegram_id, subject=subject, body=text)
except ValueError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
bot = request.app.get("bot")
settings: Settings = request.app["settings"]
ai_result = {"action": "skip"}
if bot is not None:
ai_result = await try_ai_ticket_reply(
db=db,
bot=bot,
settings=settings,
ticket_id=int(ticket["id"]),
notify_user=True,
)
if ai_result.get("action") in ("skip", "error") or (
ai_result.get("action") == "answer" and ai_result.get("ok")
):
if ai_result.get("action") != "escalate":
await notify_admins_new_ticket(
bot,
settings,
ticket_id=int(ticket["id"]),
telegram_id=user.telegram_id,
subject=subject,
body=text,
ai_handled=ai_result.get("action") == "answer",
)
ticket = db.get_ticket(int(ticket["id"])) or ticket
messages = db.list_ticket_messages(int(ticket["id"]))
return web.json_response(
{
"ok": True,
"ticket": ticket_to_dict({**ticket, "message_count": len(messages)}),
"messages": [message_to_dict(m) for m in messages],
"ai": {
"action": ai_result.get("action"),
"model": ai_result.get("model"),
},
}
)
async def handle_ticket_get(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
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 or int(ticket["telegram_id"]) != user.telegram_id:
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)}),
"messages": [message_to_dict(m) for m in messages],
}
)
async def handle_ticket_reply(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
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 or int(ticket["telegram_id"]) != user.telegram_id:
return web.json_response({"ok": False, "error": "not found"}, status=404)
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)
try:
ticket = db.add_ticket_message(
tid,
author_role="user",
author_id=user.telegram_id,
body=text,
)
except ValueError as exc:
return web.json_response({"ok": False, "error": str(exc)}, status=400)
bot = request.app.get("bot")
settings: Settings = request.app["settings"]
ai_result = {"action": "skip"}
if bot is not None:
ai_result = await try_ai_ticket_reply(
db=db,
bot=bot,
settings=settings,
ticket_id=tid,
notify_user=True,
)
if ai_result.get("action") in ("skip", "error"):
await notify_admins_ticket_reply(
bot,
settings,
ticket_id=tid,
telegram_id=user.telegram_id,
body=text,
)
ticket = db.get_ticket(tid) or ticket
messages = db.list_ticket_messages(tid)
return web.json_response(
{
"ok": True,
"ticket": ticket_to_dict({**ticket, "message_count": len(messages)}),
"messages": [message_to_dict(m) for m in messages],
"ai": {
"action": ai_result.get("action"),
"model": ai_result.get("model"),
},
}
)
async def handle_ticket_close(request: web.Request) -> web.Response:
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
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 or int(ticket["telegram_id"]) != user.telegram_id:
return web.json_response({"ok": False, "error": "not found"}, status=404)
ticket = db.set_ticket_status(tid, "closed")
messages = db.list_ticket_messages(tid)
return web.json_response(
{
"ok": True,
"ticket": ticket_to_dict(
{**(ticket or {}), "message_count": len(messages)}
),
"messages": [message_to_dict(m) for m in messages],
}
)
async def handle_clients(_: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"clients": [
{
"id": c.id,
"name": c.name,
"core": c.core,
"description": c.description,
"featured": c.featured,
"hwid": c.hwid,
"links": [{"title": l.title, "url": l.url} for l in c.links],
}
for c in TOP_CLIENTS
],
"docs": "https://docs.rw/clients",
}
)
async def handle_servers(request: web.Request) -> web.Response:
try:
nodes = await fetch_nodes_status(request.app["sdk"])
except Exception as exc: # noqa: BLE001
return web.json_response({"ok": False, "error": str(exc)}, status=500)
return web.json_response(
{
"ok": True,
"online": sum(1 for n in nodes if n.online),
"total": len(nodes),
"nodes": [
{
"name": n.name,
"country": n.country_code,
"online": n.online,
"label": n.label,
"ping_ms": n.tcp_ms,
"users_online": n.users_online,
}
for n in nodes
],
}
)
async def handle_wg_status(request: web.Request) -> web.Response:
db: Database = request.app["db"]
settings = load_wg_purchase_settings(db)
servers = list_available_servers(db) if settings.enabled else []
used = 0
try:
user = _require_user(request)
used = db.count_xui_wg_configs(user.telegram_id)
except web.HTTPUnauthorized:
pass
return web.json_response(
{
"ok": True,
**settings.as_dict(),
"used": used,
"remaining": max(0, settings.max_configs - used),
"servers": servers,
}
)
async def handle_wg_list(request: web.Request) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
gate = await enforce_wg_configs_subscription(
db=db,
sdk=sdk,
telegram_id=user.telegram_id,
login=user.login,
)
configs = list_user_wg_configs(db, user.telegram_id)
settings = load_wg_purchase_settings(db)
return web.json_response(
{
"ok": True,
"configs": configs,
"used": len(configs),
"max_configs": settings.max_configs,
"enabled": settings.enabled,
"price": settings.price,
"price_label": settings.as_dict()["price_label"],
"servers": list_available_servers(db) if settings.enabled else [],
"subscription_active": gate["subscription_active"],
"subscription_reason": gate.get("reason") or "",
}
)
async def handle_wg_buy(request: web.Request) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
try:
server_id = int(body.get("server_id") or 0)
except (TypeError, ValueError):
return web.json_response({"ok": False, "error": "bad server_id"}, status=400)
if server_id <= 0:
return web.json_response({"ok": False, "error": "Выберите сервер"}, status=400)
remark = str(body.get("remark") or "").strip()[:80]
result = await purchase_wg_config(
db=db,
sdk=sdk,
telegram_id=user.telegram_id,
server_id=server_id,
remark=remark,
login=user.login,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"balance": result.balance,
"config": result.config,
"configs": list_user_wg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_wg_enable(request: web.Request) -> web.Response:
return await _handle_wg_toggle(request, enable=True)
async def handle_wg_disable(request: web.Request) -> web.Response:
return await _handle_wg_toggle(request, enable=False)
async def _handle_wg_toggle(request: web.Request, *, enable: bool) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
result = await set_wg_config_enabled(
db=db,
telegram_id=user.telegram_id,
config_id=config_id,
enable=enable,
sdk=sdk,
login=user.login,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"config": result.config,
"configs": list_user_wg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_wg_delete(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
result = await delete_wg_config(
db=db,
telegram_id=user.telegram_id,
config_id=config_id,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"configs": list_user_wg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_wg_qr(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
row = db.get_xui_wg_config(config_id, telegram_id=user.telegram_id)
if not row:
return web.json_response({"ok": False, "error": "not found"}, status=404)
url = str(row.get("subscription_url") or "").strip()
if not url:
return web.json_response({"ok": False, "error": "no subscription url"}, status=400)
try:
svg = make_qr_svg(url, scale=5, border=2)
except Exception as exc: # noqa: BLE001
logger.exception("QR generate failed")
return web.json_response({"ok": False, "error": str(exc)}, status=500)
return web.Response(
body=svg,
content_type="image/svg+xml",
headers={
"Cache-Control": "private, max-age=300",
"Content-Disposition": f'inline; filename="wg-{config_id}.svg"',
},
)
async def handle_awg_status(request: web.Request) -> web.Response:
db: Database = request.app["db"]
settings = load_awg_purchase_settings(db)
servers = list_available_awg_servers(db) if settings.enabled else []
used = 0
try:
user = _require_user(request)
used = db.count_awg_configs(user.telegram_id)
except web.HTTPUnauthorized:
pass
return web.json_response(
{
"ok": True,
**settings.as_dict(),
"used": used,
"remaining": max(0, settings.max_configs - used),
"servers": servers,
}
)
async def handle_awg_list(request: web.Request) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
gate = await enforce_awg_configs_subscription(
db=db,
sdk=sdk,
telegram_id=user.telegram_id,
login=user.login,
)
configs = list_user_awg_configs(db, user.telegram_id)
settings = load_awg_purchase_settings(db)
return web.json_response(
{
"ok": True,
"configs": configs,
"used": len(configs),
"max_configs": settings.max_configs,
"enabled": settings.enabled,
"price": settings.price,
"price_label": settings.as_dict()["price_label"],
"servers": list_available_awg_servers(db) if settings.enabled else [],
"subscription_active": gate["subscription_active"],
"subscription_reason": gate.get("reason") or "",
}
)
async def handle_awg_buy(request: web.Request) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
body = await request.json()
except json.JSONDecodeError:
return web.json_response({"ok": False, "error": "invalid json"}, status=400)
try:
server_id = int(body.get("server_id") or 0)
except (TypeError, ValueError):
return web.json_response({"ok": False, "error": "bad server_id"}, status=400)
if server_id <= 0:
return web.json_response({"ok": False, "error": "Выберите сервер"}, status=400)
remark = str(body.get("remark") or "").strip()[:80]
result = await purchase_awg_config(
db=db,
sdk=sdk,
telegram_id=user.telegram_id,
server_id=server_id,
remark=remark,
login=user.login,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"balance": result.balance,
"config": result.config,
"configs": list_user_awg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_awg_enable(request: web.Request) -> web.Response:
return await _handle_awg_toggle(request, enable=True)
async def handle_awg_disable(request: web.Request) -> web.Response:
return await _handle_awg_toggle(request, enable=False)
async def _handle_awg_toggle(request: web.Request, *, enable: bool) -> web.Response:
db: Database = request.app["db"]
sdk = request.app["sdk"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
result = await set_awg_config_enabled(
db=db,
telegram_id=user.telegram_id,
config_id=config_id,
enabled=enable,
sdk=sdk,
login=user.login,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"config": result.config,
"configs": list_user_awg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_awg_delete(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
result = await delete_awg_config(
db=db,
telegram_id=user.telegram_id,
config_id=config_id,
)
status = 200 if result.ok else 400
return web.json_response(
{
"ok": result.ok,
"error": None if result.ok else result.message,
"message": result.message,
"configs": list_user_awg_configs(db, user.telegram_id),
},
status=status,
)
async def handle_awg_qr(request: web.Request) -> web.Response:
db: Database = request.app["db"]
try:
user = _require_user(request)
except web.HTTPUnauthorized:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
try:
config_id = int(request.match_info["config_id"])
except (KeyError, ValueError):
return web.json_response({"ok": False, "error": "bad id"}, status=400)
row = db.get_awg_config(config_id, telegram_id=user.telegram_id)
if not row:
return web.json_response({"ok": False, "error": "not found"}, status=404)
payload = str(row.get("vpn_link") or row.get("config_text") or "").strip()
if not payload:
return web.json_response({"ok": False, "error": "no config"}, status=400)
try:
svg = make_qr_svg(payload, scale=5, border=2)
except Exception as exc: # noqa: BLE001
logger.exception("AWG QR generate failed")
return web.json_response({"ok": False, "error": str(exc)}, status=500)
return web.Response(
body=svg,
content_type="image/svg+xml",
headers={
"Cache-Control": "private, max-age=300",
"Content-Disposition": f'inline; filename="awg-{config_id}.svg"',
},
)
async def start_webapp_site(app: web.Application, host: str, port: int) -> web.AppRunner:
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, host=host, port=port)
await site.start()
logger.info("Mini App HTTP on http://%s:%s", host, port)
return runner