TelegramBotAdmin/wbb/admin/telegram_webapp.py

from __future__ import annotations

import hashlib
import hmac
import json
from datetime import UTC, datetime
from typing import Any
from urllib.parse import parse_qsl


class TelegramWebAppAuthError(ValueError):
    pass


def verify_telegram_webapp_init_data(
    init_data: str,
    bot_token: str,
    *,
    max_age_seconds: int = 3600,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Validate Telegram Mini App init data and return the signed user."""
    if not init_data or not bot_token:
        raise TelegramWebAppAuthError("缺少 Telegram Mini App 鉴权信息。")
    try:
        values = dict(parse_qsl(init_data, keep_blank_values=True, strict_parsing=True))
    except ValueError as exc:
        raise TelegramWebAppAuthError("Telegram Mini App 鉴权信息格式无效。") from exc
    supplied_hash = values.pop("hash", "")
    if len(supplied_hash) != 64:
        raise TelegramWebAppAuthError("Telegram Mini App 签名无效。")
    data_check_string = "\n".join(
        f"{key}={value}" for key, value in sorted(values.items())
    )
    secret_key = hmac.new(
        b"WebAppData",
        bot_token.encode("utf-8"),
        hashlib.sha256,
    ).digest()
    expected_hash = hmac.new(
        secret_key,
        data_check_string.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected_hash, supplied_hash):
        raise TelegramWebAppAuthError("Telegram Mini App 签名无效。")

    try:
        auth_date = int(values.get("auth_date") or 0)
    except (TypeError, ValueError) as exc:
        raise TelegramWebAppAuthError("Telegram Mini App 鉴权时间无效。") from exc
    current = now or datetime.now(UTC)
    age_seconds = int(current.timestamp()) - auth_date
    if auth_date <= 0 or age_seconds < -30 or age_seconds > max_age_seconds:
        raise TelegramWebAppAuthError("Telegram Mini App 登录已过期，请从 Bot 重新打开。")

    try:
        user = json.loads(values.get("user") or "")
    except (TypeError, json.JSONDecodeError) as exc:
        raise TelegramWebAppAuthError("Telegram 用户信息无效。") from exc
    if not isinstance(user, dict):
        raise TelegramWebAppAuthError("Telegram 用户信息无效。")
    try:
        user_id = int(user.get("id"))
    except (TypeError, ValueError) as exc:
        raise TelegramWebAppAuthError("Telegram 用户编号无效。") from exc
    if user_id <= 0 or user.get("is_bot"):
        raise TelegramWebAppAuthError("Telegram 用户身份无效。")
    return {
        "user_id": user_id,
        "username": str(user.get("username") or ""),
        "display_name": " ".join(
            str(user.get(key) or "").strip()
            for key in ("first_name", "last_name")
            if str(user.get(key) or "").strip()
        )
        or f"技师 {user_id}",
        "language_code": str(user.get("language_code") or ""),
        "auth_date": auth_date,
        "query_id": str(values.get("query_id") or ""),
    }