aiform_prod/backend/app/services/telegram_auth.py

"""
Telegram WebApp (Mini App) auth helper.

В этом модуле:
- Парсим и валидируем initData от Telegram WebApp
- Проверяем подпись по токену бота из настроек
- Возвращаем разобранные данные пользователя Telegram
"""

import hashlib
import hmac
import logging
from typing import Dict, Any
from urllib.parse import parse_qsl

from ..config import settings

logger = logging.getLogger(__name__)


class TelegramAuthError(Exception):
    """Ошибка проверки подлинности Telegram initData."""


def _parse_init_data(init_data: str) -> Dict[str, Any]:
    """
    Разбирает строку initData в словарь.

    Формат initData — это query string, см. Telegram WebApp docs.
    """
    data: Dict[str, Any] = {}
    for key, value in parse_qsl(init_data, keep_blank_values=True):
        data[key] = value
    return data


def verify_telegram_init_data(init_data: str) -> Dict[str, Any]:
    """
    Проверяет подпись initData согласно Telegram WebApp правилам.

    Алгоритм из официальной документации:
    - Берём токен бота: BOT_TOKEN
    - Вычисляем secret_key = HMAC_SHA256("WebAppData", BOT_TOKEN)
    - Собираем data_check_string: строки "<key>=<value>" по всем полям, кроме 'hash',
      отсортированные по key, соединённые '\n'
    - Считаем хэш: HMAC_SHA256(secret_key, data_check_string)
    - Сравниваем с полем 'hash' из initData (hex)
    """
    if not init_data:
        logger.warning("[TG] verify_telegram_init_data: init_data пустой")
        raise TelegramAuthError("init_data is empty")

    bot_token = (getattr(settings, "telegram_bot_token", None) or "").strip()
    if not bot_token:
        logger.warning("[TG] verify_telegram_init_data: TELEGRAM_BOT_TOKEN не задан в .env")
        raise TelegramAuthError("Telegram bot token is not configured")

    parsed = _parse_init_data(init_data)
    logger.info("[TG] initData распарсен, ключи: %s", list(parsed.keys()))

    received_hash = parsed.pop("hash", None)
    if not received_hash:
        logger.warning("[TG] В initData отсутствует поле hash")
        raise TelegramAuthError("Missing hash in init_data")

    # Формируем data_check_string
    data_check_items = []
    for key in sorted(parsed.keys()):
        value = parsed[key]
        data_check_items.append(f"{key}={value}")
    data_check_string = "\n".join(data_check_items)

    # secret_key = HMAC_SHA256("WebAppData", BOT_TOKEN)
    secret_key = hmac.new(
        key="WebAppData".encode("utf-8"),
        msg=bot_token.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()

    # HMAC_SHA256(secret_key, data_check_string)
    calculated_hash = hmac.new(
        key=secret_key,
        msg=data_check_string.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(calculated_hash, received_hash):
        logger.warning("[TG] Подпись initData не совпадает (неверный токен бота или поддельные данные)")
        raise TelegramAuthError("Invalid init_data hash")

    return parsed


def extract_telegram_user(init_data: str) -> Dict[str, Any]:
    """
    Валидирует initData и возвращает данные пользователя Telegram.

    В field `user` лежит JSON-строка с полями:
    {
      "id": 123456789,
      "first_name": "...",
      "last_name": "...",
      "username": "...",
      ...
    }
    """
    import json

    parsed = verify_telegram_init_data(init_data)

    user_raw = parsed.get("user")
    if not user_raw:
        logger.warning("[TG] В initData отсутствует поле user")
        raise TelegramAuthError("No user field in init_data")

    try:
        user_obj = json.loads(user_raw)
    except Exception as e:
        raise TelegramAuthError(f"Failed to parse user JSON: {e}") from e

    if "id" not in user_obj:
        raise TelegramAuthError("Telegram user.id is missing")

    return {
        "telegram_user_id": str(user_obj.get("id")),
        "username": user_obj.get("username"),
        "first_name": user_obj.get("first_name"),
        "last_name": user_obj.get("last_name"),
        "language_code": user_obj.get("language_code"),
        "raw": user_obj,
    }