from collections.abc import Callable from fastapi import Request, status from fastapi.responses import JSONResponse from jose import JWTError, jwt from app.core.config import settings def jwt_middleware( excluded_paths: list[str] | None = None, ) -> Callable: """Middleware to check for a valid JWT token in protected routes. Args: excluded_paths: List of paths to exclude from JWT verification Returns: Middleware function """ async def middleware(request: Request, call_next): # Default excluded paths - public API endpoints and health check if excluded_paths is None: _excluded_paths = [ f"{settings.API_V1_STR}/auth/signup", f"{settings.API_V1_STR}/auth/login", "/health", "/docs", "/redoc", "/openapi.json", ] else: _excluded_paths = excluded_paths # Skip JWT verification for excluded paths if any(request.url.path.startswith(path) for path in _excluded_paths): return await call_next(request) # Get token from Authorization header auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"detail": "Not authenticated"}, headers={"WWW-Authenticate": "Bearer"}, ) token = auth_header.split(' ')[1] try: # Verify token payload = jwt.decode( token, settings.SECRET_KEY, algorithms=["HS256"] ) # Attach user_id to request state for future use request.state.user_id = int(payload.get("sub")) except JWTError: return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"detail": "Invalid authentication credentials"}, headers={"WWW-Authenticate": "Bearer"}, ) # Continue processing the request return await call_next(request) return middleware