JWT Revocation Claims — Embedding Revocation Metadata in Token Payloads

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

app = Flask(__name__)
SECRET = os.environ.get("JWT_SECRET", "dev-secret")

# User token versions — stored in DB/cache
# Increment to revoke all tokens for a user
user_token_versions = {
    "alice": 1,
    "bob": 1
}

def get_current_version(user_id):
    """Fetch current token version from database/Redis."""
    return user_token_versions.get(user_id, 0)

def issue_token(user_id):
    current_version = get_current_version(user_id)
    token = jwt.encode({
        "sub": user_id,
        "token_version": current_version,
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1),
        "jti": os.urandom(16).hex()
    }, SECRET, algorithm="HS256")
    return token

@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:
        payload = jwt.decode(auth[7:], SECRET, algorithms=["HS256"])
        user_id = payload.get("sub")

        # Stateless revocation check
        current_version = get_current_version(user_id)
        if payload.get("token_version", 0) < current_version:
            return jsonify({
                "error": "Token revoked",
                "message": "A newer token version has been issued"
            }), 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/revoke-all", methods=["POST"])
def revoke_all():
    """Revoke all tokens for the current user."""
    user_id = g.current_user.get("sub")
    user_token_versions[user_id] = user_token_versions.get(user_id, 0) + 1
    return jsonify({"message": "All tokens revoked", "new_version": user_token_versions[user_id]})

# Option B: Revoke by issued-at timestamp
user_revocation_times = {}

def revoke_user_sessions(user_id):
    """Revoke all sessions by setting revocation timestamp."""
    user_revocation_times[user_id] = datetime.datetime.utcnow()

def is_token_valid(payload):
    """Check if token was issued before the user's revocation time."""
    user_id = payload.get("sub")
    revoked_at = user_revocation_times.get(user_id)
    if not revoked_at:
        return True
    issued_at = datetime.datetime.utcfromtimestamp(payload.get("iat", 0))
    return issued_at > revoked_at

@app.before_request
def auth_with_iat():
    """Auth middleware using iat-based revocation."""
    if request.path.startswith("/api/auth/"):
        return

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

    try:
        payload = jwt.decode(auth[7:], SECRET, algorithms=["HS256"])
        if not is_token_valid(payload):
            return jsonify({
                "error": "Session revoked",
                "message": "Your session was revoked. Please log in again."
            }), 401
        g.current_user = payload
    except jwt.ExpiredSignatureError:
        return jsonify({"error": "Token expired"}), 401
    except jwt.InvalidTokenError:
        return jsonify({"error": "Invalid token"}), 401

class HybridRevocation:
    """Combines version claims with blocklist for per-token revocation."""

    @staticmethod
    def is_revoked(payload):
        jti = payload.get("jti", "")
        user_id = payload.get("sub", "")

        # Fast path — check version claim
        current_version = get_current_version(user_id)
        if payload.get("token_version", 0) < current_version:
            return True

        # Slow path — check blocklist for individually revoked tokens
        if is_token_blocked(jti):
            return True

        return False

    @staticmethod
    def revoke_all(user_id):
        """Bulk revocation — just increment version."""
        user_token_versions[user_id] = user_token_versions.get(user_id, 0) + 1

    @staticmethod
    def revoke_one(jti, exp):
        """Individual revocation — add to blocklist."""
        block_token(jti, exp)

# Usage in middleware
@app.before_request
def hybrid_auth():
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return

    try:
        payload = jwt.decode(auth[7:], SECRET, algorithms=["HS256"])
        if HybridRevocation.is_revoked(payload):
            return jsonify({"error": "Token revoked"}), 401
        g.current_user = payload
    except jwt.InvalidTokenError:
        return jsonify({"error": "Invalid token"}), 401