Denial Of Service in Fastapi

Denial of Service (DoS) attacks in Fastapi applications exploit the framework's asynchronous nature and Python's Global Interpreter Lock (GIL) to exhaust server resources. Fastapi's default behavior can leave applications vulnerable to several specific attack vectors.

The most common Fastapi DoS pattern involves synchronous blocking operations within async endpoints. When you write an async endpoint but perform blocking I/O operations without proper handling, you tie up the event loop:

from fastapi import FastAPI
import time
import requests

app = FastAPI()

@app.get("/slow-endpoint")
async def slow_endpoint():
    # Blocking call blocks the entire event loop
    time.sleep(5)  # This is the problem!
    return {"message": "Done"}

During those 5 seconds, Fastapi cannot process other requests, creating a denial of service scenario even with minimal traffic. The GIL prevents Python from truly parallelizing CPU-bound tasks in async code.

Another Fastapi-specific vulnerability is improper file upload handling. Fastapi's default upload limit is 100MB, but attackers can exploit this by sending files that trigger memory exhaustion:

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # No size validation - attacker can upload massive files
    content = await file.read()
    return {"size": len(content)}

Without size limits or streaming, a single request can consume all available memory. Fastapi's Starlette foundation uses in-memory storage for uploads by default, making this particularly dangerous.

Recursive endpoint calls through Fastapi's dependency injection system can also create DoS conditions. Consider this vulnerable pattern:

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

async def get_user_data(user_id: str = Depends(get_user_id)):
    # No depth limit on recursive calls
    return await get_user_data_from_db(user_id)

@app.get("/user")
async def read_user(data: dict = Depends(get_user_data)):
    return data

If dependencies create circular references or unbounded recursion, attackers can trigger stack overflows or infinite loops that crash the application.

Rate limiting gaps in Fastapi's async endpoints create another DoS vector. Without proper rate limiting middleware, attackers can overwhelm your application with requests that each consume database connections or external API calls:

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/expensive")
async def expensive_operation():
    # No rate limiting - can be called thousands of times per second
    async with httpx.AsyncClient() as client:
        response = await client.get("https://external-service.com/api")
    return response.json()

Each request creates a new HTTP client and makes an external call, potentially exhausting connection pools or triggering rate limits on third-party services, which then affects your application's availability.

Detecting DoS vulnerabilities in Fastapi requires understanding both the framework's architecture and common attack patterns. middleBrick's Fastapi-specific scanning identifies these issues through black-box testing without requiring access to your source code.

For blocking operations in async endpoints, middleBrick analyzes response timing patterns. Fastapi applications should maintain consistent low-latency responses. When endpoints show high variance or timeouts during load testing, it indicates blocking operations:

# middleBrick would flag this pattern
@app.get("/vulnerable")
async def vulnerable():
    # Synchronous database call blocks event loop
    result = sync_database_query()
    return result

The scanner tests endpoints with concurrent requests and measures whether response times degrade significantly, indicating blocking operations that prevent proper async handling.

File upload vulnerabilities are detected by attempting to upload files of increasing sizes and monitoring memory usage patterns. middleBrick tests for missing size validation by attempting uploads that exceed reasonable limits:

# middleBrick tests for missing validation like this
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    # No size checking - vulnerable to memory exhaustion
    content = await file.read()
    return {"received": len(content)}

The scanner also verifies whether your Fastapi application properly handles multipart form data and doesn't crash when receiving malformed or excessively large payloads.

middleBrick detects rate limiting gaps by sending rapid bursts of requests to your endpoints and analyzing response patterns. Fastapi applications without proper rate limiting will respond to all requests, while protected endpoints show throttling or blocking:

# middleBrick identifies missing rate limiting
@app.get("/no-rate-limit")
async def no_rate_limit():
    # Can be called repeatedly without restriction
    return {"ok": True}

The scanner tests various HTTP methods and endpoints to ensure consistent rate limiting across your entire API surface.

