Spaces:
Sleeping
Sleeping
from fastapi import Depends, HTTPException, status | |
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
from firebase_admin import auth | |
from .firebase import db | |
import time | |
security = HTTPBearer() | |
def get_user(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
if not credentials: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="Bearer authentication required" | |
) | |
try: | |
# Utiliser une valeur valide pour clock_skew_seconds (entre 0 et 60) | |
decoded_token = auth.verify_id_token( | |
credentials.credentials, | |
check_revoked=True, | |
clock_skew_seconds=60 # Valeur maximale autorisée | |
) | |
user_id = decoded_token['uid'] | |
user_doc = db.collection('users').document(user_id).get() | |
if not user_doc.exists: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="User not found in Firestore" | |
) | |
user_data = user_doc.to_dict() | |
decoded_token['role'] = user_data.get('role', 'user_extern') | |
return decoded_token | |
except Exception as e: | |
raise HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail=f"Invalid authentication credentials: {str(e)}" | |
) | |
def require_role(allowed_roles): | |
def role_checker(user_info=Depends(get_user)): | |
if user_info['role'] not in allowed_roles: | |
raise HTTPException( | |
status_code=status.HTTP_403_FORBIDDEN, | |
detail="Insufficient permissions" | |
) | |
return user_info | |
return role_checker |