FastAPI Authentication Middleware Example

Authentication Middleware is a crucial component for securing an API by inspecting incoming requests for credentials (like a JWT or API Key) and validating them before the request reaches the route handler. This ensures every protected endpoint is secured automatically and cleanly.

This example demonstrates how to implement a custom authentication middleware to check for a required Authorization header and either inject a valid user/context or reject the request early.

Core Authentication Middleware Structure

Middleware is implemented using the @app.middleware("http") decorator. The function must accept request and call_next. The logic to inspect credentials must run before await call_next(request).

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.authentication import SimpleUser, AuthCredentials, AuthenticationBackend, UnauthenticatedUser
import base64
import json

# --- 1. MOCK BACKEND SERVICE ---
# In a real app, this would query a database or contact an identity service.
class MockUser(SimpleUser):
    """Custom user object storing roles."""
    def __init__(self, username: str, roles: list):
        super().__init__(username)
        self.roles = roles

def decode_token_payload(token: str) -> dict | None:
    """Mock token decoding function."""
    if token == "VALID_ADMIN_TOKEN":
        return {"sub": "admin_user", "roles": ["admin", "editor"]}
    elif token == "VALID_GUEST_TOKEN":
        return {"sub": "guest_user", "roles": ["viewer"]}
    return None

# --- 2. THE CUSTOM AUTHENTICATION BACKEND ---
# This is the standard Starlette pattern for complex, reusable authentication.
class BearerAuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        # 1. Get the Authorization header
        if "Authorization" not in conn.headers:
            return
            
        auth = conn.headers["Authorization"]
        scheme, token = auth.split()
        
        if scheme.lower() != "bearer":
            return
            
        # 2. Decode and Validate the Token
        payload = decode_token_payload(token)
        
        if payload is None:
            # Return credentials and unauthenticated user if token is invalid
            return AuthCredentials(["unauthenticated"]), UnauthenticatedUser()

        # 3. Success: Create user and credentials
        user = MockUser(username=payload['sub'], roles=payload['roles'])
        # Credentials are used by higher-level permission checkers
        credentials = AuthCredentials(payload['roles'])
        
        return credentials, user

# --- 3. APPLICATION SETUP ---
app = FastAPI()

# Apply the Starlette Authentication Middleware using our custom backend
app.add_middleware(AuthenticationMiddleware, backend=BearerAuthBackend())

# Optional: Import necessary functions for route-level access
from starlette.authentication import requires

# Annotation: Using Starlette's AuthenticationMiddleware is the preferred way 
# because it correctly handles thread safety and makes the user object available
# via `request.user` and `request.auth` in the route handler.

---

Protecting Routes and Accessing User Context

Once the middleware is applied, you can protect routes using the @requires decorator or by inspecting the injected request.user object.

Accessing the Authenticated User Object

The authenticated user object (MockUser) is attached directly to the request object.

# The request parameter is needed to access the authenticated user object
@app.get("/user/me")
async def read_current_user(request: Request):
    if request.user.is_authenticated:
        # Access custom attributes attached by the backend
        return {
            "username": request.user.display_name,
            "roles": request.user.roles,
            "is_auth": True
        }
    return {"message": "User is unauthenticated"}

Protecting Routes with Role-Based Access Control (RBAC)

The @requires decorator checks the AuthCredentials injected by the middleware for specific roles (permissions) before allowing the route to execute.

# Route requiring the 'admin' role
@app.get("/admin/metrics")
@requires(scopes=["admin"])
async def get_admin_metrics(request: Request):
    return {"metrics": "System metrics only for admins."}

# Route requiring either 'admin' or 'editor' roles
@app.post("/content/publish")
@requires(scopes=["admin", "editor"])
async def publish_content(request: Request):
    return {"message": f"Content published by {request.user.username}."}

Handling Unauthenticated Access

If the token is missing or invalid, the user will be an UnauthenticatedUser. You can customize the behavior when authentication fails.

async def on_auth_error(conn, exc):
    """
    Custom handler for when the @requires decorator fails.
    """
    return JSONResponse(
        status_code=403, 
        content={"detail": f"Forbidden: {exc.detail}"}
    )

# Override the default error handler (usually added in app setup, shown here separately)
app.add_middleware(
    AuthenticationMiddleware, 
    backend=BearerAuthBackend(), 
    on_error=on_auth_error
)

---

Sources and Further Reading

1. Starlette Documentation - Authentication Middleware
2. FastAPI Documentation - Security and Authentication
3. Starlette Documentation - Authentication Backend
4. FastAPI GitHub Discussion - Custom Auth Middleware Implementation