For dependency injection vulnerabilities, middleBrick analyzes your OpenAPI specification and tests for endpoints that might create recursive dependency chains. The scanner looks for patterns where dependencies could create infinite loops or excessive resource consumption:

# middleBrick would flag potential recursion
async def recursive_dep():
    # Could create infinite recursion
    return await recursive_dep()

@app.get("/vulnerable")
async def vulnerable(dep: str = Depends(recursive_dep)):
    return {"result": dep}

The scanner also tests for endpoints that make excessive external calls or database queries without proper pagination or limits.

Remediating DoS vulnerabilities in Fastapi requires leveraging the framework's built-in features and Python's async capabilities. Here's how to fix the specific vulnerabilities identified in the previous sections.

For blocking operations in async endpoints, use asyncio.to_thread() or run_in_executor() to properly handle blocking I/O:

from fastapi import FastAPI
import asyncio
import time

app = FastAPI()

@app.get("/fixed-blocking")
async def fixed_blocking():
    # Properly offload blocking operation to thread pool
    result = await asyncio.to_thread(time.sleep, 5)
    return {"message": "Done"}

This allows the event loop to continue processing other requests while the blocking operation runs in a separate thread. For CPU-bound operations, consider using ProcessPoolExecutor:

from fastapi import FastAPI
import asyncio
from concurrent.futures import ProcessPoolExecutor

app = FastAPI()

executor = ProcessPoolExecutor(max_workers=4)

@app.get("/cpu-bound")
async def cpu_bound():
    # Offload CPU-bound work to separate process
    result = await asyncio.get_event_loop().run_in_executor(
        executor, expensive_computation
    )
    return result

For file uploads, implement size validation and streaming to prevent memory exhaustion:

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
import aiofiles

app = FastAPI()

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB limit

@app.post("/upload-safe")
async def upload_safe(file: UploadFile = File(...)):
    if int(file.size) > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail="File too large. Maximum size is 10MB"
        )
    
    # Stream to disk instead of loading into memory
    async with aiofiles.open(f"/tmp/{file.filename}", 'wb') as out_file:
        content = await file.read(1024)
        while content:
            await out_file.write(content)
            content = await file.read(1024)
    
    return JSONResponse(status_code=201, content={"filename": file.filename})

Implement rate limiting using Fastapi's middleware or third-party libraries:

from fastapi import FastAPI
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.middleware("http")
async def add_rate_limit(request, call_next):
    if request.method == "GET":
        limit = "10/minute"  # 10 requests per minute
    else:
        limit = "5/minute"
    
    if not limiter.is_allowed(request, limit=limit):
        raise RateLimitExceeded()
    
    return await call_next(request)

For dependency injection safety, implement depth limits and validation:

from fastapi import FastAPI, Depends, HTTPException
from typing import Optional

app = FastAPI()

MAX_RECURSION_DEPTH = 10

async def get_user_data(
    user_id: str, 
    depth: int = 0, 
    max_depth: int = 10
):
    if depth > max_depth:
        raise HTTPException(
            status_code=400,
            detail="Maximum recursion depth exceeded"
        )
    
    # Your actual data fetching logic here
    return await get_user_data_from_db(user_id)

@app.get("/user-safe")
async def read_user(
    data: dict = Depends(get_user_data),
    depth: int = 0
):
    return data

For external API calls, implement timeouts and connection pooling:

from fastapi import FastAPI
import httpx
from contextlib import asynccontextmanager

app = FastAPI()

@asynccontextmanager
async def get_timed_client():
    async with httpx.AsyncClient(
        timeout=10.0,  # 10 second timeout
        limits=httpx.Limits(
            max_keepalive_connections=10,
            max_connections=100
        )
    ) as client:
        yield client

@app.get("/external-safe")
async def external_safe():
    async with get_timed_client() as client:
        try:
            response = await client.get("https://api.example.com/data")
            response.raise_for_status()
        except httpx.TimeoutException:
            return {"error": "External service timeout"}
    return response.json()