Init

app/security.py

@@ -0,0 +1,54 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from firebase_admin import auth
+from .firebase_config import get_db
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
+
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    """
+    Verifies the Firebase ID token and retrieves the user document from Firestore.
+    """
+    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