JWT Middleware — Reusable JWT Validation for API Frameworks

In this tutorial, you will learn about JWT Middleware. We cover key concepts, practical examples, and best practices to help you master this topic.

JWT middleware intercepts requests, extracts and validates the JWT from the Authorization header, and either attaches the decoded payload to the request context or returns a 401 error.

What You'll Learn

How to build JWT middleware for different frameworks, handle all JWT error types, support multiple signing keys, and optimize performance with caching.

Why It Matters

Centralizing JWT validation in middleware ensures every protected endpoint has consistent authentication. Without middleware, each endpoint repeats validation logic, leading to inconsistencies and missing checks.

Real-World Use

Express.js passport-jwt, Flask-PyJWT, and FastAPI's HTTPBearer are all middleware components that handle JWT validation. Their proper configuration determines the security posture of the entire API.

flowchart LR
    A["Incoming Request"] --> B["JWT Middleware"]
    B --> C{"Valid JWT?"}
    C -->|"Yes"| D["Attach user to\nrequest context"]
    D --> E["Route Handler"]
    C -->|"Expired"| F["401 + token_expired"]
    C -->|"Invalid"| G["401 + invalid_token"]
    C -->|"Missing"| H["401 + missing_token"]
    style A fill:#dbeafe,stroke:#2563eb
    style B fill:#fef3c7,stroke:#d97706
    style E fill:#dcfce7,stroke:#16a34a
    style F fill:#fecaca,stroke:#dc2626
    style G fill:#fecaca,stroke:#dc2626
    style H fill:#fecaca,stroke:#dc2626

Code Example: Flask JWT Middleware

from flask import Flask, request, jsonify, g
from functools import wraps
import jwt
import os

app = Flask(__name__)
SECRET = os.environ.get("JWT_SECRET", "change-me")
ISSUER = "https://auth.dodatech.com"

def jwt_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth = request.headers.get("Authorization", "")

        if not auth.startswith("Bearer "):
            return jsonify({
                "error": "missing_token",
                "message": "Authorization header must use Bearer scheme"
            }), 401

        token = auth[7:]

        try:
            payload = jwt.decode(
                token,
                SECRET,
                algorithms=["HS256"],
                options={"require": ["sub", "exp", "iss"]},
                issuer=ISSUER
            )
            g.current_user = payload
        except jwt.ExpiredSignatureError:
            return jsonify({
                "error": "token_expired",
                "message": "Token has expired. Please refresh."
            }), 401
        except jwt.MissingRequiredClaimError as e:
            return jsonify({
                "error": "missing_claim",
                "message": f"Missing required claim: {e}"
            }), 401
        except jwt.InvalidIssuerError:
            return jsonify({
                "error": "invalid_issuer",
                "message": "Token issuer not trusted"
            }), 401
        except jwt.InvalidTokenError:
            return jsonify({
                "error": "invalid_token",
                "message": "Token is malformed or signature invalid"
            }), 401

        return f(*args, **kwargs)
    return decorated

# Usage
@app.route("/api/protected")
@jwt_required
def protected():
    return jsonify({
        "message": "You are authenticated",
        "user": g.current_user["sub"],
        "role": g.current_user.get("role")
    })

Code Example: JWT Middleware with JWKS

from jwt import PyJWKClient

# Initialize JWKS client (caches keys)
jwks_url = "https://auth.dodatech.com/.well-known/jwks.json"
jwks_client = PyJWKClient(jwks_url, cache_keys=True)

def jwt_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return jsonify({"error": "Missing token"}), 401

        token = auth[7:]

        try:
            # Get signing key from JWKS (kid in header selects the key)
            signing_key = jwks_client.get_signing_key_from_jwt(token)
            payload = jwt.decode(
                token,
                signing_key.key,
                algorithms=["RS256"],
                audience="https://api.dodatech.com"
            )
            g.current_user = payload
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token expired"}), 401
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid token"}), 401

        return f(*args, **kwargs)
    return decorated

Common Mistakes

1. Returning Generic Error Messages

Return specific error codes (token_expired, invalid_signature, missing_claim) so clients can respond appropriately.

2. Not Caching JWKS Responses

Fetching the JWKS on every request adds latency and load. Cache with a reasonable TTL.

3. Leaking Token Data in Logs

Never log the full token. Log truncated values or the jti only. Sanitize the Authorization header before logging.

4. Not Handling All JWT Error Types

Catching jwt.InvalidTokenError catches everything, but specific errors (expired, missing issuer) deserve specific responses.

5. Applying Middleware to All Routes

Public routes (login, health, Webhooks) should bypass JWT middleware. Use conditional middleware or per-route decorators.

Practice Questions

1. What is the purpose of JWT middleware?
2. How does middleware attach user data to the request?
3. Why should middleware return specific error codes?
4. How does JWKS-based middleware differ from secret-based?
5. How do you exclude public routes from JWT middleware?

Answers:

1. JWT middleware centralizes token extraction, validation, and user context attachment so individual route handlers do not repeat this logic.
2. Flask uses g, Express uses req.user, FastAPI uses Dependency Injection. The middleware sets these before calling the route handler.
3. Specific error codes (token_expired, invalid_issuer) let the client handle each case appropriately (silent refresh, re-login, etc.).
4. JWKS middleware fetches the signing key dynamically based on the kid header, supporting key rotation without configuration changes.
5. Use decorators only on protected routes, or check the request path in the middleware and skip validation for public paths.

Challenge: Build a JWT middleware that supports both HS256 (secret) and RS256 (JWKS) tokens, detects the algorithm from the header, and validates accordingly.

FAQ

Mini Project

Build a Flask JWT middleware that supports HS256 and RS256, handles all JWT error types with specific error codes and messages, attaches user context to Flask's g object, and includes a helper to require specific roles.

What's Next

Now learn about JWT Revocation — comprehensive strategies for invalidating tokens before expiry.