Refresh Token Rotation

Welcome to the first lesson of our course on securing your REST API application with Python. In this lesson, we will focus on implementing and rotating refresh tokens using FastAPI best practices.

Refresh tokens are long-lived credentials that allow clients to obtain new access tokens without requiring the user to re-authenticate. While access tokens are short-lived (typically minutes) and grant access to protected resources, refresh tokens have a longer lifespan (days or weeks) and are used solely to obtain new access tokens when the current ones expire.

The main purposes of refresh tokens include:

1. Improving User Experience: Users don't need to login repeatedly
2. Enhancing Security: Access tokens can be short-lived, limiting the damage if compromised
3. Maintaining Sessions: Applications can maintain user sessions across restarts or network changes

Advantages:

• Reduces the frequency of user authentication
• Allows for shorter-lived access tokens, improving security
• Enables token revocation without affecting the authentication flow
• Supports proper logout functionality by invalidating tokens

Disadvantages:

• Increases complexity of the authentication system
• Requires server-side storage and management
• Can be vulnerable to theft if not properly secured
• Adds additional API calls to the authentication flow

Refresh token rotation is a critical security practice where each refresh token can only be used once. Let's clarify how it works:

1. What is token rotation? When a client uses a refresh token to get a new access token, they also receive a new refresh token simultaneously, and the old refresh token becomes invalid.

2. The rotation flow:

   • Client sends the current refresh token to get a new access token
   • Server immediately invalidates the sent refresh token
   • Server generates and returns both a new access token AND a new refresh token
   • Client must use the new refresh token for future refreshes

3. Security benefits:

   • Creates a rotating chain of tokens that are each used exactly once
   • If a refresh token is stolen and used by an attacker, the legitimate user's next refresh attempt will fail
   • This failure serves as an early warning system for token theft
   • Limits the "window of opportunity" for attackers to use a stolen token

This approach differs from simple refresh mechanisms where the same refresh token can be used multiple times, which creates a longer vulnerability window if the token is compromised.

By the end of this lesson, you will be able to implement refresh tokens, create authentication tokens, and handle token rotation securely.

The diagram above illustrates the complete flow of token rotation when a client requests new tokens using the /refresh endpoint. Let's walk through each step to understand how our secure token rotation system works in practice:

The Token Rotation Process:

1. Client Initiates Refresh: The client sends a POST request to /refresh with their current refresh token (refreshToken₀)

2. Database Lookup: The API searches the database for the provided refresh token to verify its validity

3. Validation Check: The system evaluates the token status:

   • If the token is missing or expired, the flow immediately returns a 401 Unauthorized error
   • If the token is valid, the rotation process continues

4. Immediate Token Invalidation: Before generating new tokens, the API deletes the used refresh token from the database. This is the critical security step that ensures each token can only be used once

5. New Token Generation: The system creates both:

   • A new short-lived access token (accessToken₁)
   • A new refresh token (refreshToken₁) for future refreshes

6. Persistence: The new refresh token is stored in the database with its expiration date

7. Client Response: Both the new access token and refresh token are returned to the client, completing the rotation cycle

Security Benefits of This Flow:

This sequence demonstrates our single-use token policy in action. Notice how the old token is immediately invalidated (step 4) before new tokens are created. This ensures that:

• Any token can only be used exactly once
• If a token is compromised and used by an attacker, the legitimate user's next refresh attempt will fail
• The system maintains a complete audit trail of token usage

The diagram also shows the alternative flow (step 3) where invalid or expired tokens result in an immediate 401 response, preventing any further processing and maintaining system security.

Before we dive into the implementation, it's crucial to understand proper secret management. In this course, we use a secure approach that works out-of-the-box for learning while teaching production best practices.

Development vs. Production:

• Development: The code includes a fallback secret for convenience, but displays a prominent warning
• Production: You MUST set JWT_SECRET_KEY as an environment variable with a cryptographically secure random value

Our configuration file automatically warns you when using the development fallback:

def get_jwt_secret() -> str:
    """
    Get JWT secret key from environment variables.
    
    Returns a development-only default if not set, but warns the user.
    In production, JWT_SECRET_KEY MUST be set as an environment variable.
    """
    secret = os.getenv('JWT_SECRET_KEY', 'dev-only-secret-CHANGE-IN-PRODUCTION-12345')
    
    if secret == 'dev-only-secret-CHANGE-IN-PRODUCTION-12345':
        print("=" * 80)
        print("⚠️  SECURITY WARNING: Using default JWT secret key!")
        print("=" * 80)
        print("For production, set JWT_SECRET_KEY environment variable:")
        print("  export JWT_SECRET_KEY=$(python -c 'import secrets; print(secrets.token_urlsafe(32))')")
        print("=" * 80)
    
    return secret

To generate a secure secret for production:

python -c 'import secrets; print(secrets.token_urlsafe(32))'

Then set it as an environment variable or in a .env file (never commit this file!):

export JWT_SECRET_KEY=<your-generated-secret>

FastAPI Best Practice: Instead of manually parsing JSON, we use Pydantic models for automatic validation, type safety, and API documentation generation.

from pydantic import BaseModel, Field

class RegisterRequest(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=6)
    role: str = Field(default='user')

class LoginRequest(BaseModel):
    username: str
    password: str

class RefreshRequest(BaseModel):
    refreshToken: str = Field(alias='refreshToken')

class TokenResponse(BaseModel):
    accessToken: str
    refreshToken: str

Why Pydantic Models?

1. Automatic Validation: FastAPI validates all incoming data against your schema
2. Type Safety: Catch errors at development time, not runtime
3. Auto-Generated Docs: Swagger/ReDoc documentation is created automatically
4. Security: Invalid data is rejected before it reaches your business logic
5. Cleaner Code: No more manual .get() calls and null checks

Security Benefits:

Using Pydantic models is itself a security best practice:

• Prevents type confusion attacks
• Validates data format and length
• Provides clear error messages
• Documents expected input formats

Let's implement the RefreshToken model that will store our tokens in the database:

from sqlalchemy import Column, String, Integer, DateTime
from ..database import Base

class RefreshToken(Base):
    """
    Stores each refresh token with a unique token string, user reference,
    and an expiration date (expires_at).
    """
    __tablename__ = 'refresh_tokens'
    
    token = Column(String, primary_key=True)  # The unique identifier (random string or JWT)
    user_id = Column(Integer, nullable=False)  # Link to User model
    expires_at = Column(DateTime, nullable=False)  # When the token expires

Code Logic: This model creates a database table to track refresh tokens. The key design choices here are:

• Using the token itself as the primary key for fast lookups
• Storing a reference to the user who owns the token
• Including an expiration date to automatically invalidate old tokens
• Not storing access tokens in the database since they're self-contained JWTs

In this course, refresh tokens are stored in plaintext for educational clarity. In production, hash refresh tokens before storing them (using SHA-256 or similar) to protect against database compromise. This way, even if someone gains database access, they can't use the stolen tokens.

We keep tokens in plaintext here so you can inspect and debug them while learning the token lifecycle patterns.

Next, we implement the function that generates both access and refresh tokens:

from jose import jwt
from datetime import datetime, timedelta
import uuid
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from ..models.refresh_token import RefreshToken
from ..config import get_jwt_secret

ACCESS_TOKEN_EXPIRES_IN = 15  # minutes
REFRESH_TOKEN_LIFETIME_DAYS = 7

def generate_access_token(user_id: int) -> str:
    """Generates a JWT access token with a unique ID"""
    payload = {
        'userId': user_id,
        'jti': str(uuid.uuid4()),  # Add a unique JWT ID for each token
        'exp': datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRES_IN)
    }
    return jwt.encode(payload, get_jwt_secret(), algorithm='HS256')

async def create_auth_tokens(user_id: int, db: AsyncSession) -> dict:
    """
    Creates a new refresh token for a user, invalidating any previous ones
    """
    access_token = generate_access_token(user_id)
    refresh_token = str(uuid.uuid4())
    
    # Set expiration date for refresh token
    expires_at = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_LIFETIME_DAYS)
    
    # Remove any existing refresh tokens for this user
    result = await db.execute(select(RefreshToken).where(RefreshToken.user_id == user_id))
    existing_tokens = result.scalars().all()
    for token in existing_tokens:
        await db.delete(token)
    
    # Store the new refresh token
    new_token = RefreshToken(
        token=refresh_token,
        user_id=user_id,
        expires_at=expires_at
    )
    db.add(new_token)
    await db.commit()
    
    return {'accessToken': access_token, 'refreshToken': refresh_token}

Code Logic:

• The access token is a JWT with a short lifespan (15 minutes) that contains the user ID and a unique token ID
• We use get_jwt_secret() to retrieve the JWT secret, which reads from environment variables in production
• The refresh token is a simple UUID with a longer lifespan (7 days)
• Before creating a new refresh token, we delete any existing tokens for that user, implementing a "one token per user" policy
• Both tokens are returned to the client, but only the refresh token is stored in the database

This approach ensures that a user can only have one valid refresh token at a time, which simplifies token management and enhances security by making it easier to detect token abuse.

Design Note: We enforce the "one token per user" policy at the application level rather than with a database constraint. This keeps the schema flexible—production systems often allow multiple refresh tokens per user to support different devices or sessions. If your use case requires strict enforcement of a single token, you could add a UniqueConstraint on user_id to the model.

The /refresh endpoint is where token rotation happens. Notice how Pydantic handles validation automatically:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from datetime import datetime, timedelta
from ..schemas.auth import RefreshRequest, TokenResponse
from ..config import get_jwt_secret
import uuid
import logging

logger = logging.getLogger(__name__)
router = APIRouter()

REFRESH_TOKEN_LIFETIME_DAYS = 7

@router.post("/refresh", response_model=TokenResponse)
async def refresh(
    request: RefreshRequest,  # Pydantic validates automatically!
    db: AsyncSession = Depends(get_db)
):
    """REFRESH - Exchanges refresh token for a new access token (ROTATION)"""
    try:
        # request.refreshToken is already validated by Pydantic
        
        # Find the token in the database
        result = await db.execute(
            select(RefreshToken).where(RefreshToken.token == request.refreshToken)
        )
        stored = result.scalar_one_or_none()
        
        if not stored:
            raise HTTPException(status_code=401, detail={'error': 'Invalid refresh token'})
        
        # Check if token has expired
        if stored.expires_at < datetime.utcnow():
            await db.delete(stored)
            await db.commit()
            raise HTTPException(status_code=401, detail={'error': 'Refresh token expired'})
        
        # Begin token rotation process
        access_token = generate_access_token(stored.user_id)
        new_refresh_token = str(uuid.uuid4())  # Generate new refresh token
        
        await db.delete(stored)  # Invalidate the old token immediately
        await db.commit()
        
        # Store the new refresh token
        expires_at = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_LIFETIME_DAYS)
        new_token = RefreshToken(
            token=new_refresh_token,
            user_id=stored.user_id,
            expires_at=expires_at
        )
        db.add(new_token)
        await db.commit()
        
        return TokenResponse(
            accessToken=access_token,
            refreshToken=new_refresh_token
        )
        
    except HTTPException:
        raise
    except Exception as error:
        logger.error(f'Refresh error: {error}')
        raise HTTPException(status_code=500, detail={'error': 'Internal server error'})

Key Improvements with Pydantic:

• No manual .get() calls - Pydantic guarantees the field exists
• Automatic validation - FastAPI rejects invalid requests before they reach our code
• Type safety - request.refreshToken is guaranteed to be a string
• Better errors - Users get clear messages about what's wrong with their request
• API documentation - Swagger/ReDoc shows exactly what fields are required

Key Error Handling Aspects:

• We immediately fail with appropriate error codes if tokens are missing or invalid
• Expired tokens are automatically removed from the database
• Database errors are caught and don't expose internal details to clients
• Each error response includes a specific message to help debug client-side issues

The implementation above teaches the core security pattern effectively. However, in high-traffic production environments, concurrent refresh requests could create a race condition where the same token is used twice simultaneously.

Production Solutions:

• Database row locking (e.g., SELECT FOR UPDATE in PostgreSQL)
• Unique constraints on user_id to enforce one token per user
• Token families with reuse detection for advanced scenarios

For This Course: We focus on the fundamental token rotation pattern, which is the primary defense against token theft. The race condition is an implementation optimization rather than a security flaw. In typical learning scenarios, triggering this edge case is very unlikely.

Note: This course focuses on server-side token security. For production deployments, consider these additional aspects:

Where to Store Tokens Client-Side:

• HttpOnly Cookies (recommended): Protected from XSS, but need CSRF protection
• localStorage: Simple but vulnerable to XSS attacks
• In-memory: Most secure, but lost on page refresh

Supporting Multiple Devices:

Our "one token per user" model simplifies learning, but production systems typically support multiple devices by:

class RefreshToken(Base):
    token = Column(String, primary_key=True)
    user_id = Column(Integer, nullable=False)
    device_id = Column(String)  # Track which device
    device_name = Column(String)  # "iPhone", "Chrome Desktop"
    expires_at = Column(DateTime, nullable=False)

This enables users to stay logged in across phone, tablet, and desktop simultaneously, and revoke individual devices without affecting others. These patterns build on the foundation you're learning in this course.

In this lesson, we covered the implementation and rotation of refresh tokens in a Python-based REST API using FastAPI. You learned what refresh tokens are, their advantages and disadvantages, and how to implement a secure token rotation system.

The key security concepts we've implemented are:

• Short-lived access tokens with longer-lived refresh tokens
• One refresh token per user policy
• Single-use refresh tokens with rotation
• Input validation with Pydantic models (security best practice)
• Proper token validation and error handling
• Secure secret management using environment variables

As you move on to the practice exercises, focus on applying what you've learned to reinforce your understanding. This is just the beginning of your journey in securing your REST API. Keep exploring the course to enhance your skills and protect your API from unauthorized access and abuse. Good luck!