Time Of Check Time Of Use in Fastapi

Time Of Check Time Of Use (TOCTOU) vulnerabilities in Fastapi applications typically occur when your code validates a resource's state, then acts on it later without revalidation. This creates a window where an attacker can modify the resource between the check and the use, leading to authorization bypass, data corruption, or business logic exploitation.

In Fastapi, TOCTOU commonly appears in these specific patterns:

Database Record State Changes

@app.get('/orders/{order_id}/cancel')
def cancel_order(order_id: str):
    order = Order.objects.get(id=order_id)
    
    # TOCTOU vulnerability: state checked but not revalidated
    if order.status != 'pending':
        raise HTTPException(status_code=400, detail='Order not cancellable')
    
    # Attacker modifies order.status between check and update
    order.status = 'cancelled'
    order.save()
    return {'message': 'Order cancelled'}

The window between the status check and the update allows an attacker to change the order status via another request, potentially cancelling completed orders or manipulating inventory states.

File-Based TOCTOU

@app.post('/upload-avatar')
def upload_avatar(file: UploadFile = File(...)):
    # TOCTOU: file exists check, then use
    if not Path(f'./avatars/{file.filename}').exists():
        # Window for symlink attack
        with open(f'./avatars/{file.filename}', 'wb') as f:
            content = await file.read()
            f.write(content)
    else:
        raise HTTPException(status_code=400, detail='File exists')

Fastapi's async file handling can exacerbate TOCTOU by introducing additional context switches between operations. An attacker could create a symlink during the window between the existence check and file creation.

Cached Authorization Checks

@app.get('/user-data/{user_id}')
@requires_auth
async def get_user_data(user_id: str, current_user: User = Depends(get_current_user)):
    # TOCTOU: cached permission check
    if not await has_permission(current_user, 'view_user_data'):
        raise HTTPException(status_code=403, detail='Forbidden')
    
    # Attacker changes permissions in another request
    data = await get_user_data_from_db(user_id)
    return data

When Fastapi's dependency injection caches authorization results, the permission check might not reflect real-time changes to user roles or permissions.

Detecting TOCTOU in Fastapi applications requires both static analysis and runtime scanning. middleBrick's Fastapi-specific scanning engine identifies TOCTOU patterns by analyzing your API endpoints and their data flow.

Static Analysis Patterns

middleBrick scans Fastapi route handlers for TOCTOU indicators:

middlebrick scan https://api.example.com --fastapi --output json

{
  "findings": [
    {
      "category": "BOLA/IDOR",
      "severity": "high",
      "description": "TOCTOU vulnerability in order cancellation endpoint",
      "endpoint": "/orders/{order_id}/cancel",
      "code_snippet": "if order.status != 'pending':",
      "remediation": "Use atomic database transactions or revalidate state before update"
    }
  ]
}

Runtime TOCTOU Detection

middleBrick's active scanning tests TOCTOU by:

• Capturing initial resource states through Fastapi's dependency injection
• Modifying resource states in parallel requests
• Observing whether the original request's authorization logic still applies

The scanner specifically targets Fastapi's async/await patterns where context switches create TOCTOU windows:

{
  "fastapi_toctou_tests": [
    {
      "endpoint": "/orders/{order_id}/cancel",
      "test_type": "state_change_attack",
      "vulnerability": "TOCTOU detected - order status changed between check and update",
      "recommendation": "Wrap operations in database transaction"
    }
  ]
}

Fastapi-Specific Checks

middleBrick examines Fastapi's unique features that can introduce TOCTOU:

• Depends() injection caching and its impact on authorization freshness
• async/await context switches in database operations
• Pydantic model validation timing vs. business logic execution
• Fastapi's background tasks and their interaction with main request state

Fastapi provides several native patterns to eliminate TOCTOU vulnerabilities. The key is ensuring atomic operations and revalidation before state changes.

Database Transaction Pattern

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_db_session():
    async with AsyncSession() as session:
        async with session.begin():
            yield session

@app.get('/orders/{order_id}/cancel')
async def cancel_order(order_id: str, session: AsyncSession = Depends(get_db_session)):
    # Single atomic operation - no TOCTOU window
    order = await session.execute(
        select(Order).where(Order.id == order_id, Order.status == 'pending')
    ).scalar_one_or_none()
    
    if not order:
        raise HTTPException(status_code=400, detail='Order not cancellable or not found')
    
    order.status = 'cancelled'
    await session.commit()
    return {'message': 'Order cancelled'}

This pattern combines the check and use in a single database transaction, eliminating the TOCTOU window entirely.

Fastapi Dependency Revalidation

from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

class UserPermissions:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self._permissions = None
    
    async def get_permissions(self):
        if self._permissions is None:
            async with get_db_session() as session:
                self._permissions = await get_user_permissions(session, self.user_id)
        return self._permissions
    
    async def has_permission(self, permission: str):
        permissions = await self.get_permissions()
        return permission in permissions

@app.get('/sensitive-data')
async def sensitive_data(
    current_user: User = Depends(get_current_user),
    permissions: UserPermissions = Depends()
):
    # Revalidate permissions immediately before use
    if not await permissions.has_permission('access_sensitive_data'):
        raise HTTPException(status_code=403, detail='Forbidden')
    
    return await get_sensitive_data()

Fastapi Middleware for TOCTOU Prevention

from fastapi import Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware

class TOCTOUPreventionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Store initial state for TOCTOU detection
        request.state.initial_state = await self.capture_initial_state(request)
        
        response = await call_next(request)
        
        # Verify state hasn't changed in ways that would violate TOCTOU
        if await self.state_changed_invalidly(request):
            return Response(
                status_code=409,
                content={'detail': 'State conflict detected - operation aborted'},
                media_type='application/json'
            )
        
        return response

app.add_middleware(TOCTOUPreventionMiddleware)

Fastapi Pydantic Model Validation

from pydantic import BaseModel, Field
from typing import Literal

class OrderUpdate(BaseModel):
    status: Literal['pending', 'processing', 'shipped', 'cancelled'] = Field(default=None)
    
    @validator('status')
    def validate_status_change(cls, v, values, **kwargs):
        # Validate that status change is allowed atomically
        current_status = values.get('current_status')
        if current_status != 'pending' and v == 'cancelled':
            raise ValueError('Can only cancel pending orders')
        return v

@app.patch('/orders/{order_id}')
async def update_order(
    order_id: str,
    update_data: OrderUpdate,
    session: AsyncSession = Depends(get_db_session)
):
    order = await session.execute(
        select(Order).where(Order.id == order_id)
    ).scalar_one_or_none()
    
    # Validate with current state - atomic check
    update_data = OrderUpdate(
        **update_data.dict(),
        current_status=order.status
    )
    
    order.status = update_data.status
    await session.commit()
    return {'message': 'Order updated'}