GmRelayBot/tests/e2e/helpers/telegram_init_data.py

"""
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,
    )