FastAPI Auth: JWT Tokens — Stateless Authentication

Your application has 50,000 concurrent users and three backend instances behind a load balancer. With session-based auth, every instance needs access to the session store — one more network hop per request, one more point of failure. If Redis goes down, every user is logged out simultaneously.

JSON Web Tokens solve this by moving the session data into the token itself. The server does not need to look anything up — it validates the token’s signature and reads the claims. No shared state, no session store, no extra network hop.

But that statelesness comes with a cost: you cannot easily revoke a JWT. This post builds a complete JWT authentication system in FastAPI, including access tokens, refresh tokens, rotation, and the strategies for handling revocation.

JWT Anatomy

A JWT is three Base64url-encoded JSON objects separated by dots:

eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U

Critical misunderstanding: JWTs are signed, not encrypted. Anyone can Base64-decode the payload and read the claims. Never put sensitive data (passwords, credit card numbers, internal IDs you want to hide) in a JWT payload.

Signing Algorithms: HMAC vs. RSA

Use HS256 when only your backend signs and verifies tokens. It is faster and simpler.

Use RS256 when other services need to verify tokens without access to the signing key. The auth service holds the private key; all other services use the public key. A compromised service cannot forge tokens.

Building JWT Auth in FastAPI

Token Creation

from datetime import datetime, timedelta, timezone
from uuid import uuid4

import jwt

from src.core.config import settings

ACCESS_TOKEN_EXPIRE = timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = timedelta(days=7)


def create_access_token(user_id: str, roles: list[str] | None = None) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "type": "access",
        "roles": roles or [],
        "iat": now,
        "exp": now + ACCESS_TOKEN_EXPIRE,
        "jti": str(uuid4()),  # Unique token ID — needed for revocation
    }
    return jwt.encode(payload, settings.jwt_secret_key, algorithm="HS256")


def create_refresh_token(user_id: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "type": "refresh",
        "iat": now,
        "exp": now + REFRESH_TOKEN_EXPIRE,
        "jti": str(uuid4()),
    }
    return jwt.encode(payload, settings.jwt_secret_key, algorithm="HS256")


def decode_token(token: str) -> dict:
    """Decode and validate a JWT. Raises jwt.InvalidTokenError on failure."""
    return jwt.decode(
        token,
        settings.jwt_secret_key,
        algorithms=["HS256"],  # ALWAYS specify algorithms explicitly
        options={"require": ["sub", "type", "exp", "jti"]},
    )

Why algorithms=["HS256"] is mandatory: Without an explicit algorithm list, an attacker can send a token with "alg": "none" — a valid JWT header that tells the library to skip signature verification. This is the most well-known JWT vulnerability.

The Access/Refresh Token Pattern

A single long-lived token is a security risk — if stolen, the attacker has access for weeks. A single short-lived token is a UX nightmare — the user re-authenticates every 15 minutes.

The solution: two tokens with different lifetimes and scopes.

The access token is short-lived and widely sent. If intercepted, it expires quickly. The refresh token is long-lived but narrowly sent — it only goes to one endpoint, reducing the attack surface.

Login Endpoint (JWT Version)

from fastapi import APIRouter, Depends, HTTPException, Response, status
from fastapi.security import OAuth2PasswordRequestForm

from src.core.jwt import create_access_token, create_refresh_token
from src.core.security import verify_password

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/token")
async def login(
    response: Response,
    form: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
):
    """OAuth2-compatible login endpoint.

    Returns access_token in the response body (for the client to store in memory)
    and refresh_token in an HttpOnly cookie (invisible to JavaScript).
    """
    user = await authenticate_user(db, form.username, form.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = create_access_token(user_id=str(user.id), roles=user.roles)
    refresh_token = create_refresh_token(user_id=str(user.id))

    # Store refresh token in HttpOnly cookie
    response.set_cookie(
        key="refresh_token",
        value=refresh_token,
        httponly=True,
        secure=True,
        samesite="strict",        # Refresh only happens same-site
        max_age=7 * 24 * 3600,
        path="/auth/refresh",     # Only sent to the refresh endpoint
    )

    return {
        "access_token": access_token,
        "token_type": "bearer",
    }

Notice path="/auth/refresh" — the refresh token cookie is only sent when the browser requests /auth/refresh. It is never sent to your API endpoints, so even if an XSS script makes API calls, the refresh token is not included.

Token Validation Dependency

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

from src.core.jwt import decode_token
from src.models.user import User

import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    try:
        payload = decode_token(token)
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if payload.get("type") != "access":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token type",
        )

    user = await db.get(User, payload["sub"])
    if not user or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found or inactive",
        )

    return user

