misc stuff

Author: mmiqball

.cursorignore

@@ -0,0 +1,2 @@
+# Firebase credentials
+**/serviceAccountKey.json 

.pdm-python

@@ -0,0 +1 @@
+/Users/mujtaba/Desktop/Blueprint/w25/llsc/.venv/bin/python

backend/.gitignore

@@ -160,3 +160,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# Firebase
+serviceAccountKey.json

backend/app/middleware/__init__.py

@@ -0,0 +1,3 @@
+"""
+Middleware package for LLSC backend.
+""" 

backend/app/middleware/auth_middleware.py

@@ -0,0 +1,129 @@
+from functools import wraps
+from typing import Callable, List, Optional, Set
+import logging
+
+from fastapi import Depends, HTTPException, Security
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy.orm import Session
+
+from app.schemas.user import UserRole
+from app.services.implementations.auth_service import AuthService
+from app.services.implementations.user_service import UserService
+from app.utilities.service_utils import get_auth_service, get_db
+
+security = HTTPBearer()
+
+
+def get_token_from_header(
+    credentials: HTTPAuthorizationCredentials = Security(security),
+) -> str:
+    """Extract token from Authorization header."""
+    print("\n=== Token Extraction ===")
+    print(f"Raw credentials type: {type(credentials)}")
+    print(f"Raw credentials: {credentials}")
+    print(f"Token from header: {credentials.credentials[:50]}...")  # Print first 50 chars
+    print("=== End Token Extraction ===\n")
+    return credentials.credentials
+
+
+def require_auth(
+    auth_service: AuthService = Depends(get_auth_service),
+    token: str = Depends(get_token_from_header),
+) -> None:
+    """Verify that the request has a valid access token."""
+    try:
+        # The token validation is done in the role check, so we just need to check
+        # if the token is valid for any role
+        if not auth_service.is_authorized_by_role(token, {role.value for role in UserRole}):
+            raise HTTPException(
+                status_code=401,
+                detail="Invalid or expired token",
+            )
+    except Exception as e:
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid or expired token",
+        )
+
+
+def require_roles(roles: Set[UserRole]):
+    """Require specific roles to access the endpoint."""
+
+    def decorator(func: Callable) -> Callable:
+        @wraps(func)
+        async def wrapper(
+            *args,
+            db: Session = Depends(get_db),
+            token: str = Depends(get_token_from_header),
+            **kwargs,
+        ):
+            try:
+                print("\n=== Role Check ===")
+                print(f"Checking roles: {roles}")
+                role_values = {role.value for role in roles}
+                print(f"Role values to check: {role_values}")
+                
+                # Create auth service directly
+                logger = logging.getLogger(__name__)
+                auth_service = AuthService(logger=logger, user_service=UserService(db))
+                
+                is_authorized = auth_service.is_authorized_by_role(token, role_values)
+                print(f"Authorization result: {is_authorized}")
+                
+                if not is_authorized:
+                    raise HTTPException(
+                        status_code=403,
+                        detail="Insufficient permissions",
+                    )
+                return await func(*args, **kwargs)
+            except HTTPException as e:
+                print(f"HTTP Exception: {str(e)}")
+                raise e
+            except Exception as e:
+                print(f"Unexpected error: {str(e)}")
+                raise HTTPException(
+                    status_code=401,
+                    detail="Invalid or expired token",
+                )
+
+        return wrapper
+
+    return decorator
+
+
+def require_user_id(user_id_param: str = "user_id"):
+    """Require that the token belongs to the requested user."""
+
+    def decorator(func: Callable) -> Callable:
+        @wraps(func)
+        async def wrapper(
+            *args,
+            auth_service: AuthService = Depends(get_auth_service),
+            token: str = Depends(get_token_from_header),
+            **kwargs,
+        ):
+            try:
+                user_id = kwargs.get(user_id_param)
+                if not user_id:
+                    raise HTTPException(
+                        status_code=400,
+                        detail=f"Missing {user_id_param} parameter",
+                    )
+
+                if not auth_service.is_authorized_by_user_id(token, user_id):
+                    raise HTTPException(
+                        status_code=403,
+                        detail="Not authorized to access this resource",
+                    )
+                return await func(*args, **kwargs)
+            except HTTPException as e:
+                raise e
+            except Exception as e:
+                raise HTTPException(
+                    status_code=401,
+                    detail="Invalid or expired token",
+                )
+
+        return wrapper
+
+    return decorator 

backend/app/routes/auth.py

@@ -1,6 +1,7 @@
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Security, Body
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
-from ..schemas.auth import AuthResponse, LoginRequest, Token
+from ..schemas.auth import AuthResponse, LoginRequest, Token, RefreshRequest
 from ..services.implementations.auth_service import AuthService
 from ..services.implementations.user_service import UserService
 from ..utilities.db_utils import get_db
