JWT Middleware: Building Auth Middleware in Express, Django & More
Build JWT authentication middleware in Express.js, Django, and FastAPI. Complete guide with code examples and best practices.
In This Article
What Is Authentication Middleware?
Authentication middleware is a layer of code that sits between incoming HTTP requests and your application logic. It intercepts every request, checks for a valid JWT, and either allows the request to proceed or rejects it with a 401 Unauthorized response.
Middleware is the most common pattern for adding JWT authentication to web applications because it centralizes auth logic in one place. Instead of checking tokens in every route handler, you check once in the middleware. For foundational JWT knowledge, see our What is JWT guide.
The Middleware Design Pattern
Most web frameworks follow a similar middleware pattern:
Request → Middleware 1 → Middleware 2 → Route Handler → Response
For JWT authentication, the flow is:
Request → JWT Middleware (verify token) → Authorization Middleware (check permissions) → Route Handler
Good middleware should:
- Extract the token from the request (Authorization header, cookie, etc.)
- Verify the token's signature, expiration, and claims
- Attach user information to the request object
- Handle errors consistently with proper HTTP status codes
- Be stateless and fast
Express.js JWT Middleware
Here's a production-ready JWT middleware for Express.js:
const jwt = require('jsonwebtoken');
const jwksClient = require('jwks-rsa');
// JWKS client for fetching public keys
const client = jwksClient({
jwksUri: process.env.JWKS_URI || 'https://auth.example.com/.well-known/jwks.json',
cache: true,
cacheMaxAge: 600000,
rateLimit: true,
jwksRequestsPerMinute: 10
});
function getSigningKey(header, callback) {
client.getSigningKey(header.kid, (err, key) => {
if (err) return callback(err);
callback(null, key.getPublicKey());
});
}
function jwtAuth(options = {}) {
const {
algorithms = ['RS256'],
issuer,
audience,
getToken,
onVerified
} = options;
return async (req, res, next) => {
try {
// Extract token
let token;
if (getToken) {
token = getToken(req);
} else {
const authHeader = req.headers.authorization;
if (authHeader?.startsWith('Bearer ')) {
token = authHeader.substring(7);
} else if (req.cookies?.token) {
token = req.cookies.token;
}
}
if (!token) {
return res.status(401).json({
error: 'Authentication required',
code: 'NO_TOKEN'
});
}
// Verify token
jwt.verify(token, getSigningKey, {
algorithms,
issuer,
audience,
clockTolerance: 30
}, (err, decoded) => {
if (err) {
if (err.name === 'TokenExpiredError') {
return res.status(401).json({
error: 'Token expired',
code: 'TOKEN_EXPIRED'
});
}
return res.status(401).json({
error: 'Invalid token',
code: 'INVALID_TOKEN'
});
}
// Attach user to request
req.user = decoded;
// Optional callback for additional processing
if (onVerified) onVerified(req, decoded);
next();
});
} catch (err) {
console.error('Auth middleware error:', err);
res.status(500).json({ error: 'Authentication service error' });
}
};
}
// Usage
const app = require('express')();
// Protect all /api routes
app.use('/api', jwtAuth({
issuer: 'https://auth.example.com',
audience: 'https://api.example.com'
}));
// Authorization middleware (layered on top)
function requireRole(...roles) {
return (req, res, next) => {
if (!req.user || !roles.includes(req.user.role)) {
return res.status(403).json({ error: 'Insufficient permissions' });
}
next();
};
}
app.get('/api/admin/users', requireRole('admin'), getUsers);
app.get('/api/profile', requireRole('user', 'admin'), getProfile);
This middleware supports both Bearer tokens and cookie-based tokens, JWKS-based key rotation, and layered authorization checks.
Django JWT Middleware
For Django applications, JWT authentication is typically handled through Django REST Framework. Here's a custom middleware approach:
import jwt
import json
from django.conf import settings
from django.http import JsonResponse
from jwt import PyJWKClient
jwks_client = PyJWKClient(settings.JWKS_URL)
class JWTAuthenticationMiddleware:
EXEMPT_PATHS = ['/api/auth/login', '/api/auth/register', '/health']
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Skip exempt paths
if any(request.path.startswith(p) for p in self.EXEMPT_PATHS):
return self.get_response(request)
# Extract token
auth_header = request.META.get('HTTP_AUTHORIZATION', '')
if not auth_header.startswith('Bearer '):
return JsonResponse(
{'error': 'Authentication required', 'code': 'NO_TOKEN'},
status=401
)
token = auth_header[7:]
try:
# Get signing key from JWKS
signing_key = jwks_client.get_signing_key_from_jwt(token)
# Verify token
payload = jwt.decode(
token,
signing_key.key,
algorithms=['RS256'],
audience=settings.JWT_AUDIENCE,
issuer=settings.JWT_ISSUER,
options={'require': ['exp', 'iss', 'sub']}
)
# Attach user info to request
request.user_id = payload['sub']
request.user_role = payload.get('role', 'user')
request.jwt_payload = payload
except jwt.ExpiredSignatureError:
return JsonResponse(
{'error': 'Token expired', 'code': 'TOKEN_EXPIRED'},
status=401
)
except jwt.InvalidTokenError as e:
return JsonResponse(
{'error': 'Invalid token', 'code': 'INVALID_TOKEN'},
status=401
)
return self.get_response(request)
Add the middleware to your Django settings:
# settings.py
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'myapp.middleware.JWTAuthenticationMiddleware', # Add after security
# ... other middleware
]
JWT_ISSUER = 'https://auth.example.com'
JWT_AUDIENCE = 'https://api.example.com'
JWKS_URL = 'https://auth.example.com/.well-known/jwks.json'
FastAPI JWT Dependency Injection
FastAPI uses dependency injection instead of traditional middleware. Here's how to implement JWT auth with FastAPI:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError, ExpiredSignatureError
import httpx
security = HTTPBearer()
JWKS_URL = "https://auth.example.com/.well-known/jwks.json"
ISSUER = "https://auth.example.com"
AUDIENCE = "https://api.example.com"
# Cache JWKS
jwks_cache = None
async def get_jwks():
global jwks_cache
if jwks_cache is None:
async with httpx.AsyncClient() as client:
resp = await client.get(JWKS_URL)
jwks_cache = resp.json()
return jwks_cache
def get_signing_key(kid: str, jwks: dict) -> str:
for key in jwks['keys']:
if key['kid'] == kid:
from jose.utils import base64url_decode
import json
# Construct RSA public key from JWK
return jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key))
raise HTTPException(status_code=401, detail="Signing key not found")
async def verify_jwt(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
try:
header = jwt.get_unverified_header(token)
jwks = await get_jwks()
public_key = get_signing_key(header.get('kid', ''), jwks)
payload = jwt.decode(
token,
public_key,
algorithms=['RS256'],
audience=AUDIENCE,
issuer=ISSUER,
options={'require': ['exp', 'iss', 'sub']}
)
return payload
except ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token expired"
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token"
)
# Usage
app = FastAPI()
@app.get("/api/profile")
async def get_profile(user: dict = Depends(verify_jwt)):
return {"user_id": user["sub"], "role": user.get("role")}
@app.get("/api/admin")
async def admin_endpoint(user: dict = Depends(verify_jwt)):
if user.get("role") != "admin":
raise HTTPException(status_code=403, detail="Forbidden")
return {"message": "Admin access granted"}
Middleware Error Handling
Consistent error handling is critical for authentication middleware. Always return structured error responses:
// Standardized error format
const AuthErrors = {
NO_TOKEN: { status: 401, code: 'NO_TOKEN', message: 'Authentication required' },
TOKEN_EXPIRED: { status: 401, code: 'TOKEN_EXPIRED', message: 'Token has expired' },
INVALID_TOKEN: { status: 401, code: 'INVALID_TOKEN', message: 'Token is invalid' },
INSUFFICIENT_SCOPE: { status: 403, code: 'INSUFFICIENT_SCOPE', message: 'Insufficient permissions' }
};
function sendAuthError(res, error) {
res.status(error.status).json({
error: error.message,
code: error.code,
// Include WWW-Authenticate header per RFC 6750
headers: { 'WWW-Authenticate': 'Bearer error="invalid_token"' }
});
}
The client can use the error code to decide whether to refresh the token, redirect to login, or show an error message. See our token expiration guide for handling TOKEN_EXPIRED errors gracefully.
Performance Optimization
JWT verification is computationally expensive, especially with asymmetric algorithms. Here are optimization strategies:
1. Cache JWKS Keys
Fetch public keys from the JWKS endpoint periodically rather than on every request. Most JWKS client libraries handle this automatically.
2. Use HS256 for Internal Services
For service-to-service communication within your infrastructure, HS256 is faster than RS256. See our signing algorithms comparison for when to use each.
3. Verify at the Gateway Only
If you use an API Gateway, verify JWTs at the gateway and pass user info via headers to downstream services. This avoids redundant verification.
4. Choose Token Placement Wisely
For web apps, HttpOnly cookies avoid the overhead of parsing Authorization headers and provide XSS protection. See our security best practices for storage recommendations.
Conclusion
JWT authentication middleware centralizes token verification, keeping your route handlers clean and your auth logic consistent. Whether you're using Express.js, Django, or FastAPI, the core pattern is the same: extract the token, verify it, attach user info, and handle errors consistently. Use our JWT decoder tool to inspect your tokens and verify they contain the claims your middleware expects.
Need to decode a JWT token? Try our free JWT decoder tool — no sign-up required, runs entirely in your browser.