108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
import time
|
|
import hmac
|
|
import hashlib
|
|
import json
|
|
from urllib.parse import unquote
|
|
from fastapi import Request, HTTPException
|
|
from config import BOT_TOKEN, MAX_INITDATA_AGE
|
|
|
|
|
|
async def verify_telegram_initdata(init_data: str):
|
|
"""Проверяет подпись initData от Telegram с проверкой срока действия"""
|
|
try:
|
|
# Парсим initData
|
|
data = {}
|
|
for chunk in init_data.split('&'):
|
|
if '=' in chunk:
|
|
key, value = chunk.split('=', 1)
|
|
data[key] = unquote(value)
|
|
|
|
# Извлекаем hash
|
|
received_hash = data.pop('hash', '')
|
|
|
|
if not received_hash:
|
|
raise HTTPException(401, "Hash not found in initData")
|
|
|
|
# Проверяем auth_date (не старше MAX_INITDATA_AGE из config)
|
|
auth_date_str = data.get('auth_date', '0')
|
|
try:
|
|
auth_date = int(auth_date_str)
|
|
current_time = int(time.time())
|
|
|
|
if current_time - auth_date > MAX_INITDATA_AGE:
|
|
raise HTTPException(401, "initData expired")
|
|
|
|
if auth_date > current_time + 60:
|
|
raise HTTPException(401, "initData from future")
|
|
|
|
except ValueError:
|
|
raise HTTPException(401, "Invalid auth_date")
|
|
|
|
# Сортируем ключи и создаем строку для проверки
|
|
check_arr = []
|
|
for key in sorted(data.keys()):
|
|
check_arr.append(f"{key}={data[key]}")
|
|
check_string = '\n'.join(check_arr)
|
|
|
|
# Создаем секретный ключ
|
|
secret_key = hmac.new(
|
|
'WebAppData'.encode(),
|
|
BOT_TOKEN.encode(),
|
|
hashlib.sha256
|
|
).digest()
|
|
|
|
# Вычисляем hash
|
|
calculated_hash = hmac.new(
|
|
secret_key,
|
|
check_string.encode(),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
if calculated_hash != received_hash:
|
|
raise HTTPException(401, "Invalid initData signature")
|
|
|
|
# Извлекаем данные пользователя
|
|
user_data_str = data.get('user', '{}')
|
|
try:
|
|
user_data = json.loads(user_data_str)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(401, "Invalid user data JSON")
|
|
|
|
user_id = user_data.get('id')
|
|
username = user_data.get('username', '')
|
|
first_name = user_data.get('first_name', '')
|
|
|
|
if not user_id:
|
|
raise HTTPException(401, "User ID not found in initData")
|
|
|
|
return user_id, username
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(401, f"Auth error: {str(e)}")
|
|
|
|
|
|
async def require_auth(request: Request):
|
|
"""Извлекает и проверяет initData из запроса"""
|
|
# Пробуем из query-параметров
|
|
init_data = request.query_params.get('initData', '')
|
|
|
|
if not init_data:
|
|
# Пробуем из заголовков
|
|
init_data = request.headers.get('X-Telegram-InitData', '')
|
|
|
|
if not init_data:
|
|
# Пробуем из тела для POST запросов
|
|
# (некоторые клиенты могут отправлять initData в теле)
|
|
pass
|
|
|
|
if not init_data:
|
|
raise HTTPException(401, "initData required. Pass as query parameter 'initData' or header 'X-Telegram-InitData'")
|
|
|
|
# Декодируем URL-encoded initData если нужно
|
|
if '%' in init_data:
|
|
init_data = unquote(init_data)
|
|
|
|
user_id, username = await verify_telegram_initdata(init_data)
|
|
return user_id, username |