Files
remnawave-minishop/bot/app/web/subscription_webapp.py
T

4284 lines
167 KiB
Python

import asyncio
import base64
import hashlib
import hmac
import io
import ipaddress
import json
import logging
import os
import re
import secrets
import socket
import subprocess
import time
from collections import deque
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from aiohttp import ClientSession, ClientTimeout, web
from aiogram import Bot, Dispatcher
from aiogram.types import LabeledPrice
from pydantic import BaseModel, ConfigDict, EmailStr, ValidationError, constr, field_validator
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
from bot.app.web.admin_api import (
admin_auth_middleware,
setup_admin_routes,
)
from bot.app.web.webapp_auth import (
create_signed_telegram_oauth_state,
create_telegram_oauth_nonce,
create_webapp_session_token,
validate_telegram_oauth_id_token,
validate_telegram_login_widget_data,
validate_telegram_webapp_init_data,
verify_signed_telegram_oauth_state,
verify_telegram_oauth_nonce,
verify_webapp_session_token,
)
from bot.services.crypto_pay_service import CryptoPayService
from bot.services.email_auth_service import EmailAuthService, normalize_email
from bot.services.email_templates import render_account_merged
from bot.services.freekassa_service import FreeKassaService
from bot.services.platega_service import PlategaService
from bot.services.promo_code_service import PromoCodeService
from bot.services.referral_service import ReferralService
from bot.services.severpay_service import SeverPayService
from bot.services.subscription_service import SubscriptionService
from bot.services.yookassa_service import YooKassaService
from bot.utils.config_link import prepare_config_links
from bot.utils.text_sanitizer import sanitize_display_name, sanitize_username
from bot.utils.request_security import request_client_ip
from config.settings import Settings
from db.dal import payment_dal, subscription_dal, user_dal
from db.dal.user_dal import UserMergeConflictError
from db.models import Payment, User, UserTelegramAvatar
logger = logging.getLogger(__name__)
TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "subscription_webapp.html"
ASSET_DIR = TEMPLATE_PATH.parent
WEBAPP_LOGO_PROXY_PATH = "/webapp-logo"
WEBAPP_CONFIG_PLACEHOLDER = "<!-- WEBAPP_CONFIG_SCRIPT -->"
WEBAPP_I18N_PLACEHOLDER = "<!-- WEBAPP_I18N_SCRIPT -->"
WEBAPP_JS_PLACEHOLDER = "<!-- WEBAPP_JS_SCRIPT -->"
APP_REPOSITORY_URL = "https://github.com/3252a8/remnawave-minishop"
DEV_MOCK_START_MARKER = "<!-- WEBAPP_DEV_MOCK_START -->"
DEV_MOCK_END_MARKER = "<!-- WEBAPP_DEV_MOCK_END -->"
WEBAPP_RATE_LIMIT_WINDOW_SECONDS = 60
WEBAPP_RATE_LIMIT_MAX_REQUESTS = 30
WEBAPP_LOGO_MAX_BYTES = 2 * 1024 * 1024
WEBAPP_TELEGRAM_AVATAR_MAX_BYTES = 128 * 1024
WEBAPP_TELEGRAM_AVATAR_REFRESH_SECONDS = 24 * 60 * 60
WEBAPP_TELEGRAM_AVATAR_FETCH_TIMEOUT_SECONDS = 4
WEBAPP_SESSION_COOKIE_NAME = "rw_webapp_session"
WEBAPP_CSRF_COOKIE_NAME = "rw_webapp_csrf"
WEBAPP_TELEGRAM_OAUTH_STATE_COOKIE_NAME = "rw_tg_oauth_state"
WEBAPP_CSRF_HEADER_NAME = "X-CSRF-Token"
WEBAPP_STATE_CHANGING_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
_APP_VERSION_CACHE: Optional[str] = None
WEBAPP_CSRF_EXEMPT_PATHS = {
"/api/auth/telegram/nonce",
"/api/auth/token",
"/api/auth/email/request",
"/api/auth/email/verify",
"/api/auth/email/magic",
"/api/auth/logout",
}
class WebAppEmailPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
email: EmailStr
@field_validator("email")
@classmethod
def _normalize_and_limit_email(cls, value: EmailStr) -> str:
normalized = normalize_email(str(value))
if len(normalized) > 254:
raise ValueError("email_too_long")
return normalized
class WebAppEmailCodePayload(WebAppEmailPayload):
code: str = ""
class WebAppEmailMagicPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
token: constr(min_length=8, max_length=512)
class WebAppPaymentCreatePayload(BaseModel):
model_config = ConfigDict(extra="ignore")
method: str = ""
months: Any = None
traffic_gb: Any = None
device_count: Any = None
tariff_key: Optional[constr(max_length=128)] = None
sale_mode: Optional[constr(max_length=64)] = None
description: Optional[constr(max_length=4096)] = None
comment: Optional[constr(max_length=4096)] = None
note: Optional[constr(max_length=4096)] = None
class WebAppTariffChangePayload(BaseModel):
model_config = ConfigDict(extra="ignore")
tariff_key: constr(min_length=1, max_length=128)
mode: constr(min_length=1, max_length=64)
class WebAppLanguagePayload(BaseModel):
model_config = ConfigDict(extra="ignore")
language: constr(min_length=2, max_length=16)
class WebAppDeviceDisconnectPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
token: constr(min_length=8, max_length=128)
_SHARED_HTTP_SESSION: Optional[ClientSession] = None
_SHARED_HTTP_SESSION_LOCK = asyncio.Lock()
def create_subscription_webapp_application(
dp: Dispatcher,
bot: Bot,
settings: Settings,
async_session_factory: sessionmaker,
) -> web.Application:
app = web.Application(
middlewares=[
_security_headers_middleware,
_csrf_protection_middleware,
admin_auth_middleware,
]
)
app["bot"] = bot
app["dp"] = dp
app["settings"] = settings
app["async_session_factory"] = async_session_factory
app["i18n"] = dp.get("i18n_instance")
app["email_auth_service"] = EmailAuthService(settings)
app["webapp_logo_cache"] = None
app["webapp_logo_cache_lock"] = asyncio.Lock()
app["webapp_settings_cache"] = {"ts": 0.0, "data": {}}
app["webapp_rate_limit_buckets"] = {}
app["webapp_rate_limit_lock"] = asyncio.Lock()
async def _startup(app_obj: web.Application) -> None:
await _ensure_shared_http_session()
async def _shutdown(app_obj: web.Application) -> None:
await _close_shared_http_session()
app.on_startup.append(_startup)
app.on_shutdown.append(_shutdown)
for key in (
"subscription_service",
"yookassa_service",
"freekassa_service",
"cryptopay_service",
"platega_service",
"severpay_service",
"promo_code_service",
"referral_service",
"panel_service",
):
if hasattr(dp, "workflow_data") and key in dp.workflow_data: # type: ignore[attr-defined]
app[key] = dp.workflow_data[key] # type: ignore[index]
if hasattr(dp, "workflow_data") and "bot_username" in dp.workflow_data: # type: ignore[attr-defined]
app["bot_username"] = dp.workflow_data["bot_username"] # type: ignore[index]
setup_subscription_webapp_routes(app)
return app
def setup_subscription_webapp_routes(app: web.Application) -> None:
app.router.add_get("/", index_route)
app.router.add_get("/home", index_route)
app.router.add_get("/invite", index_route)
app.router.add_get("/devices", index_route)
app.router.add_get("/settings", index_route)
app.router.add_get("/admin", index_route)
app.router.add_get("/admin/{section:[a-z][a-z0-9_-]*}", index_route)
app.router.add_get("/admin/users/{user_id:-?[0-9]+}", index_route)
app.router.add_get("/auth/telegram/start", telegram_oauth_start_route)
app.router.add_get("/auth/telegram/callback", telegram_oauth_callback_route)
app.router.add_get("/health", health_route)
app.router.add_get(WEBAPP_LOGO_PROXY_PATH, webapp_logo_route)
app.router.add_get("/subscription_webapp.css", css_asset_route)
app.router.add_get("/subscription_webapp.min.{asset_hash}.js", js_asset_route)
app.router.add_get("/subscription_webapp.js", js_asset_route)
app.router.add_post("/api/auth/telegram/nonce", telegram_oauth_nonce_route)
app.router.add_post("/api/auth/token", auth_token_route)
app.router.add_post("/api/auth/email/request", email_auth_request_route)
app.router.add_post("/api/auth/email/verify", email_auth_verify_route)
app.router.add_post("/api/auth/email/magic", email_auth_magic_route)
app.router.add_post("/api/auth/logout", logout_route)
app.router.add_get("/api/me", me_route)
app.router.add_get("/api/account/avatar", account_avatar_route)
app.router.add_post("/api/account/language", account_language_route)
app.router.add_post("/api/account/email/request", account_email_request_route)
app.router.add_post("/api/account/email/verify", account_email_verify_route)
app.router.add_post("/api/account/telegram/link", account_telegram_link_route)
app.router.add_post("/api/promo/apply", apply_promo_route)
app.router.add_post("/api/trial/activate", activate_trial_route)
app.router.add_get("/api/devices", devices_route)
app.router.add_post("/api/devices/disconnect", disconnect_device_route)
app.router.add_get("/api/devices/topup-options", device_topup_options_route)
app.router.add_get("/api/tariffs/topup-options", tariff_topup_options_route)
app.router.add_get("/api/tariffs/change-options", tariff_change_options_route)
app.router.add_post("/api/tariffs/change", tariff_change_route)
app.router.add_post("/api/tariffs/change-payment", tariff_change_payment_route)
app.router.add_post("/api/payments", create_payment_route)
app.router.add_get("/api/payments/{payment_id}", payment_status_route)
setup_admin_routes(app)
async def health_route(request: web.Request) -> web.Response:
return web.json_response({"ok": True})
async def css_asset_route(request: web.Request) -> web.Response:
return await _serve_template_asset(request, "subscription_webapp.css", "text/css")
def _resolve_webapp_logo_url(settings: Settings) -> str:
raw_logo_url = (settings.WEBAPP_LOGO_URL or "").strip()
if not raw_logo_url:
return ""
parsed_logo_url = urlsplit(raw_logo_url)
if parsed_logo_url.scheme == "https":
return WEBAPP_LOGO_PROXY_PATH
if parsed_logo_url.scheme in {"http", "data"}:
return raw_logo_url
if raw_logo_url.startswith("/"):
return raw_logo_url
return ""
def _resolve_telegram_bot_id(bot_token: str) -> Optional[int]:
token_prefix = str(bot_token or "").strip().split(":", 1)[0]
if not token_prefix.isdigit():
return None
try:
return int(token_prefix)
except ValueError:
return None
def _resolve_telegram_oauth_client_id(settings: Settings) -> Optional[int]:
configured_client_id = getattr(settings, "TELEGRAM_OAUTH_CLIENT_ID", None)
if configured_client_id:
try:
return int(configured_client_id)
except (TypeError, ValueError):
return None
return _resolve_telegram_bot_id(settings.BOT_TOKEN)
def _resolve_telegram_oauth_request_access(settings: Settings) -> List[str]:
raw_value = str(getattr(settings, "TELEGRAM_OAUTH_REQUEST_ACCESS", "") or "")
allowed = {"write", "phone"}
scopes = []
for item in raw_value.split(","):
value = item.strip().lower()
if value in allowed and value not in scopes:
scopes.append(value)
return scopes
def _public_webapp_base_url(settings: Settings, request: web.Request) -> str:
configured_url = str(settings.SUBSCRIPTION_MINI_APP_URL or "").strip()
if configured_url:
parsed_url = urlsplit(configured_url)
if parsed_url.scheme and parsed_url.netloc:
return f"{parsed_url.scheme}://{parsed_url.netloc}"
scheme = request.headers.get("X-Forwarded-Proto") or request.scheme
host = request.headers.get("X-Forwarded-Host") or request.headers.get("Host") or request.host
return f"{scheme}://{host}".rstrip("/")
def _telegram_oauth_callback_url(settings: Settings, request: web.Request) -> str:
return f"{_public_webapp_base_url(settings, request)}/auth/telegram/callback"
def _telegram_oauth_redirect_url(path: str = "/", *, status: Optional[str] = None) -> str:
target_path = path if path.startswith("/") else "/"
if target_path not in {"/", "/settings"}:
target_path = "/"
if not status:
return target_path
separator = "&" if "?" in target_path else "?"
return f"{target_path}{separator}telegram_auth={status}"
def _set_telegram_oauth_state_cookie(
response: web.StreamResponse,
settings: Settings,
payload: Dict[str, Any],
) -> None:
max_age = max(60, int(settings.WEBAPP_LOGIN_TOKEN_TTL_SECONDS))
response.set_cookie(
WEBAPP_TELEGRAM_OAUTH_STATE_COOKIE_NAME,
create_signed_telegram_oauth_state(settings, payload, ttl_seconds=max_age),
httponly=True,
secure=True,
samesite="Lax",
path="/auth/telegram",
max_age=max_age,
)
def _clear_telegram_oauth_state_cookie(response: web.StreamResponse) -> None:
response.set_cookie(
WEBAPP_TELEGRAM_OAUTH_STATE_COOKIE_NAME,
"",
httponly=True,
secure=True,
samesite="Lax",
path="/auth/telegram",
max_age=0,
)
def _read_telegram_oauth_state_payload(
request: web.Request,
state_token: str,
) -> Optional[Dict[str, Any]]:
settings: Settings = request.app["settings"]
signed_payload = request.cookies.get(WEBAPP_TELEGRAM_OAUTH_STATE_COOKIE_NAME, "")
payload = verify_signed_telegram_oauth_state(settings, signed_payload)
if not payload:
return None
expected_state = str(payload.get("state") or "")
if not expected_state or not hmac.compare_digest(expected_state, state_token):
return None
return payload
def _urlsafe_sha256(value: str) -> str:
digest = hashlib.sha256(value.encode("ascii")).digest()
return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
async def _exchange_telegram_oauth_code(
request: web.Request,
*,
code: str,
code_verifier: str,
redirect_uri: str,
) -> Optional[Dict[str, Any]]:
settings: Settings = request.app["settings"]
client_id = _resolve_telegram_oauth_client_id(settings)
client_secret = str(getattr(settings, "TELEGRAM_OAUTH_CLIENT_SECRET", "") or "").strip()
if not client_id or not client_secret or not code or not code_verifier:
return None
credentials = base64.b64encode(f"{client_id}:{client_secret}".encode("utf-8")).decode("ascii")
session = await _get_shared_http_session()
try:
async with session.post(
"https://oauth.telegram.org/token",
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
"client_id": str(client_id),
"code_verifier": code_verifier,
},
headers={
"Authorization": f"Basic {credentials}",
"Content-Type": "application/x-www-form-urlencoded",
},
timeout=ClientTimeout(total=15),
) as response:
payload = await response.json(content_type=None)
if response.status >= 400:
logger.warning(
"Telegram OAuth token exchange failed with HTTP %s: %s",
response.status,
payload,
)
return None
return payload if isinstance(payload, dict) else None
except Exception as exc:
logger.warning("Telegram OAuth token exchange failed: %s", exc)
return None
async def webapp_logo_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
raw_logo_url = (settings.WEBAPP_LOGO_URL or "").strip()
if not raw_logo_url:
raise web.HTTPNotFound(text="webapp_logo_not_configured")
parsed_logo_url = urlsplit(raw_logo_url)
if parsed_logo_url.scheme != "https" or not parsed_logo_url.hostname:
raise web.HTTPNotFound(text="webapp_logo_not_proxied")
if not await _hostname_resolves_to_public_address(parsed_logo_url.hostname):
raise web.HTTPNotFound(text="webapp_logo_not_proxied")
source_logo_url = raw_logo_url
logo_cache: Optional[Tuple[bytes, str]] = request.app.get("webapp_logo_cache")
if logo_cache is None:
cache_lock: asyncio.Lock = request.app["webapp_logo_cache_lock"]
async with cache_lock:
logo_cache = request.app.get("webapp_logo_cache")
if logo_cache is None:
logo_cache = await _fetch_webapp_logo(source_logo_url)
request.app["webapp_logo_cache"] = logo_cache
if not logo_cache:
raise web.HTTPNotFound(text="webapp_logo_unavailable")
body, content_type = logo_cache
response = web.Response(body=body, content_type=content_type)
response.headers["Cache-Control"] = "public, max-age=3600"
return response
async def _fetch_webapp_logo(logo_url: str) -> Optional[Tuple[bytes, str]]:
"""Fetch and cache the configured logo on the server side."""
try:
session = await _get_shared_http_session()
timeout = ClientTimeout(total=3)
async with session.get(logo_url, allow_redirects=False, timeout=timeout) as response:
if response.status != 200:
logger.warning(
"WEBAPP_LOGO_URL returned HTTP %s; keeping the logo hidden.",
response.status,
)
return None
content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
if content_type and not content_type.startswith("image/"):
logger.warning(
"WEBAPP_LOGO_URL returned non-image content type %s; keeping the logo hidden.",
content_type,
)
return None
body = bytearray()
async for chunk in response.content.iter_chunked(64 * 1024):
body.extend(chunk)
if len(body) > WEBAPP_LOGO_MAX_BYTES:
logger.warning("WEBAPP_LOGO_URL exceeded the 2 MiB limit.")
return None
if not body:
logger.warning("WEBAPP_LOGO_URL returned an empty response body.")
return None
return bytes(body), content_type or "image/png"
except Exception as exc:
logger.warning("Failed to fetch WEBAPP_LOGO_URL: %s", exc)
return None
async def _get_shared_http_session() -> ClientSession:
global _SHARED_HTTP_SESSION
async with _SHARED_HTTP_SESSION_LOCK:
if _SHARED_HTTP_SESSION is None or _SHARED_HTTP_SESSION.closed:
_SHARED_HTTP_SESSION = ClientSession(
timeout=ClientTimeout(total=30),
headers={
"User-Agent": "Mozilla/5.0",
"Accept": "application/javascript,text/javascript,*/*;q=0.8",
},
)
return _SHARED_HTTP_SESSION
async def _ensure_shared_http_session() -> None:
await _get_shared_http_session()
async def _close_shared_http_session() -> None:
global _SHARED_HTTP_SESSION
async with _SHARED_HTTP_SESSION_LOCK:
if _SHARED_HTTP_SESSION and not _SHARED_HTTP_SESSION.closed:
await _SHARED_HTTP_SESSION.close()
_SHARED_HTTP_SESSION = None
async def _hostname_resolves_to_public_address(hostname: str) -> bool:
if not hostname:
return False
try:
ip_obj = ipaddress.ip_address(hostname)
return not (
ip_obj.is_private
or ip_obj.is_loopback
or ip_obj.is_link_local
or ip_obj.is_unspecified
or ip_obj.is_reserved
)
except ValueError:
pass
loop = asyncio.get_running_loop()
try:
resolved = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
except Exception:
return False
found_public_ip = False
for entry in resolved:
sockaddr = entry[4]
if not sockaddr:
continue
candidate = sockaddr[0]
try:
ip_obj = ipaddress.ip_address(candidate)
except ValueError:
continue
if (
ip_obj.is_private
or ip_obj.is_loopback
or ip_obj.is_link_local
or ip_obj.is_unspecified
or ip_obj.is_reserved
):
return False
found_public_ip = True
return found_public_ip
@web.middleware
async def _security_headers_middleware(request: web.Request, handler):
request["csp_nonce"] = secrets.token_urlsafe(16)
try:
response = await handler(request)
except web.HTTPException as exc:
response = exc
nonce = request.get("csp_nonce", "")
response.headers.setdefault(
"Content-Security-Policy",
(
"default-src 'self'; "
f"script-src 'self' 'nonce-{nonce}' 'unsafe-eval' https://telegram.org; "
"frame-src https://oauth.telegram.org; "
"frame-ancestors https://web.telegram.org https://t.me; "
"style-src 'self' 'unsafe-inline'; "
"font-src 'self' https://cdn.jsdelivr.net data:; "
"img-src 'self' data: https: http:; "
"connect-src 'self' https://oauth.telegram.org; "
"object-src 'none'; "
"base-uri 'self'; "
"form-action 'self'"
),
)
response.headers.setdefault("Referrer-Policy", "no-referrer")
response.headers.setdefault("X-Content-Type-Options", "nosniff")
response.headers.setdefault(
"Permissions-Policy",
(
"accelerometer=(), autoplay=(), camera=(), display-capture=(), "
"encrypted-media=(), geolocation=(), gyroscope=(), magnetometer=(), "
"microphone=(), midi=(), payment=(), usb=()"
),
)
return response
@web.middleware
async def _csrf_protection_middleware(request: web.Request, handler):
settings: Settings = request.app["settings"]
header = request.headers.get("Authorization", "")
prefix = "Bearer "
if header.startswith(prefix):
if verify_webapp_session_token(settings, header[len(prefix):].strip()):
return await handler(request)
if (
request.method in WEBAPP_STATE_CHANGING_METHODS
and request.path not in WEBAPP_CSRF_EXEMPT_PATHS
and request.cookies.get(WEBAPP_SESSION_COOKIE_NAME)
):
csrf_cookie = request.cookies.get(WEBAPP_CSRF_COOKIE_NAME, "")
csrf_header = request.headers.get(WEBAPP_CSRF_HEADER_NAME, "")
if (
not csrf_cookie
or not csrf_header
or not hmac.compare_digest(csrf_header, csrf_cookie)
):
return _json_error(403, "csrf_failed", "Invalid CSRF token")
return await handler(request)
def _get_cached_webapp_settings(request: web.Request) -> Dict[str, Any]:
settings: Settings = request.app["settings"]
cache = request.app["webapp_settings_cache"]
now = time.monotonic()
if now - float(cache.get("ts", 0.0)) >= 60 or not cache.get("data"):
cache["data"] = {
"logo_url": _resolve_webapp_logo_url(settings),
"subscription_options": settings.subscription_options,
"stars_subscription_options": settings.stars_subscription_options,
"traffic_packages": settings.traffic_packages,
"stars_traffic_packages": settings.stars_traffic_packages,
"support_url": settings.SUPPORT_LINK or "",
"terms_url": settings.TERMS_OF_SERVICE_URL or "",
"privacy_policy_url": settings.PRIVACY_POLICY_URL or "",
"user_agreement_url": settings.USER_AGREEMENT_URL or "",
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"email_auth_enabled": settings.email_auth_configured,
"language": _normalize_language(settings.DEFAULT_LANGUAGE),
}
cache["ts"] = now
return cache["data"]
def _run_git_command(*args: str) -> str:
repo_root = Path(__file__).resolve().parents[3]
try:
result = subprocess.run(
["git", *args],
cwd=repo_root,
check=True,
capture_output=True,
text=True,
timeout=1.5,
)
except (OSError, subprocess.SubprocessError):
return ""
return result.stdout.strip()
def _resolve_app_version() -> str:
global _APP_VERSION_CACHE
if _APP_VERSION_CACHE:
return _APP_VERSION_CACHE
env_version = os.getenv("REMNAWAVE_MINISHOP_VERSION", "").strip()
if env_version:
_APP_VERSION_CACHE = env_version
return env_version
build_version_path = Path(__file__).resolve().parents[3] / ".build-version"
try:
build_version = build_version_path.read_text(encoding="utf-8").strip()
except OSError:
build_version = ""
if build_version:
_APP_VERSION_CACHE = build_version
return build_version
tag = _run_git_command("describe", "--tags", "--abbrev=0")
sha = _run_git_command("rev-parse", "--short", "HEAD")
dirty = bool(_run_git_command("status", "--porcelain"))
if tag and sha:
commits_since_tag = _run_git_command("rev-list", f"{tag}..HEAD", "--count")
if commits_since_tag and commits_since_tag != "0":
version = f"{tag}+{commits_since_tag}.g{sha}"
else:
version = tag
elif sha:
version = f"dev+g{sha}"
else:
version = "dev+unknown"
if dirty:
version = f"{version}-dirty"
_APP_VERSION_CACHE = version
return version
async def _enforce_webapp_rate_limit(
request: web.Request,
*,
user_id: int,
action: str,
) -> Optional[web.Response]:
settings: Settings = request.app["settings"]
ip_address = request_client_ip(request, trusted_proxies=settings.trusted_proxies) or request.remote or "unknown"
key = f"{action}:{ip_address}:{int(user_id)}"
buckets: Dict[str, deque[float]] = request.app["webapp_rate_limit_buckets"]
lock: asyncio.Lock = request.app["webapp_rate_limit_lock"]
now = time.monotonic()
async with lock:
bucket = buckets.setdefault(key, deque())
while bucket and now - bucket[0] >= WEBAPP_RATE_LIMIT_WINDOW_SECONDS:
bucket.popleft()
if not bucket:
buckets.pop(key, None)
bucket = buckets.setdefault(key, deque())
if len(bucket) >= WEBAPP_RATE_LIMIT_MAX_REQUESTS:
retry_after = max(
1,
int(WEBAPP_RATE_LIMIT_WINDOW_SECONDS - (now - bucket[0])),
) if bucket else WEBAPP_RATE_LIMIT_WINDOW_SECONDS
return web.json_response(
{
"ok": False,
"error": "rate_limited",
"retry_after": retry_after,
},
status=429,
headers={"Retry-After": str(retry_after)},
)
bucket.append(now)
return None
async def js_asset_route(request: web.Request) -> web.Response:
asset_hash = request.match_info.get("asset_hash")
filename = (
f"subscription_webapp.min.{asset_hash}.js"
if asset_hash
else "subscription_webapp.js"
)
response = await _serve_template_asset(
request,
filename,
"application/javascript",
strip_dev_mock=not asset_hash,
)
response.headers["Cache-Control"] = (
"public, max-age=31536000, immutable" if asset_hash else "no-cache"
)
return response
async def index_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.WEBAPP_ENABLED:
raise web.HTTPNotFound(text="webapp_disabled")
html = TEMPLATE_PATH.read_text(encoding="utf-8")
cached = _get_cached_webapp_settings(request)
config = {
"title": settings.WEBAPP_TITLE,
"primaryColor": settings.WEBAPP_PRIMARY_COLOR,
"logoUrl": cached["logo_url"],
"logoEmoji": settings.WEBAPP_LOGO_EMOJI,
"apiBase": "/api",
"telegramLoginBotUsername": request.app.get("bot_username") or "",
"telegramLoginBotId": _resolve_telegram_bot_id(settings.BOT_TOKEN) or 0,
"telegramOAuthClientId": _resolve_telegram_oauth_client_id(settings) or 0,
"telegramOAuthRequestAccess": _resolve_telegram_oauth_request_access(settings),
"supportUrl": cached["support_url"],
"termsUrl": cached["terms_url"],
"privacyPolicyUrl": cached["privacy_policy_url"],
"userAgreementUrl": cached["user_agreement_url"],
"currency": cached["currency"],
"language": cached["language"],
"emailAuthEnabled": cached["email_auth_enabled"],
"appVersion": _resolve_app_version(),
"appRepositoryUrl": APP_REPOSITORY_URL,
}
html = _strip_marked_block(html, DEV_MOCK_START_MARKER, DEV_MOCK_END_MARKER)
i18n_instance: Optional[object] = request.app.get("i18n")
i18n_payload = getattr(i18n_instance, "locales_data", {}) if i18n_instance else {}
nonce = request.get("csp_nonce", "")
html = html.replace(
WEBAPP_CONFIG_PLACEHOLDER,
(
f"<script id=\"webapp-config\" type=\"application/json\" nonce=\"{nonce}\">"
+ json.dumps(config, ensure_ascii=False, separators=(",", ":"))
+ "</script>"
),
)
html = html.replace(
WEBAPP_I18N_PLACEHOLDER,
(
f"<script id=\"i18n\" type=\"application/json\" nonce=\"{nonce}\">"
+ json.dumps(i18n_payload, ensure_ascii=False, separators=(",", ":"))
+ "</script>"
),
)
html = html.replace(
WEBAPP_JS_PLACEHOLDER,
f'<script src="/{_resolve_webapp_js_asset_name()}" defer></script>',
)
return web.Response(text=html, content_type="text/html", charset="utf-8")
async def _serve_template_asset(
request: web.Request,
filename: str,
content_type: str,
*,
strip_dev_mock: bool = False,
) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.WEBAPP_ENABLED:
raise web.HTTPNotFound(text="webapp_disabled")
path = ASSET_DIR / filename
text = path.read_text(encoding="utf-8")
if strip_dev_mock:
text = _strip_marked_block(
text,
"/* WEBAPP_DEV_MOCK_START */",
"/* WEBAPP_DEV_MOCK_END */",
)
return web.Response(text=text, content_type=content_type, charset="utf-8")
async def telegram_oauth_nonce_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
client_id = _resolve_telegram_oauth_client_id(settings)
if not client_id:
return _json_error(400, "telegram_oauth_not_configured", "Telegram OAuth is not configured")
nonce = create_telegram_oauth_nonce(
settings,
ttl_seconds=settings.WEBAPP_LOGIN_TOKEN_TTL_SECONDS,
)
return web.json_response(
{
"ok": True,
"nonce": nonce,
"client_id": client_id,
"request_access": _resolve_telegram_oauth_request_access(settings),
}
)
async def telegram_oauth_start_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
client_id = _resolve_telegram_oauth_client_id(settings)
client_secret = str(getattr(settings, "TELEGRAM_OAUTH_CLIENT_SECRET", "") or "").strip()
if not client_id or not client_secret:
raise web.HTTPFound(_telegram_oauth_redirect_url("/", status="not_configured"))
purpose = str(request.query.get("purpose") or "login").strip().lower()
if purpose not in {"login", "link"}:
purpose = "login"
current_user_id = _extract_authenticated_user_id(request)
if purpose == "link" and not current_user_id:
raise web.HTTPFound(_telegram_oauth_redirect_url("/", status="unauthorized"))
code_verifier = secrets.token_urlsafe(32)
code_challenge = _urlsafe_sha256(code_verifier)
nonce = secrets.token_urlsafe(16)
state = secrets.token_urlsafe(16)
state_payload = {
"state": state,
"purpose": purpose,
"user_id": int(current_user_id) if current_user_id else None,
"referral_code": str(request.query.get("referral_code") or "")[:128],
"code_verifier": code_verifier,
"nonce": nonce,
}
scopes = ["openid", "profile"]
for permission in _resolve_telegram_oauth_request_access(settings):
if permission == "phone":
scopes.append("phone")
elif permission == "write":
scopes.append("telegram:bot_access")
auth_query = urlencode(
{
"client_id": str(client_id),
"redirect_uri": _telegram_oauth_callback_url(settings, request),
"response_type": "code",
"scope": " ".join(scopes),
"state": state,
"nonce": nonce,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
)
response = web.HTTPFound(f"https://oauth.telegram.org/auth?{auth_query}")
_set_telegram_oauth_state_cookie(response, settings, state_payload)
raise response
async def telegram_oauth_callback_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
def redirect(path: str = "/", status: Optional[str] = None) -> web.HTTPFound:
response = web.HTTPFound(_telegram_oauth_redirect_url(path, status=status))
_clear_telegram_oauth_state_cookie(response)
return response
error = str(request.query.get("error") or "")
if error:
raise redirect("/", "cancelled")
code = str(request.query.get("code") or "")
state = _read_telegram_oauth_state_payload(request, str(request.query.get("state") or ""))
if not code or not state:
raise redirect("/", "invalid_state")
token_payload = await _exchange_telegram_oauth_code(
request,
code=code,
code_verifier=str(state.get("code_verifier") or ""),
redirect_uri=_telegram_oauth_callback_url(settings, request),
)
id_token = str(token_payload.get("id_token") or "") if token_payload else ""
telegram_user = await validate_telegram_oauth_id_token(
id_token,
client_id=int(_resolve_telegram_oauth_client_id(settings) or 0),
expected_nonce=str(state.get("nonce") or ""),
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
if not telegram_user:
raise redirect("/", "invalid_token")
purpose = str(state.get("purpose") or "login")
redirect_path = "/settings" if purpose == "link" else "/"
async_session_factory: sessionmaker = request.app["async_session_factory"]
final_user_id: Optional[int] = None
async with async_session_factory() as session:
try:
if purpose == "link":
current_user_id = int(state.get("user_id") or 0)
db_user = await _link_telegram_to_user(
request,
session,
current_user_id=current_user_id,
telegram_user=telegram_user,
settings=settings,
)
else:
db_user = await _ensure_user_from_telegram(
session,
telegram_user,
settings,
referral_param=str(state.get("referral_code") or ""),
)
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
str(state.get("referral_code") or "") or telegram_user.get("start_param"),
)
if getattr(db_user, "_webapp_created", False) or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
str(state.get("referral_code") or "") or telegram_user.get("start_param"),
)
if db_user.is_banned:
await session.rollback()
raise redirect("/", "banned")
final_user_id = int(db_user.user_id)
await session.commit()
except web.HTTPFound:
raise
except UserMergeConflictError:
await session.rollback()
raise redirect(redirect_path, "merge_conflict")
except Exception:
await session.rollback()
logger.exception("Telegram OAuth callback failed")
raise redirect(redirect_path, "failed")
token = create_webapp_session_token(settings, int(final_user_id))
response = web.HTTPFound(_telegram_oauth_redirect_url(redirect_path, status="success"))
_clear_telegram_oauth_state_cookie(response)
_set_webapp_auth_cookies(response, settings, token, secrets.token_hex(32))
raise response
async def _validate_telegram_auth_payload(
request: web.Request,
payload: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
settings: Settings = request.app["settings"]
init_data = str(payload.get("init_data") or "")
if init_data:
return validate_telegram_webapp_init_data(
init_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
oauth_id_token = str(payload.get("id_token") or "")
if oauth_id_token:
nonce = str(payload.get("nonce") or "")
client_id = _resolve_telegram_oauth_client_id(settings)
if not client_id or not verify_telegram_oauth_nonce(settings, nonce):
return None
return await validate_telegram_oauth_id_token(
oauth_id_token,
client_id=client_id,
expected_nonce=nonce,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
auth_data = payload.get("auth_data")
if auth_data is not None:
return validate_telegram_login_widget_data(
auth_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
return None
def _resolve_webapp_js_asset_name() -> str:
minified_assets = []
for path in ASSET_DIR.glob("subscription_webapp.min.*.js"):
try:
minified_assets.append((path.stat().st_mtime, path.name))
except OSError:
continue
if minified_assets:
minified_assets.sort(reverse=True)
return minified_assets[0][1]
return "subscription_webapp.js"
def _strip_marked_block(html: str, start_marker: str, end_marker: str) -> str:
start = html.find(start_marker)
if start == -1:
return html
end = html.find(end_marker, start)
if end == -1:
return html[:start]
return html[:start] + html[end + len(end_marker):]
async def auth_token_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
referral_param = str(payload.get("referral_code") or payload.get("start_param") or "")
telegram_user = await _validate_telegram_auth_payload(request, payload)
if not telegram_user:
return _json_error(401, "invalid_auth", "Invalid Telegram auth data")
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=int(telegram_user.get("id") or 0),
action="auth_token",
)
if rate_limit_response:
return rate_limit_response
async_session_factory: sessionmaker = request.app["async_session_factory"]
authenticated_user_id: Optional[int] = None
async with async_session_factory() as session:
try:
db_user = await _ensure_user_from_telegram(
session,
telegram_user,
settings,
referral_param=referral_param,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
referral_param or telegram_user.get("start_param"),
)
if getattr(db_user, "_webapp_created", False) or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
referral_param or telegram_user.get("start_param"),
)
authenticated_user_id = int(db_user.user_id)
await session.commit()
except Exception as exc:
await session.rollback()
logger.exception("WebApp auth failed")
return _json_error(500, "auth_failed", "Auth failed")
token = create_webapp_session_token(settings, int(authenticated_user_id))
return _build_webapp_auth_response(settings, {"ok": True}, token=token)
async def logout_route(request: web.Request) -> web.Response:
response = web.json_response({"ok": True})
_clear_webapp_auth_cookies(response)
return response
async def email_auth_request_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailPayload, payload)
if validation_error:
return validation_error
email = email_payload.email
lang = _normalize_language(str(payload.get("language") or settings.DEFAULT_LANGUAGE))
return await _request_email_code(
request,
email=email,
purpose="login",
language_code=lang,
target_user_id=None,
)
async def email_auth_verify_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailCodePayload, payload)
if validation_error:
return validation_error
email = email_payload.email
code = str(email_payload.code or "")
referral_param = str(payload.get("referral_code") or payload.get("start_param") or "")
email_service: EmailAuthService = request.app["email_auth_service"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
created_user = False
new_user_referrer_id: Optional[int] = None
async with async_session_factory() as session:
try:
verify_result = await email_service.verify_code(
session,
email=email,
purpose="login",
code=code,
target_user_id=None,
)
if not verify_result.ok:
await session.commit()
status = 429 if verify_result.error == "rate_limited" else 400
return web.json_response(
{
"ok": False,
"error": verify_result.error or "invalid_code",
"retry_after": verify_result.retry_after,
"message": "Invalid code",
},
status=status,
)
db_user = await user_dal.get_user_by_email(session, email)
if not db_user:
referred_by_id = await _resolve_referrer_id(
session,
referral_param,
current_user_id=None,
)
db_user, _ = await user_dal.create_email_user(
session,
email=email,
language_code=_normalize_language(settings.DEFAULT_LANGUAGE),
email_verified_at=datetime.now(timezone.utc),
referred_by_id=referred_by_id,
)
created_user = True
new_user_referrer_id = referred_by_id
elif not db_user.email_verified_at:
db_user.email_verified_at = datetime.now(timezone.utc)
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
referral_param,
)
if created_user or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
referral_param,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
await session.commit()
except Exception as exc:
await session.rollback()
logger.exception("Email WebApp auth failed")
return _json_error(500, "auth_failed", "Auth failed")
if created_user:
try:
from bot.services.notification_service import NotificationService
bot: Bot = request.app["bot"]
notification_service = NotificationService(
bot,
settings,
request.app.get("i18n"),
)
await notification_service.notify_new_email_user_registration(
user_id=int(db_user.user_id),
email=email,
referred_by_id=new_user_referrer_id,
)
except Exception:
logger.exception("Failed to send new email user notification")
token = create_webapp_session_token(settings, int(db_user.user_id))
return _build_webapp_auth_response(
settings,
{
"ok": True,
"user_id": int(db_user.user_id),
"telegram_id": _telegram_id_for_user(db_user),
},
token=token,
)
async def email_auth_magic_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
magic_payload, validation_error = _validate_model_payload(WebAppEmailMagicPayload, payload)
if validation_error:
return validation_error
token_value = str(magic_payload.token).strip()
referral_param = str(payload.get("referral_code") or payload.get("start_param") or "")
email_service: EmailAuthService = request.app["email_auth_service"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
created_user = False
new_user_referrer_id: Optional[int] = None
verified_email: Optional[str] = None
async with async_session_factory() as session:
try:
magic_result = await email_service.verify_magic_token(
session,
token=token_value,
purpose="login",
target_user_id=None,
)
if not magic_result.ok:
await session.commit()
return web.json_response(
{
"ok": False,
"error": magic_result.error or "invalid_token",
"message": "Invalid login link",
},
status=400,
)
verified_email = magic_result.email or ""
db_user = await user_dal.get_user_by_email(session, verified_email)
if not db_user:
referred_by_id = await _resolve_referrer_id(
session,
referral_param,
current_user_id=None,
)
db_user, _ = await user_dal.create_email_user(
session,
email=verified_email,
language_code=_normalize_language(settings.DEFAULT_LANGUAGE),
email_verified_at=datetime.now(timezone.utc),
referred_by_id=referred_by_id,
)
created_user = True
new_user_referrer_id = referred_by_id
elif not db_user.email_verified_at:
db_user.email_verified_at = datetime.now(timezone.utc)
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
referral_param,
)
if created_user or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
referral_param,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
await session.commit()
except Exception:
await session.rollback()
logger.exception("Email magic-link auth failed")
return _json_error(500, "auth_failed", "Auth failed")
if created_user and verified_email:
try:
from bot.services.notification_service import NotificationService
bot: Bot = request.app["bot"]
notification_service = NotificationService(
bot,
settings,
request.app.get("i18n"),
)
await notification_service.notify_new_email_user_registration(
user_id=int(db_user.user_id),
email=verified_email,
referred_by_id=new_user_referrer_id,
)
except Exception:
logger.exception("Failed to send new email user notification")
session_token = create_webapp_session_token(settings, int(db_user.user_id))
return _build_webapp_auth_response(
settings,
{
"ok": True,
"user_id": int(db_user.user_id),
"telegram_id": _telegram_id_for_user(db_user),
},
token=session_token,
)
async def account_email_request_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailPayload, payload)
if validation_error:
return validation_error
email = email_payload.email
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
if db_user.email == email and db_user.email_verified_at:
return web.json_response({"ok": True, "already_linked": True})
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
return await _request_email_code(
request,
email=email,
purpose="link_email",
language_code=lang,
target_user_id=user_id,
)
async def account_email_verify_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="account_email_verify",
)
if rate_limit_response:
return rate_limit_response
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailCodePayload, payload)
if validation_error:
return validation_error
email = email_payload.email
code = str(email_payload.code or "")
email_service: EmailAuthService = request.app["email_auth_service"]
settings: Settings = request.app["settings"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
merge_notice: Optional[Dict[str, Any]] = None
source_panel_uuid: Optional[str] = None
final_user_id = user_id
final_email = email
final_telegram_id: Optional[int] = None
final_username: Optional[str] = None
final_first_name: Optional[str] = None
final_panel_uuid: Optional[str] = None
should_notify_email_linked = False
async with async_session_factory() as session:
try:
verify_result = await email_service.verify_code(
session,
email=email,
purpose="link_email",
code=code,
target_user_id=user_id,
)
if not verify_result.ok:
await session.commit()
status = 429 if verify_result.error == "rate_limited" else 400
return web.json_response(
{
"ok": False,
"error": verify_result.error or "invalid_code",
"retry_after": verify_result.retry_after,
"message": "Invalid code",
},
status=status,
)
current_user = await user_dal.get_user_by_id(session, user_id)
if not current_user or current_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
should_notify_email_linked = (
bool(_telegram_id_for_user(current_user))
and not current_user.email
)
existing_email_user = await user_dal.get_user_by_email(session, email)
if existing_email_user and existing_email_user.user_id != current_user.user_id:
source_panel_uuid = existing_email_user.panel_user_uuid
current_user = await user_dal.merge_users(
session,
source_user_id=existing_email_user.user_id,
target_user_id=current_user.user_id,
)
merge_notice = await _build_account_merge_notice(
session,
merged_user=current_user,
source_user_id=existing_email_user.user_id,
source_panel_uuid=source_panel_uuid,
settings=settings,
)
current_user.email = email
current_user.email_verified_at = datetime.now(timezone.utc)
await _sync_panel_identity_for_user(request, current_user)
await session.commit()
final_user_id = int(current_user.user_id)
final_telegram_id = _telegram_id_for_user(current_user)
final_username = current_user.username
final_first_name = current_user.first_name
final_panel_uuid = current_user.panel_user_uuid
if merge_notice:
merge_end_date_raw = merge_notice.get("final_end_date")
merge_end_date = (
datetime.fromisoformat(merge_end_date_raw)
if merge_end_date_raw
else None
)
await _sync_panel_identity_for_user(
request,
current_user,
expire_at=merge_end_date,
)
# Best-effort cleanup of the removed panel account after the DB merge.
if source_panel_uuid and final_panel_uuid and source_panel_uuid != final_panel_uuid:
subscription_service: SubscriptionService = request.app.get("subscription_service")
if subscription_service and subscription_service.panel_service:
try:
await subscription_service.panel_service.delete_user_from_panel(
source_panel_uuid,
log_response=False,
)
except Exception as exc:
logger.warning(
"Failed to delete merged source panel user %s: %s",
source_panel_uuid,
exc,
)
email_service: EmailAuthService = request.app.get("email_auth_service")
if email_service and final_email:
email_content = render_account_merged(
settings,
language_code=merge_notice.get("language") or settings.DEFAULT_LANGUAGE,
primary_user_id=merge_notice.get("primary_user_id"),
removed_user_id=merge_notice.get("removed_user_id"),
final_end_date_text=str(
merge_notice.get("final_end_date_text")
or merge_notice.get("final_end_date")
or ""
),
)
try:
await email_service.send_rendered_email(
email=final_email,
content=email_content,
)
except Exception as exc:
logger.warning(
"Failed to send account merge email to %s: %s",
final_email,
exc,
)
except UserMergeConflictError as exc:
await session.rollback()
return _json_error(409, "account_merge_conflict", str(exc))
except Exception as exc:
await session.rollback()
logger.exception("Email account link failed")
return _json_error(500, "link_failed", "Link failed")
if should_notify_email_linked:
try:
from bot.services.notification_service import NotificationService
bot: Bot = request.app["bot"]
notification_service = NotificationService(
bot,
settings,
request.app.get("i18n"),
)
await notification_service.notify_account_email_linked(
user_id=int(final_user_id),
email=final_email,
telegram_id=final_telegram_id,
username=final_username,
first_name=final_first_name,
)
except Exception:
logger.exception("Failed to send account email linked notification")
token = create_webapp_session_token(settings, int(final_user_id))
response_payload: Dict[str, Any] = {"ok": True}
if merge_notice:
response_payload["account_merge"] = merge_notice
response_payload["user_id"] = final_user_id
return _build_webapp_auth_response(settings, response_payload, token=token)
async def account_telegram_link_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
payload = await _read_json(request)
telegram_user = await _validate_telegram_auth_payload(request, payload)
if not telegram_user:
return _json_error(401, "invalid_auth", "Invalid Telegram auth data")
async_session_factory: sessionmaker = request.app["async_session_factory"]
merge_notice: Optional[Dict[str, Any]] = None
source_panel_uuid: Optional[str] = None
final_user_id = user_id
final_telegram_id: Optional[int] = None
final_email: Optional[str] = None
final_username: Optional[str] = None
final_first_name: Optional[str] = None
final_panel_uuid: Optional[str] = None
should_notify_telegram_linked = False
async with async_session_factory() as session:
try:
current_user_before_link = await user_dal.get_user_by_id(session, user_id)
if not current_user_before_link or current_user_before_link.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
should_notify_telegram_linked = (
bool(current_user_before_link.email)
and not _telegram_id_for_user(current_user_before_link)
)
source_panel_uuid = current_user_before_link.panel_user_uuid
db_user = await _link_telegram_to_user(
request,
session,
current_user_id=user_id,
telegram_user=telegram_user,
settings=settings,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
final_user_id = int(db_user.user_id)
final_telegram_id = _telegram_id_for_user(db_user)
final_email = db_user.email
final_username = db_user.username
final_first_name = db_user.first_name
final_panel_uuid = db_user.panel_user_uuid
if final_user_id != user_id:
merge_notice = await _build_account_merge_notice(
session,
merged_user=db_user,
source_user_id=user_id,
source_panel_uuid=source_panel_uuid,
settings=settings,
)
await session.commit()
if merge_notice:
merge_end_date_raw = merge_notice.get("final_end_date")
merge_end_date = (
datetime.fromisoformat(merge_end_date_raw)
if merge_end_date_raw
else None
)
await _sync_panel_identity_for_user(
request,
db_user,
expire_at=merge_end_date,
)
# Best-effort cleanup of the removed panel account after the DB merge.
if source_panel_uuid and final_panel_uuid and source_panel_uuid != final_panel_uuid:
subscription_service: SubscriptionService = request.app.get("subscription_service")
if subscription_service and subscription_service.panel_service:
try:
await subscription_service.panel_service.delete_user_from_panel(
source_panel_uuid,
log_response=False,
)
except Exception as exc:
logger.warning(
"Failed to delete merged source panel user %s: %s",
source_panel_uuid,
exc,
)
email_service: EmailAuthService = request.app.get("email_auth_service")
if email_service and final_email:
email_content = render_account_merged(
settings,
language_code=merge_notice.get("language") or settings.DEFAULT_LANGUAGE,
primary_user_id=merge_notice.get("primary_user_id"),
removed_user_id=merge_notice.get("removed_user_id"),
final_end_date_text=str(
merge_notice.get("final_end_date_text")
or merge_notice.get("final_end_date")
or ""
),
)
try:
await email_service.send_rendered_email(
email=final_email,
content=email_content,
)
except Exception as exc:
logger.warning(
"Failed to send account merge email to %s: %s",
final_email,
exc,
)
except UserMergeConflictError as exc:
await session.rollback()
return _json_error(409, "account_merge_conflict", str(exc))
except Exception as exc:
await session.rollback()
logger.exception("Telegram account link failed")
return _json_error(500, "link_failed", "Link failed")
if should_notify_telegram_linked and final_telegram_id:
try:
from bot.services.notification_service import NotificationService
bot: Bot = request.app["bot"]
notification_service = NotificationService(
bot,
settings,
request.app.get("i18n"),
)
await notification_service.notify_account_telegram_linked(
user_id=int(final_user_id),
email=final_email,
telegram_id=int(final_telegram_id),
username=final_username,
first_name=final_first_name,
)
except Exception:
logger.exception("Failed to send account Telegram linked notification")
token = create_webapp_session_token(settings, int(final_user_id))
response_payload: Dict[str, Any] = {
"ok": True,
"user_id": int(final_user_id),
"telegram_id": final_telegram_id,
}
if merge_notice:
response_payload["account_merge"] = merge_notice
return _build_webapp_auth_response(settings, response_payload, token=token)
async def me_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
data = await _build_user_payload(request, user_id)
return web.json_response({"ok": True, **data})
async def account_avatar_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
avatar = await _ensure_cached_telegram_avatar(request, session, db_user)
await session.commit()
if not avatar:
raise web.HTTPNotFound(text="avatar_not_cached")
etag = _telegram_avatar_etag(avatar)
if etag and request.headers.get("If-None-Match") == etag:
return web.Response(status=304, headers={"ETag": etag})
response = web.Response(
body=bytes(avatar.image_bytes),
content_type=avatar.content_type or "image/jpeg",
)
response.headers["Cache-Control"] = "private, max-age=3600"
if etag:
response.headers["ETag"] = etag
return response
async def account_language_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
payload = await _read_json(request)
language_payload, validation_error = _validate_model_payload(WebAppLanguagePayload, payload)
if validation_error:
return validation_error
language = _normalize_language(str(language_payload.language or ""))
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
if _normalize_language(db_user.language_code or "") != language:
db_user.language_code = language
await session.flush()
await session.commit()
return web.json_response({"ok": True, "language": language})
async def apply_promo_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
payload = await _read_json(request)
code = str(payload.get("code") or "").strip()
if not code:
return _json_error(400, "empty_code", "Promo code is empty")
settings: Settings = request.app["settings"]
promo_code_service: PromoCodeService = request.app.get("promo_code_service")
if not promo_code_service:
return _json_error(503, "service_unavailable", "Promo service unavailable")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
try:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
success, result = await promo_code_service.apply_promo_code(
session,
user_id,
code,
lang,
)
if not success:
await session.commit()
return _json_error(400, "promo_apply_failed", str(result))
await session.commit()
end_date = result if isinstance(result, datetime) else None
return web.json_response(
{
"ok": True,
"end_date": end_date.isoformat() if end_date else None,
"end_date_text": end_date.strftime("%d.%m.%Y %H:%M") if end_date else None,
}
)
except Exception as exc:
await session.rollback()
logger.exception("WebApp promo apply failed")
return _json_error(500, "promo_apply_failed", "Promo apply failed")
async def create_payment_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="payments_create",
)
if rate_limit_response:
return rate_limit_response
payload = await _read_json(request)
payment_payload, validation_error = _validate_model_payload(WebAppPaymentCreatePayload, payload)
if validation_error:
return validation_error
method = str(payment_payload.method or "").strip().lower()
settings: Settings = request.app["settings"]
cached = _get_cached_webapp_settings(request)
tariffs_config = settings.tariffs_config
traffic_mode = bool(settings.traffic_sale_mode)
sale_mode = "subscription"
traffic_gb_for_payment: Optional[float] = None
requested_sale_mode = _sale_mode_base(str(payment_payload.sale_mode or ""))
if tariffs_config and requested_sale_mode in {"hwid_device", "hwid_devices"}:
tariff_key = str(payment_payload.tariff_key or "").strip()
if not tariff_key:
return _json_error(400, "invalid_plan", "Tariff is not selected")
try:
tariff = tariffs_config.require(tariff_key)
except Exception:
return _json_error(400, "invalid_plan", "Tariff is not available")
try:
device_count = int(float(
payment_payload.device_count
if payment_payload.device_count is not None
else payment_payload.months
))
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid device package")
packages = tariff.hwid_device_packages
rub_packages = {int(package.count): float(package.price) for package in (packages.rub if packages else [])}
stars_packages = {int(package.count): int(float(package.price)) for package in (packages.stars if packages else [])}
price = rub_packages.get(device_count)
stars_price = stars_packages.get(device_count)
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Device package is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = device_count
sale_mode = f"hwid_devices@{tariff.key}"
elif tariffs_config and requested_sale_mode in {"topup", "premium_topup"}:
tariff_key = str(payment_payload.tariff_key or "").strip()
if not tariff_key:
return _json_error(400, "invalid_plan", "Tariff is not selected")
try:
tariff = tariffs_config.require(tariff_key)
except Exception:
return _json_error(400, "invalid_plan", "Tariff is not available")
try:
traffic_gb = float(
payment_payload.traffic_gb
if payment_payload.traffic_gb is not None
else payment_payload.months
)
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid traffic package")
packages = (
tariff.premium_topup_packages
if requested_sale_mode == "premium_topup"
else tariffs_config.topup_packages_for(tariff)
)
rub_packages = {float(package.gb): float(package.price) for package in (packages.rub if packages else [])}
stars_packages = {float(package.gb): int(float(package.price)) for package in (packages.stars if packages else [])}
package_key = _resolve_numeric_option_key(rub_packages, traffic_gb)
stars_package_key = _resolve_numeric_option_key(stars_packages, traffic_gb)
price = rub_packages.get(package_key) if package_key is not None else None
stars_price = stars_packages.get(stars_package_key) if stars_package_key is not None else None
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Traffic package is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = int(traffic_gb) if float(traffic_gb).is_integer() else traffic_gb
traffic_gb_for_payment = float(payment_units)
sale_mode = f"{requested_sale_mode}@{tariff.key}"
elif tariffs_config:
tariff_key = str(payment_payload.tariff_key or "").strip()
if not tariff_key:
return _json_error(400, "invalid_plan", "Tariff is not selected")
try:
tariff = tariffs_config.require(tariff_key)
except Exception:
return _json_error(400, "invalid_plan", "Tariff is not available")
if tariff.billing_model == "traffic":
try:
traffic_gb = float(
payment_payload.traffic_gb
if payment_payload.traffic_gb is not None
else payment_payload.months
)
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid traffic package")
if traffic_gb <= 0:
return _json_error(400, "invalid_plan", "Invalid traffic package")
rub_packages = {
float(package.gb): float(package.price)
for package in (tariff.traffic_packages.rub if tariff.traffic_packages else [])
}
stars_packages = {
float(package.gb): int(float(package.price))
for package in (tariff.traffic_packages.stars if tariff.traffic_packages else [])
}
package_key = _resolve_numeric_option_key(rub_packages, traffic_gb)
stars_package_key = _resolve_numeric_option_key(stars_packages, traffic_gb)
price = rub_packages.get(package_key) if package_key is not None else None
stars_price = (
stars_packages.get(stars_package_key)
if stars_package_key is not None
else None
)
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Traffic package is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = int(traffic_gb) if float(traffic_gb).is_integer() else traffic_gb
traffic_gb_for_payment = float(payment_units)
sale_mode = f"traffic_package@{tariff.key}"
else:
try:
months = int(float(payment_payload.months))
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid subscription period")
if months not in tariff.enabled_periods:
return _json_error(400, "invalid_plan", "Subscription period is not available")
price = tariff.period_price(months, "rub")
stars_price_raw = tariff.period_price(months, "stars")
stars_price = int(stars_price_raw) if stars_price_raw and stars_price_raw > 0 else None
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Subscription period is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = months
sale_mode = f"subscription@{tariff.key}"
elif traffic_mode:
try:
traffic_gb = float(
payment_payload.traffic_gb
if payment_payload.traffic_gb is not None
else payment_payload.months
)
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid traffic package")
if traffic_gb <= 0:
return _json_error(400, "invalid_plan", "Invalid traffic package")
package_key = _resolve_numeric_option_key(cached["traffic_packages"], traffic_gb)
stars_package_key = _resolve_numeric_option_key(cached["stars_traffic_packages"], traffic_gb)
price = cached["traffic_packages"].get(package_key) if package_key is not None else None
stars_price = (
cached["stars_traffic_packages"].get(stars_package_key)
if stars_package_key is not None
else None
)
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Traffic package is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = int(traffic_gb) if float(traffic_gb).is_integer() else traffic_gb
traffic_gb_for_payment = float(payment_units)
sale_mode = "traffic"
else:
try:
months = int(float(payment_payload.months))
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid subscription period")
price = cached["subscription_options"].get(months)
stars_price = cached["stars_subscription_options"].get(months)
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Subscription period is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
payment_units = months
sale_mode = "subscription"
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
lang = db_user.language_code or settings.DEFAULT_LANGUAGE
return await _create_subscription_payment(
request=request,
session=session,
user_id=user_id,
method=method,
months=payment_units,
price=float(price or 0),
stars_price=stars_price,
lang=lang,
sale_mode=sale_mode,
traffic_gb=traffic_gb_for_payment,
)
async def activate_trial_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="trial_activate",
)
if rate_limit_response:
return rate_limit_response
settings: Settings = request.app["settings"]
if not settings.TRIAL_ENABLED or settings.TRIAL_DURATION_DAYS <= 0:
return _json_error(400, "trial_unavailable", "Trial is not available")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
activation_result = await subscription_service.activate_trial_subscription(session, user_id)
if not activation_result or not activation_result.get("activated"):
await session.rollback()
message_key = (
activation_result.get("message_key", "trial_activation_failed")
if activation_result
else "trial_activation_failed"
)
status = 400 if message_key != "trial_activation_failed_panel_update" else 502
return _json_error(status, message_key, message_key)
end_date = activation_result.get("end_date")
config_link, connect_url = await prepare_config_links(
settings,
activation_result.get("subscription_url"),
)
i18n_instance = request.app.get("i18n")
if settings.LOG_TRIAL_ACTIVATIONS and i18n_instance:
try:
notification_service = NotificationService(request.app["bot"], settings, i18n_instance)
await notification_service.notify_trial_activation(user_id, end_date)
except Exception:
logger.exception("Failed to send WebApp trial activation notification")
try:
from db.dal import ad_dal as _ad_dal
await _ad_dal.mark_trial_activated(session, user_id)
await session.commit()
except Exception:
await session.rollback()
logger.exception("Failed to mark WebApp trial activation for ad attribution")
return web.json_response(
{
"ok": True,
"activated": True,
"days": activation_result.get("days", settings.TRIAL_DURATION_DAYS),
"end_date": end_date.isoformat() if isinstance(end_date, datetime) else None,
"end_date_text": _format_webapp_datetime(end_date) if isinstance(end_date, datetime) else None,
"traffic_gb": activation_result.get("traffic_gb", settings.TRIAL_TRAFFIC_LIMIT_GB),
"config_link": config_link,
"connect_url": connect_url or config_link,
}
)
async def tariff_topup_options_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
config = settings.tariffs_config
if not config:
return _json_error(404, "tariffs_unavailable", "Tariffs are not configured")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
sub = await subscription_dal.get_active_subscription_by_user_id(session, user_id, db_user.panel_user_uuid)
if not sub or not sub.tariff_key:
return _json_error(400, "subscription_required", "Active tariff subscription is required")
lang = db_user.language_code or settings.DEFAULT_LANGUAGE
tariff = config.require(sub.tariff_key)
plans = _serialize_topup_packages(settings, tariff, config.topup_packages_for(tariff), lang)
premium_plans = _serialize_topup_packages(
settings,
tariff,
tariff.premium_topup_packages,
lang,
sale_mode="premium_topup",
title_prefix="Premium ",
) if tariff.premium_squad_uuids else []
premium_limit_bytes = (
int(sub.premium_baseline_bytes or 0)
+ int(sub.premium_topup_balance_bytes or 0)
+ int(getattr(sub, "premium_topup_used_bytes", 0) or 0)
)
premium_access = await request.app["subscription_service"].premium_access_for_tariff(tariff)
return web.json_response(
{
"ok": True,
"tariff_key": tariff.key,
"tariff_name": tariff.name(lang),
"traffic_percent": _traffic_percent(sub.traffic_used_bytes, sub.traffic_limit_bytes),
"premium_traffic_percent": _traffic_percent(
sub.premium_used_bytes,
premium_limit_bytes,
),
"premium_limit_bytes": premium_limit_bytes,
"premium_used_bytes": int(sub.premium_used_bytes or 0),
"premium_baseline_bytes": int(sub.premium_baseline_bytes or 0),
"premium_topup_balance_bytes": int(sub.premium_topup_balance_bytes or 0),
"premium_topup_used_bytes": int(getattr(sub, "premium_topup_used_bytes", 0) or 0),
"premium_is_limited": bool(sub.premium_is_limited),
"premium_squad_labels": premium_access.get("squad_labels") or [],
"premium_node_labels": premium_access.get("node_labels") or [],
"warning_levels": settings.tariff_traffic_warning_levels,
"plans": plans + premium_plans,
}
)
async def tariff_change_options_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
config = settings.tariffs_config
if not config:
return _json_error(404, "tariffs_unavailable", "Tariffs are not configured")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
sub = await subscription_dal.get_active_subscription_by_user_id(session, user_id, db_user.panel_user_uuid)
if not sub or not sub.tariff_key:
return _json_error(400, "subscription_required", "Active tariff subscription is required")
lang = db_user.language_code or settings.DEFAULT_LANGUAGE
current = config.require(sub.tariff_key)
targets = []
for tariff in config.enabled_tariffs:
if tariff.key == current.key:
continue
options = subscription_service.calculate_tariff_switch_options(sub, tariff)
targets.append(_serialize_tariff_change_target(settings, config, tariff, options, lang))
return web.json_response(
{
"ok": True,
"current": {
"tariff_key": current.key,
"title": current.name(lang),
"description": current.description(lang),
"billing_model": current.billing_model,
},
"targets": targets,
}
)
async def tariff_change_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
payload = await _read_json(request)
change_payload, validation_error = _validate_model_payload(WebAppTariffChangePayload, payload)
if validation_error:
return validation_error
mode = str(change_payload.mode or "").strip()
if mode not in {"recalc_days", "convert_days_to_gb"}:
return _json_error(400, "invalid_change_mode", "This tariff change requires payment")
settings: Settings = request.app["settings"]
if not settings.tariffs_config:
return _json_error(404, "tariffs_unavailable", "Tariffs are not configured")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
result = await subscription_service.switch_tariff_without_payment(
session,
user_id,
str(change_payload.tariff_key),
mode,
)
if not result:
await session.rollback()
return _json_error(400, "change_failed", "Tariff change failed")
await session.commit()
return web.json_response({"ok": True, **result})
async def tariff_change_payment_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
payload = await _read_json(request)
payment_payload, validation_error = _validate_model_payload(WebAppPaymentCreatePayload, payload)
if validation_error:
return validation_error
method = str(payment_payload.method or "").strip().lower()
tariff_key = str(payment_payload.tariff_key or "").strip()
settings: Settings = request.app["settings"]
config = settings.tariffs_config
if not config:
return _json_error(404, "tariffs_unavailable", "Tariffs are not configured")
if not tariff_key:
return _json_error(400, "invalid_plan", "Tariff is not selected")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
sub = await subscription_dal.get_active_subscription_by_user_id(session, user_id, db_user.panel_user_uuid)
if not sub:
return _json_error(400, "subscription_required", "Active tariff subscription is required")
target = config.require(tariff_key)
options = subscription_service.calculate_tariff_switch_options(sub, target)
price = float(options.get("paid_diff_rub") or 0)
if price <= 0:
return _json_error(400, "payment_not_required", "Payment is not required for this tariff change")
return await _create_subscription_payment(
request=request,
session=session,
user_id=user_id,
method=method,
months=1,
price=price,
stars_price=None,
lang=db_user.language_code or settings.DEFAULT_LANGUAGE,
sale_mode=f"tariff_upgrade@{target.key}",
)
async def devices_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
if not settings.MY_DEVICES_SECTION_ENABLED:
return _json_error(404, "devices_disabled", "Devices section is disabled")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
active = await subscription_service.get_active_subscription_details(session, user_id)
panel_user_uuid = active.get("user_id") if active else None
if not panel_user_uuid:
return _json_error(400, "subscription_not_active", "Subscription is not active")
panel_service = getattr(subscription_service, "panel_service", None)
if not panel_service:
return _json_error(503, "panel_unavailable", "Panel service unavailable")
try:
devices_response = await panel_service.get_user_devices(panel_user_uuid)
except Exception:
logger.exception("Failed to load WebApp devices for user %s", user_id)
return _json_error(502, "devices_load_failed", "Failed to load devices")
devices = _normalize_devices_response(devices_response)
max_devices = _coerce_int_or_none(active.get("max_devices")) if active else None
return web.json_response(
{
"ok": True,
"enabled": True,
"current_devices": len(devices),
"max_devices": max_devices,
"max_devices_label": _format_devices_limit(max_devices),
"devices": [_serialize_device(device, index) for index, device in enumerate(devices, start=1)],
}
)
async def disconnect_device_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="devices_disconnect",
)
if rate_limit_response:
return rate_limit_response
settings: Settings = request.app["settings"]
if not settings.MY_DEVICES_SECTION_ENABLED:
return _json_error(404, "devices_disabled", "Devices section is disabled")
payload = await _read_json(request)
disconnect_payload, validation_error = _validate_model_payload(WebAppDeviceDisconnectPayload, payload)
if validation_error:
return validation_error
token = str(disconnect_payload.token or "").strip()
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
active = await subscription_service.get_active_subscription_details(session, user_id)
panel_user_uuid = active.get("user_id") if active else None
if not panel_user_uuid:
return _json_error(400, "subscription_not_active", "Subscription is not active")
panel_service = getattr(subscription_service, "panel_service", None)
if not panel_service:
return _json_error(503, "panel_unavailable", "Panel service unavailable")
try:
devices_response = await panel_service.get_user_devices(panel_user_uuid)
except Exception:
logger.exception("Failed to load WebApp devices before disconnect for user %s", user_id)
return _json_error(502, "devices_load_failed", "Failed to load devices")
target_hwid = None
for device in _normalize_devices_response(devices_response):
hwid = str(device.get("hwid") or "").strip()
if hwid and hmac.compare_digest(_device_hwid_token(hwid), token):
target_hwid = hwid
break
if not target_hwid:
return _json_error(404, "device_not_found", "Device not found")
success = await panel_service.disconnect_device(panel_user_uuid, target_hwid)
if not success:
return _json_error(502, "device_disconnect_failed", "Failed to disconnect device")
await session.commit()
return web.json_response({"ok": True})
async def device_topup_options_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
config = settings.tariffs_config
if not settings.MY_DEVICES_SECTION_ENABLED:
return _json_error(404, "devices_disabled", "Devices section is disabled")
if not config:
return _json_error(404, "tariffs_unavailable", "Tariffs are not configured")
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
sub = await subscription_dal.get_active_subscription_by_user_id(session, user_id, db_user.panel_user_uuid)
if not sub or not sub.tariff_key:
return _json_error(400, "subscription_required", "Active tariff subscription is required")
tariff = config.require(sub.tariff_key)
active = await subscription_service.get_active_subscription_details(session, user_id)
plans = _serialize_hwid_device_packages(
settings,
tariff,
tariff.hwid_device_packages,
db_user.language_code or settings.DEFAULT_LANGUAGE,
)
return web.json_response(
{
"ok": True,
"tariff_key": tariff.key,
"tariff_name": tariff.name(db_user.language_code or settings.DEFAULT_LANGUAGE),
"current_limit": _coerce_int_or_none(active.get("max_devices")) if active else None,
"extra_hwid_devices": int(sub.extra_hwid_devices or 0),
"plans": plans,
}
)
async def payment_status_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
try:
payment_id = int(request.match_info["payment_id"])
except (TypeError, ValueError):
return _json_error(400, "invalid_payment", "Invalid payment id")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
payment = await payment_dal.get_payment_by_db_id(session, payment_id)
if not payment or payment.user_id != user_id:
return _json_error(404, "not_found", "Payment not found")
return web.json_response(
{
"ok": True,
"payment_id": payment.payment_id,
"status": payment.status,
"paid": payment.status == "succeeded",
}
)
async def _read_json(request: web.Request) -> Dict[str, Any]:
try:
data = await request.json()
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _json_error(status: int, code: str, message: str) -> web.Response:
return web.json_response(
{"ok": False, "error": code, "message": message},
status=status,
)
def _validation_error_response(exc: ValidationError) -> web.Response:
for error in exc.errors():
loc = error.get("loc") or ()
field = str(loc[0]) if loc else ""
error_type = str(error.get("type") or "")
message = str(error.get("msg") or "")
message_lower = message.lower()
if field == "email":
if ("too_long" in message_lower or "too long" in message_lower or error_type == "string_too_long"):
return _json_error(400, "email_too_long", "Email is too long")
return _json_error(400, "invalid_email", "Invalid email")
if field in {"description", "comment", "note"} and error_type == "string_too_long":
return _json_error(400, f"{field}_too_long", f"{field.capitalize()} is too long")
if error_type == "string_too_long":
return _json_error(400, "text_too_long", "Text is too long")
return _json_error(400, "invalid_request", "Invalid request")
def _validate_model_payload(
model_cls: type[BaseModel],
payload: Dict[str, Any],
) -> tuple[Optional[BaseModel], Optional[web.Response]]:
try:
return model_cls.model_validate(payload), None
except ValidationError as exc:
return None, _validation_error_response(exc)
def _set_webapp_auth_cookies(
response: web.StreamResponse,
settings: Settings,
session_token: str,
csrf_token: str,
) -> None:
max_age = max(60, int(settings.WEBAPP_SESSION_TTL_SECONDS))
response.set_cookie(
WEBAPP_SESSION_COOKIE_NAME,
session_token,
httponly=True,
secure=True,
samesite="None",
path="/",
max_age=max_age,
)
response.set_cookie(
WEBAPP_CSRF_COOKIE_NAME,
csrf_token,
httponly=False,
secure=True,
samesite="None",
path="/",
max_age=max_age,
)
def _clear_webapp_auth_cookies(response: web.StreamResponse) -> None:
response.set_cookie(
WEBAPP_SESSION_COOKIE_NAME,
"",
httponly=True,
secure=True,
samesite="None",
path="/",
max_age=0,
)
response.set_cookie(
WEBAPP_CSRF_COOKIE_NAME,
"",
httponly=False,
secure=True,
samesite="None",
path="/",
max_age=0,
)
def _build_webapp_auth_response(
settings: Settings,
payload: Dict[str, Any],
*,
token: str,
csrf_token: Optional[str] = None,
) -> web.Response:
response_payload = dict(payload)
response_payload["ok"] = True
response_payload["token"] = token
csrf_value = csrf_token or secrets.token_hex(32)
response_payload["csrf_token"] = csrf_value
response = web.json_response(response_payload)
_set_webapp_auth_cookies(response, settings, token, csrf_value)
return response
def _extract_authenticated_user_id(request: web.Request) -> Optional[int]:
settings: Settings = request.app["settings"]
header = request.headers.get("Authorization", "")
prefix = "Bearer "
if header.startswith(prefix):
user_id = verify_webapp_session_token(settings, header[len(prefix):].strip())
if user_id:
return user_id
cookie_token = request.cookies.get(WEBAPP_SESSION_COOKIE_NAME, "")
if cookie_token:
return verify_webapp_session_token(settings, cookie_token)
return None
def _require_user_id(request: web.Request) -> int:
user_id = _extract_authenticated_user_id(request)
if not user_id:
raise web.HTTPUnauthorized(
text=json.dumps({"ok": False, "error": "unauthorized"}),
content_type="application/json",
)
return user_id
async def _request_email_code(
request: web.Request,
*,
email: str,
purpose: str,
language_code: str,
target_user_id: Optional[int],
) -> web.Response:
email_service: EmailAuthService = request.app["email_auth_service"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
try:
result = await email_service.request_code(
session,
email=email,
purpose=purpose,
language_code=language_code,
target_user_id=target_user_id,
)
if not result.ok:
await session.rollback()
status = 429 if result.error == "rate_limited" else 400
if result.error == "email_auth_not_configured":
status = 503
return web.json_response(
{
"ok": False,
"error": result.error,
"retry_after": result.retry_after,
},
status=status,
)
await session.commit()
return web.json_response({"ok": True})
except Exception as exc:
await session.rollback()
logger.exception("Failed to send email verification code")
return _json_error(502, "email_send_failed", "Failed to send email")
def _telegram_id_for_user(user: User) -> Optional[int]:
if user.telegram_id:
return int(user.telegram_id)
if user.user_id and int(user.user_id) > 0:
return int(user.user_id)
return None
def _panel_description_for_user(user: User) -> str:
lines = [
user.email or "",
user.username or "",
user.first_name or "",
user.last_name or "",
]
return "\n".join(line for line in lines if line).strip()
async def _sync_panel_identity_for_user(
request: web.Request,
user: User,
*,
expire_at: Optional[datetime] = None,
) -> bool:
if not user.panel_user_uuid:
return False
subscription_service: SubscriptionService = request.app.get("subscription_service")
if not subscription_service or not subscription_service.panel_service:
return False
payload: Dict[str, Any] = {
"description": _panel_description_for_user(user),
}
telegram_id = _telegram_id_for_user(user)
if telegram_id:
payload["telegramId"] = telegram_id
if user.email:
payload["email"] = user.email
if expire_at is not None:
payload["expireAt"] = expire_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
try:
await subscription_service.panel_service.update_user_details_on_panel(
user.panel_user_uuid,
payload,
log_response=False,
)
return True
except Exception as exc:
logger.warning(
"Failed to sync linked identities to panel for user %s: %s",
user.user_id,
exc,
)
return False
def _format_webapp_datetime(value: Optional[datetime]) -> Optional[str]:
if not value:
return None
normalized = value if value.tzinfo else value.replace(tzinfo=timezone.utc)
return normalized.strftime("%d.%m.%Y %H:%M")
async def _build_account_merge_notice(
session: AsyncSession,
*,
merged_user: User,
source_user_id: int,
source_panel_uuid: Optional[str],
settings: Settings,
) -> Dict[str, Any]:
merged_subscription = None
if merged_user.panel_user_uuid:
merged_subscription = await subscription_dal.get_active_subscription_by_user_id(
session,
merged_user.user_id,
merged_user.panel_user_uuid,
)
if not merged_subscription:
merged_subscription = await subscription_dal.get_active_subscription_by_user_id(
session,
merged_user.user_id,
)
final_end_date = merged_subscription.end_date if merged_subscription else None
if final_end_date and final_end_date.tzinfo is None:
final_end_date = final_end_date.replace(tzinfo=timezone.utc)
return {
"merged": True,
"language": _normalize_language(merged_user.language_code or settings.DEFAULT_LANGUAGE),
"primary_user_id": int(merged_user.user_id),
"removed_user_id": int(source_user_id),
"primary_panel_user_uuid": merged_user.panel_user_uuid,
"removed_panel_user_uuid": source_panel_uuid,
"final_end_date": final_end_date.isoformat() if final_end_date else None,
"final_end_date_text": _format_webapp_datetime(final_end_date),
}
def _telegram_photo_url_value(telegram_user: Dict[str, Any]) -> Optional[str]:
raw_value = telegram_user.get("photo_url")
if not raw_value:
return None
value = str(raw_value).strip()
return value or None
def _telegram_avatar_is_stale(avatar: Optional[UserTelegramAvatar]) -> bool:
if not avatar or not avatar.updated_at:
return True
updated_at = avatar.updated_at
if updated_at.tzinfo is None:
updated_at = updated_at.replace(tzinfo=timezone.utc)
return (datetime.now(timezone.utc) - updated_at).total_seconds() >= WEBAPP_TELEGRAM_AVATAR_REFRESH_SECONDS
def _telegram_avatar_etag(avatar: UserTelegramAvatar) -> str:
digest = hashlib.sha256(bytes(avatar.image_bytes)).hexdigest()[:16]
return f'"tg-avatar-{int(avatar.user_id)}-{digest}"'
def _telegram_avatar_url(avatar: Optional[UserTelegramAvatar]) -> str:
if not avatar:
return ""
updated_at = avatar.updated_at
if updated_at and updated_at.tzinfo is None:
updated_at = updated_at.replace(tzinfo=timezone.utc)
version = int(updated_at.timestamp()) if updated_at else hashlib.sha256(bytes(avatar.image_bytes)).hexdigest()[:8]
return f"/api/account/avatar?v={version}"
def _select_compact_telegram_photo_size(sizes: List[Any]) -> Optional[Any]:
if not sizes:
return None
suitable = [size for size in sizes if int(getattr(size, "width", 0) or 0) >= 160]
candidates = suitable or sizes
return min(
candidates,
key=lambda size: (
int(getattr(size, "file_size", 0) or 0) or int(getattr(size, "width", 0) or 0) * int(getattr(size, "height", 0) or 0),
int(getattr(size, "width", 0) or 0),
),
)
def _telegram_file_content_type(file_path: Optional[str]) -> str:
path = str(file_path or "").lower()
if path.endswith(".png"):
return "image/png"
if path.endswith(".webp"):
return "image/webp"
return "image/jpeg"
async def _fetch_compact_telegram_avatar(bot: Bot, telegram_id: int) -> Optional[Tuple[bytes, str, Optional[str]]]:
photos = await bot.get_user_profile_photos(user_id=telegram_id, limit=1)
if not photos or not photos.photos:
return None
photo_size = _select_compact_telegram_photo_size(list(photos.photos[0] or []))
if not photo_size:
return None
file_info = await bot.get_file(photo_size.file_id)
destination = io.BytesIO()
await bot.download_file(file_info.file_path, destination=destination)
body = destination.getvalue()
if not body or len(body) > WEBAPP_TELEGRAM_AVATAR_MAX_BYTES:
return None
return body, _telegram_file_content_type(file_info.file_path), getattr(photo_size, "file_unique_id", None)
async def _ensure_cached_telegram_avatar(
request: web.Request,
session: AsyncSession,
user: User,
) -> Optional[UserTelegramAvatar]:
avatar = await user_dal.get_user_telegram_avatar(session, int(user.user_id))
telegram_id = _telegram_id_for_user(user)
if not telegram_id:
return avatar
if avatar and not _telegram_avatar_is_stale(avatar):
return avatar
bot: Bot = request.app["bot"]
try:
fetched = await asyncio.wait_for(
_fetch_compact_telegram_avatar(bot, int(telegram_id)),
timeout=WEBAPP_TELEGRAM_AVATAR_FETCH_TIMEOUT_SECONDS,
)
except Exception as exc:
logger.info("Failed to refresh Telegram avatar for user %s: %s", user.user_id, exc)
return avatar
if not fetched:
return avatar
body, content_type, file_unique_id = fetched
return await user_dal.upsert_user_telegram_avatar(
session,
user_id=int(user.user_id),
file_unique_id=file_unique_id,
content_type=content_type,
image_bytes=body,
)
def _apply_telegram_profile_to_user(
user: User,
telegram_user: Dict[str, Any],
settings: Settings,
) -> None:
language_code = telegram_user.get("language_code") or user.language_code or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = user.language_code or settings.DEFAULT_LANGUAGE
user.telegram_id = int(telegram_user["id"])
user.username = sanitize_username(telegram_user.get("username"))
user.first_name = sanitize_display_name(telegram_user.get("first_name"))
user.last_name = sanitize_display_name(telegram_user.get("last_name"))
user.language_code = language_code
telegram_photo_url = _telegram_photo_url_value(telegram_user)
if telegram_photo_url:
user.telegram_photo_url = telegram_photo_url
async def _link_telegram_to_user(
request: web.Request,
session: AsyncSession,
*,
current_user_id: int,
telegram_user: Dict[str, Any],
settings: Settings,
) -> User:
telegram_id = int(telegram_user["id"])
current_user = await user_dal.get_user_by_id(session, current_user_id)
if not current_user:
raise ValueError("Current user not found.")
existing_telegram_user = await user_dal.get_user_by_telegram_id(session, telegram_id)
if not existing_telegram_user:
existing_telegram_user = await user_dal.get_user_by_id(session, telegram_id)
if existing_telegram_user and existing_telegram_user.user_id != current_user.user_id:
if (
current_user.email
and existing_telegram_user.email
and current_user.email != existing_telegram_user.email
):
raise UserMergeConflictError(
"Telegram account is already linked to a different email."
)
merged_user = await user_dal.merge_users(
session,
source_user_id=current_user.user_id,
target_user_id=existing_telegram_user.user_id,
)
_apply_telegram_profile_to_user(merged_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, merged_user)
return merged_user
if not existing_telegram_user and int(current_user.user_id) < 0:
language_code = telegram_user.get("language_code") or current_user.language_code or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = current_user.language_code or settings.DEFAULT_LANGUAGE
target_user, _ = await user_dal.create_user(
session,
{
"user_id": telegram_id,
"telegram_id": telegram_id,
"username": sanitize_username(telegram_user.get("username")),
"first_name": sanitize_display_name(telegram_user.get("first_name")),
"last_name": sanitize_display_name(telegram_user.get("last_name")),
"language_code": language_code,
"registration_date": current_user.registration_date or datetime.now(timezone.utc),
},
)
target_user.referral_code = None
await session.flush()
merged_user = await user_dal.merge_users(
session,
source_user_id=current_user.user_id,
target_user_id=target_user.user_id,
)
_apply_telegram_profile_to_user(merged_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, merged_user)
return merged_user
if current_user.telegram_id and int(current_user.telegram_id) != telegram_id:
raise UserMergeConflictError("Current account is already linked to Telegram.")
_apply_telegram_profile_to_user(current_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, current_user)
return current_user
def _normalize_referral_param(raw: Optional[str]) -> Optional[str]:
value = (raw or "").strip()
if not value:
return None
value_lower = value.lower()
if value_lower.startswith("ref_u"):
value = value[5:]
elif value_lower.startswith("ref_"):
value = value[4:]
elif value and value[0].lower() == "u" and len(value) == 10:
value = value[1:]
if not re.fullmatch(r"[A-Za-z0-9]{1,32}", value):
return None
return value.upper()
async def _resolve_referrer_id(
session: AsyncSession,
raw_referral_param: Optional[str],
*,
current_user_id: Optional[int],
) -> Optional[int]:
normalized = _normalize_referral_param(raw_referral_param)
if not normalized:
return None
ref_user = None
if normalized.isdigit():
ref_user = await user_dal.get_user_by_id(session, int(normalized))
if not ref_user:
ref_user = await user_dal.get_user_by_referral_code(session, normalized)
if not ref_user:
return None
if current_user_id is not None and int(ref_user.user_id) == int(current_user_id):
return None
return int(ref_user.user_id)
async def _apply_referral_to_existing_user(
request: web.Request,
session: AsyncSession,
user: User,
raw_referral_param: Optional[str],
) -> bool:
if not raw_referral_param or user.referred_by_id is not None:
return False
referred_by_id = await _resolve_referrer_id(
session,
raw_referral_param,
current_user_id=int(user.user_id),
)
if not referred_by_id:
return False
subscription_service: SubscriptionService = request.app["subscription_service"]
try:
is_active_now = await subscription_service.has_active_subscription(
session,
int(user.user_id),
)
except Exception:
is_active_now = False
if is_active_now:
return False
user.referred_by_id = referred_by_id
await session.flush()
return True
async def _apply_referral_welcome_bonus_if_needed(
request: web.Request,
session: AsyncSession,
user: User,
raw_referral_param: Optional[str],
) -> Optional[datetime]:
if not raw_referral_param or not user.referred_by_id:
return None
settings: Settings = request.app["settings"]
referral_welcome_days = max(
0,
int(getattr(settings, "REFERRAL_WELCOME_BONUS_DAYS", 0) or 0),
)
if referral_welcome_days <= 0:
return None
subscription_service: SubscriptionService = request.app["subscription_service"]
try:
if await subscription_service.has_active_subscription(session, int(user.user_id)):
return None
except Exception:
pass
return await subscription_service.extend_active_subscription_days(
session,
int(user.user_id),
referral_welcome_days,
reason="referral_welcome_bonus",
)
async def _ensure_user_from_telegram(
session: AsyncSession,
telegram_user: Dict[str, Any],
settings: Settings,
*,
referral_param: Optional[str] = None,
) -> User:
user_id = int(telegram_user["id"])
language_code = telegram_user.get("language_code") or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = settings.DEFAULT_LANGUAGE
update_data = {
"telegram_id": user_id,
"username": sanitize_username(telegram_user.get("username")),
"first_name": sanitize_display_name(telegram_user.get("first_name")),
"last_name": sanitize_display_name(telegram_user.get("last_name")),
"language_code": language_code,
}
telegram_photo_url = _telegram_photo_url_value(telegram_user)
if telegram_photo_url:
update_data["telegram_photo_url"] = telegram_photo_url
db_user = await user_dal.get_user_by_telegram_id(session, user_id)
if not db_user:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user:
referred_by_id = await _resolve_referrer_id(
session,
referral_param or telegram_user.get("start_param"),
current_user_id=user_id,
)
db_user, created = await user_dal.create_user(
session,
{
"user_id": user_id,
**update_data,
"referred_by_id": referred_by_id,
"registration_date": datetime.now(timezone.utc),
},
)
setattr(db_user, "_webapp_created", bool(created))
return db_user
changed = {
key: value
for key, value in update_data.items()
if getattr(db_user, key) != value
}
if changed:
db_user = await user_dal.update_user(session, db_user.user_id, changed) or db_user
return db_user
async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, Any]:
settings: Settings = request.app["settings"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
cached = _get_cached_webapp_settings(request)
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
raise web.HTTPForbidden(
text=json.dumps({"ok": False, "error": "access_denied"}),
content_type="application/json",
)
active = await subscription_service.get_active_subscription_details(
session, user_id
)
referral_code = await user_dal.ensure_referral_code(session, db_user)
referral_service: Optional[ReferralService] = request.app.get("referral_service")
bot_username = request.app.get("bot_username") or ""
referral_link = None
if referral_service and bot_username:
referral_link = await referral_service.generate_referral_link(
session,
bot_username,
user_id,
)
webapp_referral_link = _build_webapp_referral_link(
request.app["settings"].SUBSCRIPTION_MINI_APP_URL,
referral_code,
)
referral_stats = (
await referral_service.get_referral_stats(session, user_id)
if referral_service
else {"invited_count": 0, "purchased_count": 0}
)
local_sub = await subscription_dal.get_active_subscription_by_user_id(
session,
user_id,
db_user.panel_user_uuid,
) if db_user.panel_user_uuid else None
trial_available = bool(
settings.TRIAL_ENABLED
and settings.TRIAL_DURATION_DAYS > 0
and not await subscription_service.has_had_any_subscription(session, user_id)
)
avatar = await _ensure_cached_telegram_avatar(request, session, db_user)
try:
await session.commit()
except Exception:
await session.rollback()
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
admin_ids = {int(x) for x in (settings.ADMIN_IDS or [])}
is_admin = bool(db_user.telegram_id and int(db_user.telegram_id) in admin_ids)
return {
"user": {
"id": user_id,
"username": db_user.username,
"email": db_user.email,
"email_verified": bool(db_user.email_verified_at),
"telegram_id": db_user.telegram_id,
"telegram_linked": bool(_telegram_id_for_user(db_user)),
"telegram_photo_url": _telegram_avatar_url(avatar),
"first_name": db_user.first_name,
"language_code": lang,
"is_admin": is_admin,
},
"subscription": _serialize_subscription(settings, active, local_sub, lang),
"referral": {
"code": referral_code,
"bot_link": referral_link,
"webapp_link": webapp_referral_link,
"invited_count": referral_stats.get("invited_count", 0),
"purchased_count": referral_stats.get("purchased_count", 0),
"welcome_bonus_days": max(0, int(getattr(settings, "REFERRAL_WELCOME_BONUS_DAYS", 0) or 0)),
"one_bonus_per_referee": bool(getattr(settings, "REFERRAL_ONE_BONUS_PER_REFEREE", False)),
"bonus_details": _serialize_referral_bonus_details(settings, lang),
},
"plans": _serialize_plans(
settings,
lang,
subscription_options=cached["subscription_options"],
stars_subscription_options=cached["stars_subscription_options"],
traffic_packages=cached["traffic_packages"],
stars_traffic_packages=cached["stars_traffic_packages"],
),
"payment_methods": _serialize_payment_methods(settings, request.app),
"settings": {
"support_url": settings.SUPPORT_LINK,
"traffic_mode": bool(settings.traffic_sale_mode),
"my_devices_enabled": bool(settings.MY_DEVICES_SECTION_ENABLED),
"user_hwid_device_limit": (
int(settings.USER_HWID_DEVICE_LIMIT)
if settings.USER_HWID_DEVICE_LIMIT is not None
else None
),
"trial_enabled": bool(settings.TRIAL_ENABLED),
"trial_available": trial_available,
"trial_duration_days": int(settings.TRIAL_DURATION_DAYS or 0),
"trial_traffic_limit_gb": float(settings.TRIAL_TRAFFIC_LIMIT_GB or 0),
"trial_traffic_strategy": getattr(settings, "TRIAL_TRAFFIC_STRATEGY", "NO_RESET"),
"email_auth_enabled": settings.email_auth_configured,
},
}
def _serialize_referral_bonus_details(settings: Settings, lang: str) -> List[Dict[str, Any]]:
if getattr(settings, "traffic_sale_mode", False):
return []
details: List[Dict[str, Any]] = []
for months, _price in sorted(settings.subscription_options.items()):
inviter_days = settings.referral_bonus_inviter.get(months)
friend_days = settings.referral_bonus_referee.get(months)
if inviter_days is None and friend_days is None:
continue
details.append(
{
"months": int(months),
"title": _format_months_title(int(months), lang),
"inviter_days": int(inviter_days or 0),
"friend_days": int(friend_days or 0),
}
)
return details
def _build_webapp_referral_link(
base_url: Optional[str],
referral_code: Optional[str],
) -> Optional[str]:
if not base_url or not referral_code:
return None
parts = urlsplit(base_url)
query = dict(parse_qsl(parts.query, keep_blank_values=True))
query["ref"] = f"u{referral_code}"
return urlunsplit(
(
parts.scheme,
parts.netloc,
parts.path or "/",
urlencode(query),
parts.fragment,
)
)
def _serialize_subscription(
settings: Settings,
active: Optional[Dict[str, Any]],
local_sub: Optional[Any],
lang: str,
) -> Dict[str, Any]:
if not active:
return {
"active": False,
"status": "INACTIVE",
"remaining_text": _format_remaining(0, lang),
"days_left": 0,
"config_link": None,
"connect_url": None,
}
end_date = active.get("end_date")
if end_date and end_date.tzinfo is None:
end_date = end_date.replace(tzinfo=timezone.utc)
seconds_left = 0
if end_date:
seconds_left = max(
0,
int((end_date - datetime.now(timezone.utc)).total_seconds()),
)
can_topup_traffic = False
if settings.tariffs_config and active.get("tariff_key"):
try:
tariff = settings.tariffs_config.require(str(active.get("tariff_key")))
packages = settings.tariffs_config.topup_packages_for(tariff)
can_topup_traffic = bool(
(packages and packages.has_any())
or (tariff.premium_topup_packages and tariff.premium_topup_packages.has_any())
)
except Exception:
can_topup_traffic = False
return {
"active": seconds_left > 0,
"status": active.get("status_from_panel") or "UNKNOWN",
"end_date": end_date.isoformat() if end_date else None,
"end_date_text": end_date.strftime("%d.%m.%Y %H:%M") if end_date else "N/A",
"days_left": seconds_left // 86400,
"remaining_text": _format_remaining(seconds_left, lang),
"config_link": active.get("config_link"),
"connect_url": active.get("connect_button_url") or active.get("config_link"),
"traffic_limit": _format_bytes(active.get("traffic_limit_bytes")),
"traffic_used": _format_bytes(active.get("traffic_used_bytes")),
"traffic_limit_bytes": _coerce_int_or_none(active.get("traffic_limit_bytes")),
"traffic_used_bytes": _coerce_int_or_none(active.get("traffic_used_bytes")),
"tariff_key": active.get("tariff_key"),
"tariff_name": active.get("tariff_name"),
"tariff_description": active.get("tariff_description"),
"billing_model": active.get("billing_model"),
"traffic_limit_strategy": str(active.get("traffic_limit_strategy") or ""),
"tier_baseline_bytes": _coerce_int_or_none(active.get("tier_baseline_bytes")),
"topup_balance_bytes": _coerce_int_or_none(active.get("topup_balance_bytes")),
"premium_limit": _format_bytes(active.get("premium_limit_bytes")),
"premium_used": _format_bytes(active.get("premium_used_bytes")),
"premium_limit_bytes": _coerce_int_or_none(active.get("premium_limit_bytes")),
"premium_used_bytes": _coerce_int_or_none(active.get("premium_used_bytes")),
"premium_baseline_bytes": _coerce_int_or_none(active.get("premium_baseline_bytes")),
"premium_topup_balance_bytes": _coerce_int_or_none(active.get("premium_topup_balance_bytes")),
"premium_topup_used_bytes": _coerce_int_or_none(active.get("premium_topup_used_bytes")),
"premium_is_limited": bool(active.get("premium_is_limited")),
"premium_squad_labels": list(active.get("premium_squad_labels") or []),
"premium_node_labels": list(active.get("premium_node_labels") or []),
"can_topup_traffic": can_topup_traffic,
"period_start_at": active.get("period_start_at").isoformat() if active.get("period_start_at") else None,
"is_throttled": bool(active.get("is_throttled")),
"max_devices": _coerce_int_or_none(active.get("max_devices")),
"base_hwid_device_limit": _coerce_int_or_none(active.get("base_hwid_device_limit")),
"extra_hwid_devices": _coerce_int_or_none(active.get("extra_hwid_devices")) or 0,
"auto_renew_enabled": bool(getattr(local_sub, "auto_renew_enabled", False)),
"provider": getattr(local_sub, "provider", None),
}
def _serialize_plans(
settings: Settings,
lang: str,
*,
subscription_options: Optional[Dict[int, float]] = None,
stars_subscription_options: Optional[Dict[int, int]] = None,
traffic_packages: Optional[Dict[float, float]] = None,
stars_traffic_packages: Optional[Dict[float, int]] = None,
) -> List[Dict[str, Any]]:
tariffs_config = settings.tariffs_config
if tariffs_config:
plans: List[Dict[str, Any]] = []
for tariff in tariffs_config.enabled_tariffs:
common = {
"tariff_key": tariff.key,
"tariff_name": tariff.name(lang),
"billing_model": tariff.billing_model,
"description": tariff.description(lang),
"squad_uuids": tariff.squad_uuids,
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"hwid_device_limit": tariff.hwid_device_limit,
"hwid_device_packages": _serialize_hwid_device_packages(
settings,
tariff,
tariff.hwid_device_packages,
lang,
),
}
if tariff.billing_model == "period":
for months in sorted(tariff.enabled_periods):
price = tariff.period_price(int(months), "rub")
stars_price = tariff.period_price(int(months), "stars")
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
plan = {
**common,
"id": f"{tariff.key}:period:{int(months)}",
"sale_mode": "subscription",
"months": int(months),
"price": float(price or 0),
"title": tariff.name(lang),
"subtitle": _format_months_title(int(months), lang),
"monthly_gb": tariff.monthly_gb,
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
else:
rub_packages = {
float(package.gb): float(package.price)
for package in (tariff.traffic_packages.rub if tariff.traffic_packages else [])
}
stars_packages = {
float(package.gb): int(float(package.price))
for package in (tariff.traffic_packages.stars if tariff.traffic_packages else [])
}
for traffic_gb in sorted(set(rub_packages) | set(stars_packages)):
price = rub_packages.get(traffic_gb)
stars_price = stars_packages.get(traffic_gb)
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
traffic_value = float(traffic_gb)
plan = {
**common,
"id": f"{tariff.key}:traffic:{_format_number_for_payload(traffic_value)}",
"sale_mode": "traffic_package",
"months": int(traffic_value) if traffic_value.is_integer() else traffic_value,
"traffic_gb": traffic_value,
"price": float(price or 0),
"title": tariff.name(lang),
"subtitle": _format_traffic_title(traffic_value, lang),
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
if getattr(settings, "traffic_sale_mode", False):
active_traffic_packages = traffic_packages or settings.traffic_packages
active_stars_traffic_packages = stars_traffic_packages or settings.stars_traffic_packages
traffic_units = sorted(set(active_traffic_packages) | set(active_stars_traffic_packages))
plans: List[Dict[str, Any]] = []
for traffic_gb in traffic_units:
price = active_traffic_packages.get(traffic_gb)
stars_price = active_stars_traffic_packages.get(traffic_gb)
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
traffic_value = float(traffic_gb)
plan = {
"months": int(traffic_value) if traffic_value.is_integer() else traffic_value,
"traffic_gb": traffic_value,
"price": float(price or 0),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"title": _format_traffic_title(traffic_value, lang),
"sale_mode": "traffic",
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
active_subscription_options = subscription_options or settings.subscription_options
active_stars_subscription_options = stars_subscription_options or settings.stars_subscription_options
plans: List[Dict[str, Any]] = []
for months in sorted(set(active_subscription_options) | set(active_stars_subscription_options)):
price = active_subscription_options.get(months)
stars_price = active_stars_subscription_options.get(months)
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
plan = {
"months": int(months),
"price": float(price or 0),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"title": _format_months_title(int(months), lang),
"sale_mode": "subscription",
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
def _traffic_percent(used: Optional[int], limit: Optional[int]) -> int:
used_val = int(used or 0)
limit_val = int(limit or 0)
if limit_val <= 0:
return 0
return max(0, min(100, round((used_val / limit_val) * 100)))
def _serialize_topup_packages(
settings: Settings,
tariff: Any,
packages: Optional[Any],
lang: str,
*,
sale_mode: str = "topup",
title_prefix: str = "",
) -> List[Dict[str, Any]]:
rub_packages = {float(package.gb): float(package.price) for package in (packages.rub if packages else [])}
stars_packages = {float(package.gb): int(float(package.price)) for package in (packages.stars if packages else [])}
plans: List[Dict[str, Any]] = []
for traffic_gb in sorted(set(rub_packages) | set(stars_packages)):
price = rub_packages.get(traffic_gb)
stars_price = stars_packages.get(traffic_gb)
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
traffic_value = float(traffic_gb)
plan: Dict[str, Any] = {
"id": f"{tariff.key}:{sale_mode}:{_format_number_for_payload(traffic_value)}",
"tariff_key": tariff.key,
"tariff_name": tariff.name(lang),
"billing_model": tariff.billing_model,
"sale_mode": sale_mode,
"months": int(traffic_value) if traffic_value.is_integer() else traffic_value,
"traffic_gb": traffic_value,
"price": float(price or 0),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"title": f"{title_prefix}{_format_traffic_title(traffic_value, lang)}",
"subtitle": ("Premium-серверы" if lang == "ru" else "Premium servers") if sale_mode == "premium_topup" else tariff.name(lang),
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
def _serialize_hwid_device_packages(
settings: Settings,
tariff: Any,
packages: Optional[Any],
lang: str,
) -> List[Dict[str, Any]]:
rub_packages = {int(package.count): float(package.price) for package in (packages.rub if packages else [])}
stars_packages = {int(package.count): int(float(package.price)) for package in (packages.stars if packages else [])}
plans: List[Dict[str, Any]] = []
for count in sorted(set(rub_packages) | set(stars_packages)):
price = rub_packages.get(count)
stars_price = stars_packages.get(count)
if price is None and (stars_price is None or int(stars_price) <= 0):
continue
plan: Dict[str, Any] = {
"id": f"{tariff.key}:hwid:{count}",
"tariff_key": tariff.key,
"tariff_name": tariff.name(lang),
"billing_model": tariff.billing_model,
"sale_mode": "hwid_devices",
"months": int(count),
"device_count": int(count),
"price": float(price or 0),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"title": f"+{count}",
"subtitle": tariff.name(lang),
}
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
def _serialize_tariff_change_target(
settings: Settings,
config: Any,
tariff: Any,
options: Dict[str, Any],
lang: str,
) -> Dict[str, Any]:
actions: List[Dict[str, Any]] = []
mode = str(options.get("mode") or "")
if mode == "period_to_period":
actions.append(
{
"mode": "recalc_days",
"kind": "free",
"title": "recalc_days",
"days_after": int(options.get("recalc_days") or 0),
"remaining_days": int(options.get("remaining_days") or 0),
}
)
paid_diff = float(options.get("paid_diff_rub") or 0)
if paid_diff > 0:
actions.append(
{
"mode": "paid_diff",
"kind": "payment",
"title": "paid_diff",
"price": paid_diff,
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
}
)
elif mode == "period_to_traffic":
actions.append(
{
"mode": "convert_days_to_gb",
"kind": "free",
"title": "convert_days_to_gb",
"converted_gb": float(options.get("converted_gb") or 0),
"remaining_days": int(options.get("remaining_days") or 0),
}
)
actions.extend(
{
"mode": "buy_package",
"kind": "payment",
"title": f"+{package.gb:g} GB",
"traffic_gb": float(package.gb),
"price": float(package.price),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
}
for package in (tariff.traffic_packages.rub if tariff.traffic_packages else [])
)
else:
for months in tariff.enabled_periods:
price = tariff.period_price(int(months), "rub")
if price:
actions.append(
{
"mode": "buy_period",
"kind": "payment",
"months": int(months),
"title": _format_months_title(int(months), lang),
"price": float(price),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
}
)
return {
"tariff_key": tariff.key,
"title": tariff.name(lang),
"description": tariff.description(lang),
"billing_model": tariff.billing_model,
"monthly_gb": tariff.monthly_gb,
"options": options,
"actions": actions,
}
def _serialize_payment_methods(
settings: Settings,
app: web.Application,
) -> List[Dict[str, Any]]:
labels = {
"severpay": "SeverPay",
"freekassa": "FreeKassa / СБП",
"platega_sbp": "Platega · СБП",
"platega_crypto": "Platega · Crypto",
"yookassa": "Банковская карта",
"stars": "Telegram Stars",
"cryptopay": "CryptoPay",
}
methods: List[Dict[str, Any]] = []
for method in settings.payment_methods_order:
method = method.lower()
if method == "severpay" and settings.SEVERPAY_ENABLED and _service_configured(app, "severpay_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "freekassa" and settings.FREEKASSA_ENABLED and _service_configured(app, "freekassa_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "platega_sbp" and settings.PLATEGA_ENABLED and settings.PLATEGA_SBP_ENABLED and _service_configured(app, "platega_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "platega_crypto" and settings.PLATEGA_ENABLED and settings.PLATEGA_CRYPTO_ENABLED and _service_configured(app, "platega_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "yookassa" and settings.YOOKASSA_ENABLED and _service_configured(app, "yookassa_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "stars" and settings.STARS_ENABLED:
methods.append({"id": method, "name": labels[method]})
elif method == "cryptopay" and settings.CRYPTOPAY_ENABLED and _service_configured(app, "cryptopay_service"):
methods.append({"id": method, "name": labels[method]})
return methods
def _service_configured(app: web.Application, key: str) -> bool:
service = app.get(key)
return bool(service and getattr(service, "configured", False))
def _sale_mode_base(sale_mode: str) -> str:
return str(sale_mode or "subscription").split("@", 1)[0].split("|", 1)[0]
def _sale_mode_tariff_key(sale_mode: str) -> Optional[str]:
if "@" not in str(sale_mode or ""):
return None
return str(sale_mode).split("@", 1)[1].split("|", 1)[0] or None
def _sale_mode_is_traffic(sale_mode: str) -> bool:
return _sale_mode_base(sale_mode) in {"traffic", "traffic_package", "topup", "premium_topup"}
def _sale_mode_is_hwid_devices(sale_mode: str) -> bool:
return _sale_mode_base(sale_mode) in {"hwid_device", "hwid_devices"}
async def _create_subscription_payment(
*,
request: web.Request,
session: AsyncSession,
user_id: int,
method: str,
months: Any,
price: float,
stars_price: Optional[int],
lang: str,
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
settings: Settings = request.app["settings"]
sale_mode = str(sale_mode or "subscription")
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
description = (
_traffic_payment_description(float(traffic_gb if traffic_gb is not None else months), lang)
if traffic_sale
else _hwid_devices_payment_description(int(float(months)), lang)
if hwid_devices_sale
else _payment_description(int(months), lang)
)
if method == "yookassa":
if not settings.YOOKASSA_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_yookassa_payment(
request, session, user_id, months, price, description, sale_mode=sale_mode, traffic_gb=traffic_gb
)
if method == "freekassa":
if not settings.FREEKASSA_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_freekassa_payment(
request, session, user_id, months, price, description, sale_mode=sale_mode, traffic_gb=traffic_gb
)
if method in ("platega", "platega_sbp", "platega_crypto"):
if not settings.PLATEGA_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
if method == "platega_sbp" and not settings.PLATEGA_SBP_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
if method == "platega_crypto" and not settings.PLATEGA_CRYPTO_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_platega_payment(
request, session, user_id, months, price, description, variant=method, sale_mode=sale_mode, traffic_gb=traffic_gb
)
if method == "severpay":
if not settings.SEVERPAY_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_severpay_payment(
request, session, user_id, months, price, description, sale_mode=sale_mode, traffic_gb=traffic_gb
)
if method == "cryptopay":
service: CryptoPayService = request.app["cryptopay_service"]
if not settings.CRYPTOPAY_ENABLED or not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
url = await service.create_invoice(
session=session,
user_id=user_id,
months=months,
amount=price,
description=description,
sale_mode=sale_mode,
url_kind="web",
)
if not url:
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{"ok": True, "action": "open_link", "payment_url": url, "payment_id": None}
)
if method == "stars":
if not settings.STARS_ENABLED or stars_price is None:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_stars_payment(
request, session, user_id, months, int(stars_price), description, sale_mode=sale_mode, traffic_gb=traffic_gb
)
return _json_error(400, "payment_unavailable", "Payment method unavailable")
async def _create_base_payment_record(
session: AsyncSession,
*,
user_id: int,
amount: float,
currency: str,
status: str,
description: str,
months: int,
provider: str,
sale_mode: Optional[str] = None,
tariff_key: Optional[str] = None,
purchased_gb: Optional[float] = None,
purchased_hwid_devices: Optional[int] = None,
) -> Payment:
payment = await payment_dal.create_payment_record(
session,
{
"user_id": user_id,
"amount": amount,
"currency": currency,
"status": status,
"description": description,
"subscription_duration_months": months,
"provider": provider,
"sale_mode": sale_mode,
"tariff_key": tariff_key,
"purchased_gb": purchased_gb,
"purchased_hwid_devices": purchased_hwid_devices,
},
)
await session.commit()
return payment
async def _create_yookassa_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: Any,
price: float,
description: str,
*,
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
settings: Settings = request.app["settings"]
service: YooKassaService = request.app["yookassa_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency="RUB",
status="pending_yookassa",
description=description,
months=int(float(months)) if not traffic_sale else int(float(traffic_gb or months)),
provider="yookassa",
sale_mode=sale_mode,
tariff_key=_sale_mode_tariff_key(sale_mode),
purchased_gb=float(traffic_gb or months) if traffic_sale else None,
purchased_hwid_devices=int(float(months)) if hwid_devices_sale else None,
)
metadata = {
"user_id": str(user_id),
"subscription_months": str(int(float(months)) if not traffic_sale and not hwid_devices_sale else 0),
"payment_db_id": str(payment.payment_id),
"sale_mode": sale_mode,
"source": "webapp",
}
if traffic_sale:
metadata["traffic_gb"] = _format_number_for_payload(traffic_gb or months)
if hwid_devices_sale:
metadata["hwid_devices"] = str(int(float(months)))
if _sale_mode_tariff_key(sale_mode):
metadata["tariff_key"] = _sale_mode_tariff_key(sale_mode)
response = await service.create_payment(
amount=price,
currency="RUB",
description=description,
metadata=metadata,
receipt_email=settings.YOOKASSA_DEFAULT_RECEIPT_EMAIL,
save_payment_method=bool(
settings.yookassa_autopayments_active
and settings.YOOKASSA_AUTOPAYMENTS_REQUIRE_CARD_BINDING
),
)
payment_url = response.get("confirmation_url") if response else None
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
await payment_dal.update_payment_status_by_db_id(
session,
payment.payment_id,
response.get("status", "pending"),
yk_payment_id=response.get("id"),
)
await session.commit()
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("YooKassa WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_freekassa_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: Any,
price: float,
description: str,
*,
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
settings: Settings = request.app["settings"]
service: FreeKassaService = request.app["freekassa_service"]
if not service or not service.configured or not service.payment_method_id:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=service.default_currency,
status="pending_freekassa",
description=description,
months=int(float(months)) if not traffic_sale else int(float(traffic_gb or months)),
provider="freekassa",
sale_mode=sale_mode,
tariff_key=_sale_mode_tariff_key(sale_mode),
purchased_gb=float(traffic_gb or months) if traffic_sale else None,
purchased_hwid_devices=int(float(months)) if hwid_devices_sale else None,
)
success, response_data = await service.create_order(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months,
amount=price,
currency=service.default_currency,
payment_method_id=service.payment_method_id,
ip_address=service.server_ip,
extra_params={"us_method": service.payment_method_id},
)
payment_url = response_data.get("location") if success else None
provider_id = response_data.get("orderHash") or response_data.get("orderId")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session, payment.payment_id, str(provider_id), payment.status
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("FreeKassa WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_platega_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: Any,
price: float,
description: str,
variant: str = "platega_sbp",
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
settings: Settings = request.app["settings"]
service: PlategaService = request.app["platega_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
if variant == "platega_crypto":
if not settings.PLATEGA_CRYPTO_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
platega_method_id = settings.PLATEGA_CRYPTO_METHOD
else:
if variant == "platega_sbp" and not settings.PLATEGA_SBP_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
platega_method_id = settings.platega_sbp_method_resolved
try:
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
status="pending_platega",
description=description,
months=int(float(months)) if not traffic_sale else int(float(traffic_gb or months)),
provider="platega",
sale_mode=sale_mode,
tariff_key=_sale_mode_tariff_key(sale_mode),
purchased_gb=float(traffic_gb or months) if traffic_sale else None,
purchased_hwid_devices=int(float(months)) if hwid_devices_sale else None,
)
months_for_provider = int(float(months)) if not traffic_sale else int(float(traffic_gb or months))
payload = json.dumps(
{
"payment_db_id": payment.payment_id,
"user_id": user_id,
"months": months_for_provider if not traffic_sale else 0,
"sale_mode": sale_mode,
"traffic_gb": _format_number_for_payload(traffic_gb or months) if traffic_sale else None,
"hwid_devices": int(float(months)) if hwid_devices_sale else None,
"source": "webapp",
"platega_variant": "crypto" if variant == "platega_crypto" else "sbp",
}
)
success, response_data = await service.create_transaction(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months_for_provider,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
description=description,
payload=payload,
payment_method=platega_method_id,
)
payment_url = (
response_data.get("redirect")
or response_data.get("url")
or response_data.get("paymentUrl")
) if success else None
provider_id = response_data.get("transactionId") or response_data.get("id")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session,
payment.payment_id,
str(provider_id),
str(response_data.get("status", payment.status)),
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("Platega WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_severpay_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: Any,
price: float,
description: str,
*,
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
settings: Settings = request.app["settings"]
service: SeverPayService = request.app["severpay_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
status="pending_severpay",
description=description,
months=int(float(months)) if not traffic_sale else int(float(traffic_gb or months)),
provider="severpay",
sale_mode=sale_mode,
tariff_key=_sale_mode_tariff_key(sale_mode),
purchased_gb=float(traffic_gb or months) if traffic_sale else None,
purchased_hwid_devices=int(float(months)) if hwid_devices_sale else None,
)
success, response_data = await service.create_payment(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
description=description,
)
payment_url = (
response_data.get("url")
or response_data.get("payment_url")
or response_data.get("paymentUrl")
) if success else None
provider_id = response_data.get("id") or response_data.get("uid")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session, payment.payment_id, str(provider_id), payment.status
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("SeverPay WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_stars_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: Any,
stars_price: int,
description: str,
sale_mode: str = "subscription",
traffic_gb: Optional[float] = None,
) -> web.Response:
bot: Bot = request.app["bot"]
try:
traffic_sale = _sale_mode_is_traffic(sale_mode)
hwid_devices_sale = _sale_mode_is_hwid_devices(sale_mode)
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=float(stars_price),
currency="XTR",
status="pending_stars",
description=description,
months=int(float(months)) if not traffic_sale else int(float(traffic_gb or months)),
provider="telegram_stars",
sale_mode=sale_mode,
tariff_key=_sale_mode_tariff_key(sale_mode),
purchased_gb=float(traffic_gb or months) if traffic_sale else None,
purchased_hwid_devices=int(float(months)) if hwid_devices_sale else None,
)
payload_units = traffic_gb if traffic_sale and traffic_gb is not None else months
payload = f"{payment.payment_id}:{_format_number_for_payload(payload_units)}:{sale_mode}"
prices = [LabeledPrice(label=description, amount=stars_price)]
create_invoice_link = getattr(bot, "create_invoice_link", None)
if callable(create_invoice_link):
invoice_url = await create_invoice_link(
title=description,
description=description,
payload=payload,
provider_token="", # Required to be empty for Telegram Stars (XTR) per Telegram Bot API.
currency="XTR",
prices=prices,
)
return web.json_response(
{
"ok": True,
"action": "open_invoice",
"payment_url": invoice_url,
"payment_id": payment.payment_id,
}
)
await bot.send_invoice(
chat_id=user_id,
title=description,
description=description,
payload=payload,
provider_token="",
currency="XTR",
prices=prices,
)
return web.json_response(
{
"ok": True,
"action": "invoice_sent",
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("Stars WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create invoice")
def _normalize_language(lang: Optional[str]) -> str:
value = (lang or "ru").split("-")[0].lower()
return value if value in {"ru", "en"} else "ru"
def _format_remaining(seconds: int, lang: str) -> str:
if seconds <= 0:
if lang == "en":
return "Subscription inactive"
return "Подписка не активна"
days, rem = divmod(seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes = rem // 60
if lang == "en":
if days > 0:
return f"{days} d. {hours} h."
if hours > 0:
return f"{hours} h. {minutes} min."
return f"{max(1, minutes)} min."
if days > 0:
return f"{days} д. {hours} ч."
if hours > 0:
return f"{hours} ч. {minutes} мин."
return f"{max(1, minutes)} мин."
def _coerce_int_or_none(value: Optional[Any]) -> Optional[int]:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _format_bytes(value: Optional[Any]) -> str:
if value is None:
return "N/A"
try:
size = float(value)
except (TypeError, ValueError):
return str(value)
if size <= 0:
return "∞"
units = ["B", "KB", "MB", "GB", "TB"]
index = 0
while size >= 1024 and index < len(units) - 1:
size /= 1024
index += 1
return f"{size:.2f} {units[index]}"
def _device_hwid_token(hwid: str) -> str:
return hashlib.sha256(str(hwid or "").encode()).hexdigest()[:32]
def _shorten_hwid_for_display(hwid: Optional[str], max_length: int = 24) -> str:
value = str(hwid or "").strip()
if len(value) <= max_length:
return value
return f"{value[:8]}...{value[-6:]}"
def _normalize_devices_response(devices_response: Any) -> List[Dict[str, Any]]:
if isinstance(devices_response, dict):
devices = devices_response.get("devices") or []
else:
devices = devices_response or []
if not isinstance(devices, list):
return []
return [device for device in devices if isinstance(device, dict)]
def _format_devices_limit(max_devices: Optional[int]) -> str:
if max_devices in (None, 0):
return "Unlimited"
return str(max_devices)
def _format_device_datetime(value: Any) -> str:
if not value:
return ""
text = str(value)
try:
normalized = datetime.fromisoformat(text.replace("Z", "+00:00"))
return normalized.strftime("%d.%m.%Y %H:%M")
except Exception:
return text
def _serialize_device(device: Dict[str, Any], index: int) -> Dict[str, Any]:
hwid = str(device.get("hwid") or "").strip()
model = str(device.get("deviceModel") or "").strip()
platform = str(device.get("platform") or "").strip()
os_version = str(device.get("osVersion") or "").strip()
user_agent = str(device.get("userAgent") or "").strip()
display_name = model or platform or f"Device {index}"
platform_label = " ".join(part for part in (platform, os_version) if part).strip()
return {
"index": index,
"display_name": display_name,
"platform": platform,
"os_version": os_version,
"platform_label": platform_label,
"user_agent": user_agent,
"created_at": device.get("createdAt"),
"created_at_text": _format_device_datetime(device.get("createdAt")),
"hwid_short": _shorten_hwid_for_display(hwid),
"token": _device_hwid_token(hwid) if hwid else "",
"can_disconnect": bool(hwid),
}
def _format_months_title(months: int, lang: str) -> str:
if lang == "en":
if months == 1:
return "1 month"
return f"{months} months"
if months == 1:
return "1 месяц"
if 2 <= months <= 4:
return f"{months} месяца"
return f"{months} месяцев"
def _format_number_for_payload(value: Any) -> str:
numeric = float(value or 0)
return str(int(numeric)) if numeric.is_integer() else f"{numeric:g}"
def _format_traffic_title(traffic_gb: float, lang: str) -> str:
return f"{_format_number_for_payload(traffic_gb)} GB"
def _traffic_payment_description(traffic_gb: float, lang: str) -> str:
if lang == "en":
return f"Traffic package {_format_traffic_title(traffic_gb, lang)}"
return f"Пакет трафика {_format_traffic_title(traffic_gb, lang)}"
def _hwid_devices_payment_description(device_count: int, lang: str) -> str:
if lang == "en":
return f"HWID device package +{device_count}"
return f"Докупка устройств HWID +{device_count}"
def _resolve_numeric_option_key(options: Dict[Any, Any], target: float) -> Optional[Any]:
for key in options:
try:
if abs(float(key) - float(target)) < 0.000001:
return key
except (TypeError, ValueError):
continue
return None
def _payment_description(months: int, lang: str) -> str:
if lang == "en":
return f"Subscription for {_format_months_title(months, lang)}"
return f"Подписка на {_format_months_title(months, lang)}"