JWT Blocklist with Redis — Centralized Token Revocation for Distributed Systems

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

A JWT blocklist stored in Redis tracks revoked tokens centrally, allowing distributed API services to reject compromised tokens before their natural expiry.

What You'll Learn

How to implement a Redis-backed JWT blocklist, automatic cleanup with TTL, Pub/Sub for cross-region invalidation, and performance considerations for high-throughput APIs.

Why It Matters

JWTs are stateless — once issued, they cannot be revoked without server-side state. A blocklist adds revocation capability while keeping JWTs otherwise stateless. Redis provides the speed and TTL features needed for production blocklists.

Real-World Use

Auth0 uses a blacklist for immediate token revocation. Stripe invalidates API keys within seconds across all regions. Durga Antivirus Pro maintains a Redis blocklist for partner tokens, ensuring revoked access takes effect within milliseconds.

flowchart TD
    A["API Request\n+ Bearer JWT"] --> B["Auth Middleware"]
    B --> C{"JWT signature\nvalid?"}
    C -->|"No"| D["401 Invalid"]
    C -->|"Yes"| E{"Check Redis\nblocklist"}
    E -->|"Token in blocklist"| D
    E -->|"Not in blocklist"| F["Process request"]
    G["Logout / Revoke"] --> H["Add token jti\n+ exp to Redis"]
    H --> I["Redis TTL auto-cleanup"]
    style F fill:#dcfce7,stroke:#16a34a
    style D fill:#fecaca,stroke:#dc2626
    style G fill:#dbeafe,stroke:#2563eb

Code Example: Redis Blocklist Implementation

import redis, jwt, datetime, os
from flask import Flask, request, jsonify, g

app = Flask(__name__)
SECRET = os.environ.get("JWT_SECRET", "dev-secret")
r = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=6379,
    decode_responses=True
)
BLOCKLIST_PREFIX = "jwt_blocklist:"

def is_token_blocked(jti):
    """Check if a token ID is in the blocklist."""
    return r.exists(f"{BLOCKLIST_PREFIX}{jti}")

def block_token(jti, expiry_timestamp):
    """Add token to blocklist with TTL matching token expiry."""
    ttl = max(0, int(expiry_timestamp - datetime.datetime.utcnow().timestamp()))
    if ttl > 0:
        r.setex(f"{BLOCKLIST_PREFIX}{jti}", ttl, "revoked")

@app.before_request
def authenticate():
    if request.path.startswith("/api/auth/"):
        return

    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return jsonify({"error": "Unauthorized"}), 401

    try:
        token = auth[7:]
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])

        if is_token_blocked(payload.get("jti", "")):
            return jsonify({"error": "Token revoked"}), 401

        g.current_user = payload
    except jwt.ExpiredSignatureError:
        return jsonify({"error": "Token expired"}), 401
    except jwt.InvalidTokenError:
        return jsonify({"error": "Invalid token"}), 401

@app.route("/api/auth/logout", methods=["POST"])
def logout():
    """Revoke the current token."""
    auth = request.headers.get("Authorization", "")
    token = auth[7:]
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])

    block_token(payload["jti"], payload["exp"])
    return jsonify({"message": "Logged out successfully"})

Code Example: Bulk Revocation by User

def revoke_all_user_tokens(user_id):
    """Revoke all tokens for a user by adding to a user-level blocklist."""
    key = f"user_blocklist:{user_id}"
    # Add current timestamp as the revocation point
    r.setex(key, 86400, datetime.datetime.utcnow().timestamp())
    return True

def is_user_blocked(user_id, issued_at):
    """Check if token was issued before user's blocklist timestamp."""
    key = f"user_blocklist:{user_id}"
    blocked_since = r.get(key)
    if blocked_since and issued_at < float(blocked_since):
        return True
    return False

