admin_app/security.py - master - medmikhr/ctk_dm_bot_tg

admin_app/security.py
86 строк3 KB
DAnaKhristenko
тг и приложение для словарей
30 окт 2025, 23:43
30 окт 2025, 23:437177793
from __future__ import annotations

import hashlib
import hmac
import json
import time
from typing import Any, Dict, Optional
from urllib.parse import parse_qsl

from pydantic import BaseModel, ValidationError


class TelegramUser(BaseModel):
    id: int
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    username: Optional[str] = None
    language_code: Optional[str] = None
    is_premium: Optional[bool] = None


class TelegramInitData(BaseModel):
    query_id: Optional[str] = None
    user: TelegramUser
    auth_date: int
    start_param: Optional[str] = None
    hash: str


class TelegramAuthError(Exception):
    pass


def parse_init_data(init_data: str) -> Dict[str, Any]:
    try:
        parsed_items = dict(parse_qsl(init_data, strict_parsing=True))
    except ValueError as exc:
        raise TelegramAuthError("Failed to parse init data") from exc

    if "hash" not in parsed_items:
        raise TelegramAuthError("Init data missing hash")

    if "user" not in parsed_items:
        raise TelegramAuthError("Init data missing user payload")

    try:
        parsed_items["user"] = json.loads(parsed_items["user"])
    except json.JSONDecodeError as exc:
        raise TelegramAuthError("Invalid user payload in init data") from exc

    return parsed_items


def build_data_check_string(data: Dict[str, Any]) -> str:
    prepared_pairs = []
    for key, value in data.items():
        if key == "hash":
            continue
        if isinstance(value, (dict, list)):
            value = json.dumps(value, separators=(",", ":"), ensure_ascii=False)
        prepared_pairs.append(f"{key}={value}")
    prepared_pairs.sort()
    return "\n".join(prepared_pairs)


def verify_init_data(init_data: str, bot_token: str, *, max_age_seconds: int = 300) -> TelegramInitData:
    payload = parse_init_data(init_data)
    data_check_string = build_data_check_string(payload)

    secret_key = hmac.new("WebAppData".encode(), bot_token.encode(), hashlib.sha256).digest()
    calculated_hash = hmac.new(secret_key, data_check_string.encode(), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(calculated_hash, payload["hash"]):
        raise TelegramAuthError("Invalid Telegram init data signature")

    auth_date = int(payload.get("auth_date", 0))
    if auth_date == 0:
        raise TelegramAuthError("Init data missing auth_date")

    if int(time.time()) - auth_date > max_age_seconds:
        raise TelegramAuthError("Telegram init data expired")

    try:
        return TelegramInitData(**payload)
    except ValidationError as exc:
        raise TelegramAuthError("Invalid init data payload") from exc