""" Generate Telegram Mini App initData and Login Widget payloads for local E2E tests. This mirrors GmRelay.Shared.Telegram.TelegramAuthPayloadBuilder so the Python E2E runner can produce authentication payloads that pass GmRelay.Web.Services.TelegramAuthService validation without talking to real Telegram servers. """ from __future__ import annotations import hashlib import hmac import json import time import urllib.parse from dataclasses import dataclass from typing import Optional def _login_widget_secret_key(bot_token: str) -> bytes: return hashlib.sha256(bot_token.encode("utf-8")).digest() def _mini_app_secret_key(bot_token: str) -> bytes: return hmac.new( key=b"WebAppData", msg=bot_token.encode("utf-8"), digestmod=hashlib.sha256, ).digest() def compute_login_widget_hash(bot_token: str, values: dict[str, str]) -> str: """Compute HMAC-SHA256 hash used by Telegram Login Widget callbacks.""" data_check_string = "\n".join( f"{k}={values[k]}" for k in sorted(values.keys()) if k != "hash" ) secret_key = _login_widget_secret_key(bot_token) return hmac.new( key=secret_key, msg=data_check_string.encode("utf-8"), digestmod=hashlib.sha256, ).hexdigest() def compute_mini_app_hash(bot_token: str, values: dict[str, str]) -> str: """Compute HMAC-SHA256 hash used by Telegram Mini App initData.""" data_check_string = "\n".join( f"{k}={values[k]}" for k in sorted(values.keys()) if k != "hash" ) secret_key = _mini_app_secret_key(bot_token) return hmac.new( key=secret_key, msg=data_check_string.encode("utf-8"), digestmod=hashlib.sha256, ).hexdigest() @dataclass(frozen=True) class LoginWidgetResult: telegram_id: int first_name: str last_name: Optional[str] username: Optional[str] photo_url: Optional[str] auth_date: int hash: str query_string: str def build_login_widget( bot_token: str, telegram_id: int, first_name: str, last_name: Optional[str] = None, username: Optional[str] = None, photo_url: Optional[str] = None, auth_date: Optional[int] = None, ) -> LoginWidgetResult: """Build a Telegram Login Widget query string and hash.""" timestamp = auth_date if auth_date is not None else int(time.time()) values: dict[str, str] = { "auth_date": str(timestamp), "first_name": first_name, "id": str(telegram_id), } if last_name: values["last_name"] = last_name if photo_url: values["photo_url"] = photo_url if username: values["username"] = username hash_value = compute_login_widget_hash(bot_token, values) values["hash"] = hash_value query_string = "&".join( f"{urllib.parse.quote(k)}={urllib.parse.quote(v)}" for k, v in values.items() ) return LoginWidgetResult( telegram_id=telegram_id, first_name=first_name, last_name=last_name, username=username, photo_url=photo_url, auth_date=timestamp, hash=hash_value, query_string=query_string, ) @dataclass(frozen=True) class MiniAppInitDataResult: telegram_id: int first_name: str last_name: Optional[str] username: Optional[str] photo_url: Optional[str] auth_date: int hash: str init_data_raw: str def build_mini_app_init_data( bot_token: str, telegram_id: int, first_name: str, last_name: Optional[str] = None, username: Optional[str] = None, photo_url: Optional[str] = None, language_code: Optional[str] = None, is_premium: bool = False, chat_id: Optional[int] = None, chat_type: Optional[str] = None, chat_title: Optional[str] = None, query_id: Optional[str] = None, start_param: Optional[str] = None, auth_date: Optional[int] = None, ) -> MiniAppInitDataResult: """Build a Telegram Mini App initData raw string.""" user_payload: dict[str, object] = { "id": telegram_id, "first_name": first_name, } if last_name is not None: user_payload["last_name"] = last_name if username is not None: user_payload["username"] = username if photo_url is not None: user_payload["photo_url"] = photo_url if language_code is not None: user_payload["language_code"] = language_code if is_premium: user_payload["is_premium"] = True user_json = json.dumps(user_payload, separators=(",", ":")) timestamp = auth_date if auth_date is not None else int(time.time()) values: dict[str, str] = { "auth_date": str(timestamp), "user": user_json, } if query_id: values["query_id"] = query_id if start_param: values["start_param"] = start_param if chat_id is not None: chat_payload: dict[str, object] = {"id": chat_id, "type": chat_type or "private"} if chat_title is not None: chat_payload["title"] = chat_title values["chat"] = json.dumps(chat_payload, separators=(",", ":")) hash_value = compute_mini_app_hash(bot_token, values) pairs = [ f"{urllib.parse.quote(k)}={urllib.parse.quote(v)}" for k, v in values.items() ] pairs.append(f"hash={hash_value}") init_data_raw = "&".join(pairs) return MiniAppInitDataResult( telegram_id=telegram_id, first_name=first_name, last_name=last_name, username=username, photo_url=photo_url, auth_date=timestamp, hash=hash_value, init_data_raw=init_data_raw, )