mirror of
https://gitverse.ru/kpa39l/telepost.git
synced 2026-09-29 18:05:08 +00:00
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
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)
|
|
} |