FastAPI Middleware Ordering: Why Your CORS, Auth, and Tenant Context Stack Matters in Production
I spent three days debugging why JWT validation was running after CORS rejection in my CitizenApp staging environment. Three days. The issue? I'd stacked middleware like a junior developer—by convention, not by understanding. The counterintuitive truth: FastAPI executes middleware in reverse order of registration, and in a multi-tenant SaaS, that's a security and operational minefield.
Here's what I've learned the hard way: middleware ordering directly impacts whether your auth context exists when error handlers fire, whether tenant isolation happens before permission checks, and whether sensitive data leaks through error responses. This isn't theoretical. Let me show you the exact pattern I use in CitizenApp now.
The Reverse Execution Model: Read It Backwards
FastAPI's middleware stack works like onion layers. You register middleware top-to-bottom, but execution happens bottom-to-top on request, then top-to-bottom on response. If you don't internalize this, you'll write code that works locally and breaks in production.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
app = FastAPI()
# Registered FIRST, executes LAST on request
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
print("1. LOGGING: Request incoming")
start = time.time()
response = await call_next(request)
duration = time.time() - start
print(f"1. LOGGING: Response took {duration}s")
return response
# Registered SECOND, executes in the MIDDLE
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
print("2. AUTH: Validating JWT")
# Your JWT logic here
request.state.user_id = "user-123"
response = await call_next(request)
print("2. AUTH: Adding security headers")
return response
# Registered THIRD, executes FIRST on request
@app.middleware("http")
async def cors_middleware(request: Request, call_next):
print("3. CORS: Checking origin")
if request.headers.get("origin") not in ["http://localhost:3000"]:
print("3. CORS: Origin rejected!")
return JSONResponse({"error": "CORS"}, status_code=403)
response = await call_next(request)
return response
Request flow: 3 (CORS) → 2 (AUTH) → 1 (LOGGING) → endpoint
Response flow: 1 (LOGGING) → 2 (AUTH) → 3 (CORS)
Notice the problem? CORS runs before auth validation. If you have a strict CORS policy, you're rejecting requests before the JWT context even loads. That's not always wrong, but it means your error handler won't have request.state.user_id. In a multi-tenant app, that's dangerous.
The CitizenApp Pattern: Tenant Context Before Permission Checks
In CitizenApp, I have nine AI features that all need tenant isolation. I learned the hard way that tenant context must load before any permission checks, and CORS should be permissive (or handled in the reverse order) to avoid context loss.
Here's my production stack, in registration order:
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from typing import Optional
import jwt
app = FastAPI()
# Database session (for tenant lookup)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# REGISTERED FIRST (executes LAST on request)
# Error handling wraps everything
@app.middleware("http")
async def error_context_middleware(request: Request, call_next):
"""
This runs AFTER all other middleware on the way in.
It ensures ANY error thrown downstream still has access to tenant context.
"""
try:
response = await call_next(request)
return response
except HTTPException as e:
# Tenant context exists here because it was set by tenant_middleware
tenant_id = getattr(request.state, "tenant_id", "unknown")
# Log with tenant context for debugging multi-tenant issues
print(f"HTTP Error in tenant {tenant_id}: {e.detail}")
raise
except Exception as e:
tenant_id = getattr(request.state, "tenant_id", "unknown")
print(f"Unhandled error in tenant {tenant_id}: {str(e)}")
raise
# REGISTERED SECOND
# Tenant context loads here, before permission checks
@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
"""
Extract tenant from subdomain or header, load tenant context.
This MUST run before auth checks because auth depends on knowing the tenant.
"""
# Get tenant from subdomain (acme.app.local) or X-Tenant-ID header
subdomain = request.headers.get("X-Tenant-Subdomain") or extract_subdomain(request)
if not subdomain:
return JSONResponse({"error": "Tenant not identified"}, status_code=400)
# In production, you'd query the database
tenant_id = resolve_tenant_id(subdomain) # Returns UUID or raises
request.state.tenant_id = tenant_id
request.state.tenant_subdomain = subdomain
response = await call_next(request)
return response
# REGISTERED THIRD
# JWT validation runs here, and it can reference tenant context
@app.middleware("http")
async def jwt_auth_middleware(request: Request, call_next):
"""
Validate JWT. Tenant context already exists.
We can even validate that the JWT's tenant claim matches request tenant.
"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse({"error": "Missing token"}, status_code=401)
token = auth_header.split(" ")[1]
try:
payload = jwt.decode(token, "your-secret", algorithms=["HS256"])
request.state.user_id = payload["sub"]
request.state.user_tenant = payload["tenant_id"]
# CRITICAL: Verify token tenant matches request tenant
if payload["tenant_id"] != request.state.tenant_id:
print(f"TOKEN TENANT MISMATCH: {payload['tenant_id']} != {request.state.tenant_id}")
return JSONResponse({"error": "Tenant mismatch"}, status_code=403)
except jwt.InvalidTokenError as e:
return JSONResponse({"error": f"Invalid token: {str(e)}"}, status_code=401)
response = await call_next(request)
return response
# REGISTERED FOURTH (executes FIRST on request)
# CORS is permissive, lets request through
@app.middleware("http")
async def cors_middleware(request: Request, call_next):
"""
Basic CORS handling. Keep this permissive because stricter validation
happens upstream (tenant + auth). You want requests to reach those layers.
"""
response = await call_next(request)
# Add CORS headers to response
response.headers["Access-Control-Allow-Origin"] = "http://localhost:3000"
response.headers["Access-Control-Allow-Credentials"] = "true"
response.headers["Access-Control-Allow-Methods"] = "GET,POST,PUT,DELETE"
return response
Request execution order: CORS → JWT → Tenant → Error Handler → Endpoint
Why this order?
- CORS first because it's a transport-level concern. Let the request in; actual validation happens next.
- JWT second because we need to know who's making the request.
-
Tenant third and extracted from headers/subdomain. This populates
request.state.tenant_id. - Error handler outermost so it wraps everything and always has context.
Gotcha: Request State Lost in Dependency Injection
Here's what burned me: I assumed request.state would persist through FastAPI's dependency injection system. It doesn't always. When you use Depends(), the dependency function runs in a different scope.
# ❌ This doesn't work
def get_current_user(request: Request):
return request.state.user_id # May be None if middleware hasn't run
@app.get("/me")
async def read_me(user_id: str = Depends(get_current_user)):
return {"user_id": user_id}
# ✅ This works
from fastapi.security import HTTPBearer, HTTPAuthCredentials
security = HTTPBearer()
def get_current_user(credentials: HTTPAuthCredentials = Depends(security)):
token = credentials.credentials
payload = jwt.decode(token, "your-secret", algorithms=["HS256"])
return payload["sub"]
@app.get("/me")
async def read_me(user_id: str = Depends(get_current_user)):
return {"user_id": user_id}
In CitizenApp, I use a hybrid: middleware sets request.state for logging/error context, but endpoints rely on Depends() for actual auth. This prevents state from being lost and keeps dependencies explicit.
What I Missed: Testing Middleware Order
The hardest part isn't writing the middleware—it's testing it. You need to verify that context exists when error handlers fire.
python
from fastapi.testclient import TestClient
client = TestClient(app)
def test_tenant_context_in_error_handler():
"""Verify tenant_id is in request state even when endpoint raises"""
response = client.get(
"/some-endpoint",
headers={"X-Tenant-Subdomain": "acme", "Authorization": "Bearer invalid-token"}
)
# If tenant_middleware runs before jwt_auth_middleware,
# tenant context
