middleware + updated dependencies

mmiqball · mmiqball · commit 596bbfcdd5b8 · 2025-02-25T18:43:25.000-05:00
diff --git a/backend/app/middleware/auth.py b/backend/app/middleware/auth.py
@@ -0,0 +1,49 @@
+import logging
+from typing import List, Dict, Any
+
+from fastapi import Depends, HTTPException, Request, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+
+from ..utilities.constants import LOGGER_NAME
+from ..utilities.service_utils import get_auth_service
+
+security = HTTPBearer()
+logger = logging.getLogger(LOGGER_NAME("auth"))
+
+
+def has_roles(required_roles: List[str]):
+    """
+    FastAPI dependency that checks if the authenticated user has one of the required roles.
+    
+    Args:
+        required_roles: List of roles that can access the endpoint
+        
+    Returns:
+        A dependency that validates the user has one of the specified roles
+    
+    Example:
+        @app.get("/admin-only")
+        async def admin_endpoint(authorized: bool = has_roles(["admin"])):
+            return {"message": "You have admin access"}
+    """
+    async def role_validator(
+        request: Request,
+        credentials: HTTPAuthorizationCredentials = Depends(security),
+        auth_service = Depends(get_auth_service)
+    ) -> bool:
+        # Get the token from authorization header
+        token = credentials.credentials
+        
+        # Use the auth service to check if the user has the required role
+        is_authorized = auth_service.is_authorized_by_role(token, set(required_roles))
+        
+        if not is_authorized:
+            logger.warning(f"Access denied: user doesn't have required roles: {required_roles}")
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Access denied: requires one of these roles: {required_roles}"
+            )
+        
+        return True
+    
+    return Depends(role_validator) 
diff --git a/backend/app/middleware/auth_middleware.py b/backend/app/middleware/auth_middleware.py
@@ -0,0 +1,88 @@
+import logging
+from typing import List
+import firebase_admin.auth
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.types import ASGIApp
+from app.utilities.constants import LOGGER_NAME
+
+
+class AuthMiddleware(BaseHTTPMiddleware):
+    def __init__(self, app: ASGIApp, public_paths: List[str] = None):
+        super().__init__(app)
+        self.public_paths = public_paths or []
+        self.logger = logging.getLogger(LOGGER_NAME("auth_middleware"))
+
+    def is_public_path(self, path: str) -> bool:
+        return path in self.public_paths
+
+    async def dispatch(self, request: Request, call_next):
+
+        if self.is_public_path(request.url.path):
+            self.logger.info(f"Skipping auth for public path: {request.url.path}")
+            return await call_next(request)
+
+        # Get authentication token from header
+        auth_header = request.headers.get("Authorization")
+        if not auth_header or not auth_header.startswith("Bearer "):
+            self.logger.warning(f"Missing or invalid auth header for {request.url.path}")
+            return JSONResponse(
+                status_code=401,
+                content={"detail": "Authentication required"}
+            )
+
+        token = auth_header.split(" ")[1]
+        try:
+            # Verify the token with Firebase
+            self.logger.info(f"Verifying token for request to {request.url.path}")
+            decoded_token = firebase_admin.auth.verify_id_token(token, check_revoked=True)
+            
+            # Get Firebase user information
+            firebase_user = firebase_admin.auth.get_user(decoded_token["uid"])
+            
+            request.state.user_id = decoded_token["uid"]
+            request.state.user_email = decoded_token.get("email")
+            request.state.email_verified = firebase_user.email_verified
+            request.state.user_claims = decoded_token.get("claims", {})
+            
+            # Add complete user info for convenience
+            request.state.user_info = {
+                "uid": decoded_token["uid"],
+                "email": decoded_token.get("email"),
+                "name": decoded_token.get("name", ""),
+                "picture": decoded_token.get("picture", ""),
+                "email_verified": firebase_user.email_verified
+            }
+            
+            response = await call_next(request)
+            
+            if isinstance(response, Response):
+                response.headers["X-Auth-User-ID"] = decoded_token["uid"]
+            
+            return response
+            
+        except firebase_admin.auth.RevokedIdTokenError:
+            self.logger.warning(f"Token has been revoked: {request.url.path}")
+            return JSONResponse(
+                status_code=401, 
+                content={"detail": "Token has been revoked. Please reauthenticate."}
+            )
+        except firebase_admin.auth.ExpiredIdTokenError:
+            self.logger.warning(f"Token has expired: {request.url.path}")
+            return JSONResponse(
+                status_code=401, 
+                content={"detail": "Token has expired. Please reauthenticate."}
+            )
+        except firebase_admin.auth.InvalidIdTokenError as e:
+            self.logger.warning(f"Invalid token: {request.url.path}, error: {str(e)}")
+            return JSONResponse(
+                status_code=401, 
+                content={"detail": "Invalid authentication token"}
+            )
+        except Exception as e:
+            self.logger.error(f"Authentication error for {request.url.path}: {str(e)}")
+            return JSONResponse(
+                status_code=401, 
+                content={"detail": f"Authentication failed: {str(e)}"}
+            )
diff --git a/backend/app/server.py b/backend/app/server.py
@@ -7,7 +7,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from . import models
-from .middleware import AuthMiddleware
+from .middleware.auth_middleware import AuthMiddleware
 from .routes import auth, send_email, user
 from .utilities.constants import LOGGER_NAME
 from .utilities.firebase_init import initialize_firebase
