import hashlib import hmac from typing import Dict, Any, Optional from datetime import datetime, timedelta from fastapi import HTTPException, status from sqlalchemy.orm import Session from ..database.config import settings from ..models.user import User from ..schemas.user import UserCreate, UserInDB from .jwt import create_access_token def verify_telegram_login_data(auth_data: Dict[str, Any]) -> bool: """ Verify the authenticity of Telegram login data """ if not settings.TELEGRAM_BOT_TOKEN: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Telegram bot token not configured" ) # Extract the hash received_hash = auth_data.pop('hash', None) if not received_hash: return False # Get the auth date auth_date = auth_data.pop('auth_date', None) if auth_date: # Check if the data is not too old (more than 1 day) auth_time = datetime.fromtimestamp(int(auth_date)) if (datetime.now() - auth_time).days > 0: return False # Sort the remaining parameters alphabetically by key data_check_arr = [] for key, value in sorted(auth_data.items()): data_check_arr.append(f"{key}={value}") data_check_string = '\n'.join(data_check_arr) # Create secret key using SHA256 of bot token secret_key = hashlib.sha256(settings.TELEGRAM_BOT_TOKEN.encode()).digest() # Calculate HMAC-SHA256 signature calculated_hash = hmac.new( secret_key, data_check_string.encode(), hashlib.sha256 ).hexdigest() # Compare the calculated hash with the received hash return calculated_hash == received_hash def authenticate_user_telegram(db: Session, auth_data: Dict[str, Any]): """ Authenticate user via Telegram login data and return JWT token """ # Verify the authenticity of the data if not verify_telegram_login_data(auth_data.copy()): # Copy to avoid modifying original raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Telegram login data" ) # Extract user data telegram_id = str(auth_data.get('id')) username = auth_data.get('username') first_name = auth_data.get('first_name') last_name = auth_data.get('last_name') photo_url = auth_data.get('photo_url') # Try to find existing user by telegram_id user = db.query(User).filter(User.telegram_id == telegram_id).first() if user: # Update user data if it has changed user.username = username user.first_name = first_name user.last_name = last_name user.avatar_url = photo_url user.is_verified = True db.commit() db.refresh(user) else: # Create new user user = User( telegram_id=telegram_id, username=username, first_name=first_name, last_name=last_name, avatar_url=photo_url, is_verified=True, is_active=True ) db.add(user) db.commit() db.refresh(user) # Create JWT token access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username or user.telegram_id}, expires_delta=access_token_expires ) return { "access_token": access_token, "token_type": "bearer", "user": UserInDB.from_orm(user) if hasattr(UserInDB, 'from_orm') else UserInDB.model_validate(user) }