auth.py

import hmac
import hashlib
import json
import os
import time
import urllib.parse

from fastapi import Request, HTTPException
from config import TELEGRAM_BOT_TOKEN

DEV_MODE = os.environ.get("DEV_MODE", "").lower() in ("true", "1", "yes")


def verify_init_data(init_data: str) -> dict | None:
    if DEV_MODE and init_data.startswith("dev_"):
        parts = init_data.split("_")
        return {
            "id": int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 5703679073,
            "first_name": parts[2] if len(parts) > 2 else "Dev",
            "last_name": "",
            "username": parts[3] if len(parts) > 3 else "devuser",
            "language_code": "en",
            "is_premium": False,
        }

    try:
        parsed = dict(urllib.parse.parse_qsl(init_data))
    except Exception:
        return None

    hash_val = parsed.pop("hash", None)
    if not hash_val:
        return None

    auth_date = parsed.get("auth_date")
    if not auth_date:
        return None
    try:
        if time.time() - int(auth_date) > 86400:
            return None
    except (ValueError, TypeError):
        return None

    sorted_items = sorted(parsed.items())
    check_str = "\n".join(f"{k}={v}" for k, v in sorted_items)

    secret_key = hmac.new(b"WebAppData", TELEGRAM_BOT_TOKEN.encode(), hashlib.sha256).digest()
    computed_hash = hmac.new(secret_key, check_str.encode(), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(computed_hash, hash_val):
        return None

    try:
        user_info = json.loads(parsed.get("user", "{}"))
    except (json.JSONDecodeError, TypeError):
        user_info = {}

    return {
        "id": user_info.get("id"),
        "first_name": user_info.get("first_name", ""),
        "last_name": user_info.get("last_name", ""),
        "username": user_info.get("username", ""),
        "language_code": user_info.get("language_code", "en"),
        "is_premium": user_info.get("is_premium", False),
        "photo_url": user_info.get("photo_url", ""),
    }


async def get_current_user(request: Request) -> dict:
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("tma "):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
    user_data = verify_init_data(auth_header[4:])
    if not user_data or not user_data.get("id"):
        raise HTTPException(status_code=401, detail="Invalid init data")
    return user_data


async def optional_current_user(request: Request) -> dict | None:
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("tma "):
        return None
    return verify_init_data(auth_header[4:])