initial commit (backend)

auth.py

@@ -0,0 +1,108 @@
+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