File size: 3,000 Bytes
3470cf9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | 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
|