Why check payload["type"]? Without it, an attacker who captures a refresh token can use it as an access token. The type claim prevents token confusion attacks.

Token Refresh and Rotation

@router.post("/refresh")
async def refresh(
    response: Response,
    refresh_token: str = Cookie(None, alias="refresh_token"),
    db: AsyncSession = Depends(get_db),
    redis_client: redis.Redis = Depends(get_redis),
):
    if not refresh_token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No refresh token")

    try:
        payload = decode_token(refresh_token)
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token")

    if payload.get("type") != "refresh":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type")

    # Check if this refresh token has been used before (rotation detection)
    jti = payload["jti"]
    if await redis_client.get(f"used_refresh:{jti}"):
        # This token was already used — possible token theft!
        # Invalidate ALL refresh tokens for this user
        await redis_client.set(f"user_revoked:{payload['sub']}", "1", ex=7 * 24 * 3600)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token reuse detected. All sessions revoked.",
        )

    # Mark this refresh token as used
    await redis_client.set(f"used_refresh:{jti}", "1", ex=7 * 24 * 3600)

    # Check user-level revocation
    if await redis_client.get(f"user_revoked:{payload['sub']}"):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="All sessions revoked")

    user = await db.get(User, payload["sub"])
    if not user or not user.is_active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")

    # Issue new token pair
    new_access = create_access_token(user_id=str(user.id), roles=user.roles)
    new_refresh = create_refresh_token(user_id=str(user.id))

    response.set_cookie(
        key="refresh_token",
        value=new_refresh,
        httponly=True,
        secure=True,
        samesite="strict",
        max_age=7 * 24 * 3600,
        path="/auth/refresh",
    )

    return {"access_token": new_access, "token_type": "bearer"}

Refresh Token Rotation Explained

Every time a refresh token is used, a new refresh token is issued and the old one is invalidated. If an attacker steals and uses a refresh token, one of two things happens:

1. Attacker uses it first: The legitimate user’s next refresh fails (old token already used). The user re-authenticates.
2. User uses it first: The attacker’s refresh attempt triggers reuse detection. All sessions for that user are revoked.

The Revocation Problem

The biggest criticism of JWTs: you cannot revoke an access token. It is valid until it expires. If a user logs out, changes their password, or gets banned — the old access token still works for up to 15 minutes.

Here are the three strategies, ordered by complexity:

Strategy 1: Short Expiry (Simplest)

Set access tokens to 5-15 minutes. Accept the gap. Most applications can tolerate a brief window where a revoked user still has access.

Tradeoff: More frequent refresh calls. The client must handle 401 responses and automatically refresh.

Strategy 2: Blocklist (Moderate)

Maintain a Redis set of revoked token IDs (jti). Check every request. Entries expire when the token would have expired naturally.

async def is_token_revoked(jti: str, redis_client: redis.Redis) -> bool:
    return await redis_client.exists(f"revoked:{jti}")

async def revoke_token(jti: str, ttl: int, redis_client: redis.Redis) -> None:
    await redis_client.set(f"revoked:{jti}", "1", ex=ttl)

Tradeoff: You are back to checking a central store on every request — partly defeating the purpose of stateless JWTs. But the blocklist is small (only revoked tokens, not all tokens) and entries auto-expire.

Strategy 3: Versioned Tokens (Advanced)

Store a token_version on the user record. Include it in every JWT. When you need to revoke all tokens, increment the version. On validation, compare the token’s version with the database.

# In the JWT payload
{"sub": "user_id", "ver": 3, ...}

# On validation
if payload["ver"] != user.token_version:
    raise HTTPException(status_code=401, detail="Token revoked")

Tradeoff: Requires a database read per request (to check the version), but only for the user table — no separate revocation store.

Session Cookies vs. JWT: The Honest Comparison

The honest answer: Use session cookies for server-rendered web apps. Use JWTs for SPAs, mobile apps, and microservice architectures. Plenty of production systems use both — session cookies for the web frontend, JWTs for API-to-API communication.

What’s Next

In Part 4, we move from human users to machine clients. API keys are the simplest way to authenticate programmatic access, but they come with their own security model around scoping, rotation, and rate limiting.