app/api/auth.py

from __future__ import annotations

import hashlib
import hmac
import json
from urllib.parse import parse_qs, unquote

from fastapi import Header, HTTPException, status

from app.core.config import get_settings, is_user_allowed

_DEV_USER_ID = 377324522


def _extract_user_id(init_data: str) -> int:
    """Parse user.id from an initData query string without verifying the signature."""
    parsed = parse_qs(init_data)
    user_data = parsed.get("user", [""])[0]
    if not user_data:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing user")
    try:
        user_obj = json.loads(unquote(user_data))
        return int(user_obj["id"])
    except (json.JSONDecodeError, KeyError, ValueError) as exc:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "bad user data") from exc


def _validate_init_data(init_data: str, bot_token: str) -> int:
    """Verify Telegram Mini App initData HMAC-SHA256 and return the user_id."""
    parsed = parse_qs(init_data)
    check_hash = parsed.get("hash", [""])[0]
    if not check_hash:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing hash")

    data_pairs = []
    for key_val in init_data.split("&"):
        key, _, val = key_val.partition("=")
        if key != "hash":
            data_pairs.append(f"{key}={unquote(val)}")
    data_pairs.sort()
    data_check_string = "\n".join(data_pairs)

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

    if not hmac.compare_digest(computed, check_hash):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid hash")

    return _extract_user_id(init_data)


def _is_dev_mode() -> bool:
    """True when running behind ngrok (local dev tunnel)."""
    return "ngrok" in (get_settings().webhook_base_url or "")


def resolve_user_id(init_data: str | None) -> int:
    """Resolve the Telegram user_id from Mini App initData.

    With initData present: verify the HMAC in production; in dev (ngrok) parse the
    real user.id leniently so each Telegram user keeps a separate account instead of
    all collapsing onto the dev fallback. Without initData: dev returns a fixed id,
    production rejects with 401.
    """
    if init_data:
        if _is_dev_mode():
            return _extract_user_id(init_data)
        token = get_settings().telegram_bot_token.get_secret_value()
        return _validate_init_data(init_data, token)
    if _is_dev_mode():
        return _DEV_USER_ID
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing initData")


async def get_current_user_id(
    x_telegram_init_data: str | None = Header(default=None),
) -> int:
    """FastAPI dependency: resolve the Telegram user_id and enforce the allowlist."""
    user_id = resolve_user_id(x_telegram_init_data)
    if not is_user_allowed(user_id):
        raise HTTPException(status.HTTP_403_FORBIDDEN, "user not allowed")
    return user_id