55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from firebase_admin import auth
|
|
from .firebase_config import get_db
|
|
|
|
http_bearer_scheme = HTTPBearer()
|
|
|
|
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(http_bearer_scheme)):
|
|
"""
|
|
Verifies the Firebase ID token and retrieves the user document from Firestore.
|
|
"""
|
|
token = credentials.credentials
|
|
try:
|
|
decoded_token = auth.verify_id_token(token)
|
|
uid = decoded_token['uid']
|
|
|
|
db = get_db()
|
|
user_doc = db.collection('users').document(uid).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()
|
|
return user_data
|
|
|
|
except auth.InvalidIdTokenError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid Firebase ID token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"An error occurred: {e}",
|
|
)
|
|
|
|
def is_user(current_user: dict = Depends(get_current_user)):
|
|
"""
|
|
Dependency to ensure the current user has 'user' or 'admin' role.
|
|
"""
|
|
if current_user.get('role') not in ["user", "admin"]:
|
|
raise HTTPException(status_code=403, detail="Not enough permissions")
|
|
return current_user
|
|
|
|
def is_admin(current_user: dict = Depends(get_current_user)):
|
|
"""
|
|
Dependency to ensure the current user has 'admin' role.
|
|
"""
|
|
if current_user.get('role') != "admin":
|
|
raise HTTPException(status_code=403, detail="Requires admin role")
|
|
return current_user |