@@ -9,6 +10,7 @@
 import logging
 
 router = APIRouter(prefix="/auth", tags=["auth"])
+security = HTTPBearer()
 
 @router.post("/login", response_model=AuthResponse)
 async def login(
@@ -24,19 +26,19 @@ async def logout(
     auth_service: AuthService = Depends(get_auth_service)
 ):
     try:
-        auth_service.revoke_tokens(current_user)
+        auth_service.revoke_tokens(current_user["user"])
         return {"message": "Successfully logged out"}
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
 
 @router.post("/refresh", response_model=Token)
 async def refresh(
-    refresh_token: str,
+    refresh_data: RefreshRequest,
     auth_service: AuthService = Depends(get_auth_service)
 ):
     try:
-        return auth_service.renew_token(refresh_token)
+        return auth_service.renew_token(refresh_data.refresh_token)
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
 

backend/app/routes/test_endpoints.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, Depends
+from ..middleware.auth_middleware import require_auth, require_roles, require_user_id
+from ..schemas.user import UserRole
+
+router = APIRouter(prefix="/test", tags=["test"])
+
+# # Basic auth test - any valid token
+# @router.get("/auth")
+# @require_auth
+# async def test_auth():
+#     """Test endpoint requiring just authentication"""
+#     return {"message": "You are authenticated!"}
+
+# Role-based tests
+@router.get("/admin-only")
+@require_roles({UserRole.ADMIN})
+async def test_admin_only():
+    """Test endpoint requiring admin role"""
+    return {"message": "You are an admin!"}
+
+@router.get("/volunteer-or-admin")
+@require_roles({UserRole.VOLUNTEER, UserRole.ADMIN})
+async def test_volunteer_or_admin():
+    """Test endpoint requiring volunteer or admin role"""
+    return {"message": "You are a volunteer or admin!"}
+
+@router.get("/participant-only")
+@require_roles({UserRole.PARTICIPANT})
+async def test_participant_only():
+    """Test endpoint requiring participant role"""
+    return {"message": "You are a participant!"}
+
+# User-specific tests
+@router.get("/users/{user_id}/profile")
+@require_user_id()
+async def test_user_specific(user_id: str):
+    """Test endpoint requiring specific user access"""
+    return {"message": f"You can access user {user_id}'s profile!"}
+
+# Combined tests
+@router.get("/users/{user_id}/admin-action")
+@require_roles({UserRole.ADMIN})
+@require_user_id()
+async def test_admin_user_specific(user_id: str):
+    """Test endpoint requiring both admin role and specific user access"""
+    return {"message": f"You are an admin accessing user {user_id}'s data!"} 

backend/app/schemas/auth.py

@@ -16,6 +16,10 @@ class Token(BaseModel):
 
     model_config = ConfigDict(from_attributes=True)
 
+class RefreshRequest(BaseModel):
+    """Request body for token refresh"""
+    refresh_token: str
+
 class AuthResponse(Token):
     """Authentication response containing tokens and user info"""
     user: UserCreateResponse

backend/app/server.py

@@ -5,7 +5,7 @@
 from dotenv import load_dotenv
 from fastapi import FastAPI
 
-from app.routes import auth, email
+from app.routes import auth, email, test_endpoints
 
 load_dotenv()
 
@@ -32,7 +32,7 @@ async def lifespan(_: FastAPI):
 app.include_router(auth.router)
 app.include_router(user.router)
 app.include_router(email.router)
-
+app.include_router(test_endpoints.router)
 
 @app.get("/")
 def read_root():

backend/app/services/implementations/auth_service.py

@@ -55,11 +55,20 @@ def send_email_verification_link(self, email: str) -> None:
 
     def is_authorized_by_role(self, access_token: str, roles: set[str]) -> bool:
         try:
+            print("\n=== Auth Service Role Check ===")
+            # print(f"Verifying token: {access_token[:50]}...")
             decoded_token = firebase_admin.auth.verify_id_token(access_token, check_revoked=True)
+            print(f"Decoded token UID: {decoded_token.get('uid')}")
             user_role = self.user_service.get_user_role_by_auth_id(decoded_token["uid"])
+            print(f"User role from DB: {user_role}")
+            print(f"Checking against roles: {roles}")
             firebase_user = firebase_admin.auth.get_user(decoded_token["uid"])
-            return firebase_user.email_verified and user_role in roles
-        except:
+            print(f"Email verified: {firebase_user.email_verified}")
+            result = firebase_user.email_verified and user_role in roles
+            print(f"Final authorization result: {result}")
+            return result
+        except Exception as e:
+            print(f"Authorization error: {str(e)}")
             return False
 
     def is_authorized_by_user_id(self, access_token: str, requested_user_id: str) -> bool: