Token Blacklisting and Revocation

Welcome to the Blacklisting and Forced Revocation in Python REST API. In this lesson, we will focus on token blacklisting and forced revocation, two critical techniques for maintaining the security of your API. We'll use Pydantic models throughout to ensure type safety and proper API documentation.

Token blacklisting is a security mechanism where you maintain a list of tokens that are no longer considered valid, despite not being expired. When a token is added to this blacklist, the system will reject it even if it contains a valid signature and hasn't reached its expiration date.

Forced revocation is the act of deliberately invalidating a token before its natural expiration. This is typically done through blacklisting and is essential for scenarios where you need to immediately terminate a user's session.

There are several approaches to invalidating tokens:

1. Blacklisting - Adding tokens to a "deny list" that is checked during validation
2. Token Removal - Simply removing tokens from the database if you're storing them
3. Direct Invalidation - Setting a "valid" flag to false in your token storage

Each approach has its trade-offs. Token removal is simple but only works if you store tokens in the first place. Direct invalidation requires modifying your token storage schema. Blacklisting works universally but requires an additional database structure.

For this course, we'll focus on the blacklisting approach as it provides a robust solution that works well with JWT-based authentication without requiring us to store valid tokens.

These techniques are particularly useful in several scenarios:

• When a user logs out and you want to invalidate their token immediately
• When a security breach is detected and you need to invalidate potentially compromised tokens
• When a user's account is deactivated or suspended
• When you need to enforce access policy changes immediately

Pros:

• Provides immediate security response capabilities
• Allows for centralized security control
• Creates an audit trail of revocation events
• Enhances user account security

Cons:

• Introduces database dependency for token validation
• Adds latency to request processing
• Requires storage and management of blacklisted tokens
• Creates potential performance bottlenecks at scale

FastAPI Best Practice: We use Pydantic models for all API requests and responses to ensure validation, type safety, and automatic documentation:

from pydantic import BaseModel, Field
from typing import Optional

class RevokeTokenRequest(BaseModel):
    """Request model for admin token revocation"""
    tokenToRevoke: str = Field(alias='tokenToRevoke')
    reason: Optional[str] = None
    
    class Config:
        populate_by_name = True

class RevokeTokenResponse(BaseModel):
    """Response model for successful token revocation"""
    message: str
    token: str  # Partial token for confirmation

class LogoutResponse(BaseModel):
    """Response model for logout endpoint"""
    message: str

The TokenBlacklist model is the foundation of our implementation. It stores invalidated tokens and provides a way to check if a token has been revoked.

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

class TokenBlacklist(Base):
    """
    Stores blacklisted tokens to prevent their use
    """
    __tablename__ = 'token_blacklist'
    
    token = Column(String(1024), primary_key=True)  # Tokens can be long, especially JWTs
    reason = Column(String, nullable=True)  # Optional reason for blacklisting
    expires_at = Column(DateTime, nullable=False)  # When this blacklist entry should expire

Key Concepts:

• We store the complete token as a unique identifier, allowing quick lookups
• We include a reason field to track why a token was blacklisted (useful for auditing)
• The expires_at field is critical for database maintenance - it matches the token's expiration so we can eventually clean up the blacklist

This model is optimized for quick lookups, which is essential as every authenticated request will check against this table.

The revoke_access_token function handles the core functionality of adding tokens to the blacklist:

async def revoke_access_token(token: str, db: AsyncSession, reason: str = 'Admin revocation'):
    """
    Add a token to the blacklist database
    
    SECURITY: Verifies token signature before trusting expiration claim.
    This prevents DoS attacks where attackers submit forged tokens with
    distant expiration dates to bloat the blacklist database.
    """
    print(f"Revoking token: {token[:20]}...")
    
    try:
        # CRITICAL: Verify signature first - reject forged tokens
        decoded = jwt.decode(token, get_jwt_secret(), algorithms=['HS256'])
        expires_at = datetime.utcfromtimestamp(decoded['exp'])
    except jwt.JWTError:
        # Invalid/forged token - use conservative 24h default to prevent DoS
        # This ensures attackers can't bloat our database with fake far-future tokens
        expires_at = datetime.utcnow() + timedelta(hours=24)
    
    # Create blacklist entry
    blacklist_entry = TokenBlacklist(
        token=token,
        reason=reason,
        expires_at=expires_at
    )
    
    db.add(blacklist_entry)
    await db.commit()
    await db.refresh(blacklist_entry)
    
    return blacklist_entry

Key Security Logic:

• We always verify the token signature using get_jwt_secret() before trusting its claims
• If the token is invalid or forged, we use a conservative 24-hour expiration
• This prevents attackers from injecting fake tokens with expiration dates 100 years in the future
• Without signature verification, an attacker could perform a DoS attack by bloating the database with millions of entries that never expire
• Valid tokens use their actual expiration time for automatic cleanup

This approach ensures that the blacklist grows and shrinks naturally with token lifetimes, preventing it from becoming too large over time while protecting against malicious abuse.

The authentication middleware is where token validation and blacklist checking come together:

async def get_current_user(authorization: str, db: AsyncSession):
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing authorization header")
    
    token = authorization.split(' ')[1] if authorization.startswith('Bearer ') else authorization
    
    # First check if token is blacklisted
    if await check_token_blacklist(token, db):
        raise HTTPException(
            status_code=403, 
            detail={
                'error': 'Access token has been revoked',
                'reason': 'Token revoked'
            }
        )
    
    try:
        # If not blacklisted, verify token
        decoded = jwt.decode(token, get_jwt_secret(), algorithms=['HS256'])
        user_id = decoded['userId']
        
        result = await db.execute(select(User).where(User.id == user_id))
        user = result.scalar_one_or_none()
        
        if not user:
            raise HTTPException(status_code=401, detail="User not found")
        
        return user
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

async def check_token_blacklist(token: str, db: AsyncSession) -> bool:
    """
    Check if a token is blacklisted
    """
    result = await db.execute(
        select(TokenBlacklist).where(TokenBlacklist.token == token)
    )
    blacklisted = result.scalar_one_or_none()
    
    if blacklisted:
        print(f"Blocked blacklisted token: {token[:20]}...")
        return True
    
    return False

Core Functionality:

• Before verifying the token itself, we check if it exists in the blacklist
• If the token is blacklisted, we immediately reject the request with a 403 status
• The order of operations is important: we check the blacklist first, then verify the token
• We use get_jwt_secret() to verify the token signature
• We include the revocation reason in the response to provide context to clients

This middleware creates a decision point where blacklisted tokens are filtered out before any protected resources are accessed.

To provide administrative control, we implement a dedicated endpoint for token revocation using Pydantic models:

@router.post("/revoke", response_model=RevokeTokenResponse)
async def revoke_token(
    request: RevokeTokenRequest,
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db)
):
    """
    Admin endpoint to forcibly revoke an access token
    
    Accepts a RevokeTokenRequest and returns a RevokeTokenResponse
    """
    try:
        await verify_admin(authorization, db)
        
        # request.tokenToRevoke is already validated by Pydantic!
        if not request.tokenToRevoke:
            raise HTTPException(status_code=400, detail={'error': 'Must provide tokenToRevoke'})

        # Revoke token - database implementation
        blacklist_entry = await revoke_access_token(
            request.tokenToRevoke, 
            db,
            request.reason or 'Admin revocation'
        )
        
        return RevokeTokenResponse(
            message='Token revoked successfully.',
            token=request.tokenToRevoke[:20] + '...'  # Show part of token for confirmation
        )
    except HTTPException:
        raise
    except Exception as error:
        print(f'Error revoking token: {error}')
        raise HTTPException(status_code=500, detail={'error': 'Internal server error'})

Key Improvements with Pydantic:

• No manual parsing - Pydantic handles all validation
• Type safety - request.tokenToRevoke is guaranteed to be a string
• Optional fields - request.reason has proper type handling
• Structured response - RevokeTokenResponse ensures consistent output
• API documentation - Swagger shows exact request/response format

This admin-controlled revocation creates a powerful security tool that can be used in response to security events or as part of normal administrative operations.

A clean logout implementation using Pydantic models:

@router.post("/logout", response_model=LogoutResponse)
async def logout(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db)
):
    """
    Logout endpoint - blacklist the current token
    
    Returns a LogoutResponse with confirmation message
    """
    try:
        if not authorization:
            raise HTTPException(status_code=400, detail='No token provided')
        
        token = authorization.split(' ')[1] if authorization.startswith('Bearer ') else authorization
        
        # Add the token to the database blacklist
        await revoke_access_token(token, db, 'User logout')
        
        return LogoutResponse(message='Logged out successfully')
    except HTTPException:
        raise
    except Exception as error:
        print(f"Error during logout: {error}")
        raise HTTPException(status_code=500, detail='Internal server error')

Benefits:

• Clean, self-documenting endpoint
• Type-safe response guaranteed by Pydantic
• Automatic API documentation
• Consistent response format

In this lesson, we've explored token blacklisting and forced revocation in depth. We've seen how these techniques provide critical security controls for your API by allowing you to invalidate tokens before their natural expiration.

The implementation we've covered creates a robust framework for token security that balances:

• Security requirements (immediate revocation with DoS protection)
• Performance considerations (optimized lookups)
• Database management (automatic cleanup via expiration)
• Type safety with Pydantic models (validation and documentation)

Key Security Takeaway: Always verify token signatures before trusting their claims. This prevents attackers from exploiting your security mechanisms to perform DoS attacks.

Key Development Takeaway: Using Pydantic models for all API interactions ensures type safety, provides automatic validation, and generates comprehensive API documentation.

Now, it's time to apply what you've learned in the practice exercises. These exercises will reinforce your understanding and help you master the token blacklisting and revocation techniques covered in this lesson.