@@ -73,7 +73,11 @@ def read_item(item_id: int, q: Union[str, None] = None):
 async def test_middleware(request: Request) -> Dict[str, Any]:
     """
     Test endpoint that requires authentication and shows middleware-added state.
-    This will only work if you provide a valid Firebase token.
+    This will only work if you provide a valid Firebase token in the Authorization header.
+    
+    Example: Authorization: Bearer your-firebase-token
+    
+    The response will show all user information added by the Firebase auth middleware.
     """
     # Get all the attributes from request.state
     state_dict = {}
@@ -83,20 +87,24 @@ async def test_middleware(request: Request) -> Dict[str, Any]:
             state_dict[key] = getattr(request.state, key)
 
     return {
-        "message": "Middleware test - Auth required",
+        "message": "Authentication successful! User info from Firebase token:",
         "middleware_state": state_dict,
         "user_id": getattr(request.state, "user_id", None),
-        "request_id": getattr(request.state, "request_id", None),
+        "user_email": getattr(request.state, "user_email", None),
+        "email_verified": getattr(request.state, "email_verified", None),
         "user_claims": getattr(request.state, "user_claims", None),
-        "headers": dict(request.headers)
+        "user_info": getattr(request.state, "user_info", None),
+        "request_id": getattr(request.state, "request_id", None),
+        "authorization_header": request.headers.get("Authorization", "Not provided")
     }
 
 
 @app.get("/test-middleware-public")
 async def test_middleware_public(request: Request) -> Dict[str, Any]:
     """
     Public test endpoint that shows middleware-added state.
-    This should work without authentication as it's in PUBLIC_PATHS.
+    This endpoint is in PUBLIC_PATHS, so it works without authentication.
+    No Firebase token is required to access this endpoint.
     """
     # Get all the attributes from request.state
     state_dict = {}
@@ -105,9 +113,93 @@ async def test_middleware_public(request: Request) -> Dict[str, Any]:
         if not key.startswith("_") and not callable(getattr(request.state, key)):
             state_dict[key] = getattr(request.state, key)
 
+    # Check if any auth header was provided (optional for this endpoint)
+    auth_header = request.headers.get("Authorization")
+    auth_message = "No authentication required for this endpoint"
+    if auth_header:
+        auth_message += " (but you provided an auth header anyway)"
+
     return {
-        "message": "Middleware test - Public",
+        "message": "Public endpoint - No authentication required",
+        "auth_status": auth_message,
         "middleware_state": state_dict,
         "request_id": getattr(request.state, "request_id", None),
-        "headers": dict(request.headers)
+        "authorization_header": request.headers.get("Authorization", "Not provided"),
+    }
+
+
+# Role-based access test endpoints
+from .middleware.auth import has_roles
+from .schemas.user import UserRole
+
+@app.get("/test-role-admin")
+async def test_role_admin(
+    request: Request,
+    authorized: bool = has_roles([UserRole.ADMIN])
+) -> Dict[str, Any]:
+    """
+    Test endpoint that requires the Admin role.
+    
+    This demonstrates role-based access control using the has_roles dependency.
+    Only users with the Admin role can access this endpoint.
+    """
+    return {
+        "message": "You have successfully accessed an admin-only endpoint",
+        "user_id": request.state.user_id,
+        "user_email": request.state.user_email,
+        "role": "admin"
+    }
+
+@app.get("/test-role-volunteer")
+async def test_role_volunteer(
+    request: Request,
+    authorized: bool = has_roles([UserRole.VOLUNTEER])
+) -> Dict[str, Any]:
+    """
+    Test endpoint that requires the Volunteer role.
+    
+    This demonstrates role-based access control using the has_roles dependency.
+    Only users with the Volunteer role can access this endpoint.
+    """
+    return {
+        "message": "You have successfully accessed a volunteer-only endpoint",
+        "user_id": request.state.user_id,
+        "user_email": request.state.user_email,
+        "role": "volunteer"
+    }
+
+@app.get("/test-role-participant")
+async def test_role_participant(
+    request: Request,
+    authorized: bool = has_roles([UserRole.PARTICIPANT])
+) -> Dict[str, Any]:
+    """
+    Test endpoint that requires the Participant role.
+    
+    This demonstrates role-based access control using the has_roles dependency.
+    Only users with the Participant role can access this endpoint.
+    """
+    return {
+        "message": "You have successfully accessed a participant-only endpoint",
+        "user_id": request.state.user_id,
+        "user_email": request.state.user_email,
+        "role": "participant"
+    }
+
+@app.get("/test-role-multiple")
+async def test_role_multiple(
+    request: Request,
+    authorized: bool = has_roles([UserRole.ADMIN, UserRole.VOLUNTEER])
+) -> Dict[str, Any]:
+    """
+    Test endpoint that requires either Admin OR Volunteer role.
+    
+    This demonstrates role-based access control with multiple allowed roles.
+    Users with either Admin or Volunteer roles can access this endpoint.
+    """
+    return {
+        "message": "You have successfully accessed an endpoint requiring admin OR volunteer role",
+        "user_id": request.state.user_id,
+        "user_email": request.state.user_email,
+        "roles_allowed": ["admin", "volunteer"]
     }