@app.before_request
def check_user_blocklist():
    if not hasattr(g, 'current_user') or not g.current_user:
        return
    if is_user_blocked(g.current_user["sub"], g.current_user.get("iat", 0)):
        return jsonify({
            "error": "Session revoked",
            "message": "Your access has been revoked. Please log in again."
        }), 401

Code Example: Redis Pub/Sub for Cross-Region Invalidation

import threading, json

class CrossRegionBlocklist:
    def __init__(self):
        self.pubsub = r.pubsub()
        self.pubsub.subscribe(**{"jwt-revocation": self.handle_revocation})
        threading.Thread(target=self.pubsub.run_in_thread, daemon=True).start()

    def handle_revocation(self, message):
        """Handle revocation messages from other regions."""
        data = json.loads(message["data"])
        jti = data.get("jti")
        exp = data.get("exp")
        if jti and exp:
            block_token(jti, exp)
            print(f"Cross-region revocation: {jti}")

    @staticmethod
    def publish_revocation(jti, exp):
        """Publish revocation event to all regions."""
        r.publish("jwt-revocation", json.dumps({
            "jti": jti,
            "exp": exp,
            "region": os.environ.get("REGION", "unknown"),
            "timestamp": datetime.datetime.utcnow().isoformat()
        }))

# Usage
@app.route("/api/auth/logout", methods=["POST"])
def logout_cross_region():
    auth = request.headers.get("Authorization", "")
    token = auth[7:]
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])

    # Block locally
    block_token(payload["jti"], payload["exp"])
    # Publish to other regions
    CrossRegionBlocklist.publish_revocation(payload["jti"], payload["exp"])

    return jsonify({"message": "Logged out globally"})

Common Mistakes

1. Blocklist Without TTL

Tokens must eventually expire from the blocklist. Without TTL, the blocklist grows unbounded. Set TTL to match the token's remaining lifetime.

2. Checking Blocklist Before Signature

Always verify the JWT signature before checking the blocklist. An invalid signature should be rejected immediately without a Redis lookup.

3. Single Redis Instance as Single Point of Failure

Use Redis Sentinel or Cluster for high availability. If Redis is down, consider allowing requests through or failing closed — document the trade-off.

4. Blocking by Token Value Instead of jti

The jti (JWT ID) claim is the correct identifier. Blocking by the full token string wastes memory and fails if the token is re-encoded with different whitespace.

5. No Monitoring on Blocklist Size

Monitor the blocklist cardinality and hit rate. A growing blocklist may indicate a revocation storm or a missing TTL configuration.

Practice Questions

1. Why use jti instead of the full token for blocklisting?
2. How does TTL-based cleanup keep the blocklist manageable?
3. What is the purpose of Pub/Sub in cross-region blocklisting?
4. Should the blocklist check happen before or after signature verification?
5. How does a user-level blocklist differ from individual token revocation?

Answers:

1. The jti is a unique, fixed-length identifier. The full token is longer, may vary in encoding, and would waste Redis memory.
2. Each blocklist entry has a TTL matching the token's remaining lifetime. Redis automatically deletes expired keys, keeping memory usage bounded.
3. Pub/Sub broadcasts revocation events to all Redis instances, ensuring blocked tokens are rejected across all regions within milliseconds.
4. After signature verification. Invalid tokens are rejected immediately without a Redis lookup, reducing load on Redis for malicious requests.
5. A user-level blocklist invalidates all tokens issued before a certain timestamp. Individual revocation targets a specific session by jti.

Challenge: Build a JWT authentication system with Redis blocklist that supports individual token revocation, user-level bulk revocation, and cross-region Pub/Sub invalidation.

FAQ

Mini Project

Build a Flask API with Redis-backed JWT blocklist supporting logout, cross-region Pub/Sub invalidation, user-level revocation, and a health-check endpoint that monitors blocklist size and Redis connectivity.

What's Next

Now explore JWT Revocation Claims for embedding revocation metadata directly in token claims.