Distributed Denial Of Service in Fastapi

Distributed Denial of Service (DDoS) attacks against Fastapi applications exploit the framework's asynchronous nature and request handling patterns. Unlike traditional synchronous frameworks, Fastapi's async/await model can create unique attack vectors that overwhelm specific Fastapi components.

The most common Fastapi-specific DDoS pattern targets the async def endpoint functions. When attackers flood an endpoint with requests, Fastapi's event loop becomes saturated with pending coroutines. Each pending request consumes memory for its context, and if the endpoint performs I/O operations (database queries, external API calls), these operations can block the event loop, creating a cascading failure.

Consider this vulnerable Fastapi endpoint:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/slow-endpoint")
async def slow_endpoint():
    await asyncio.sleep(10)  # Simulates slow I/O
    return {"message": "Processed"}

An attacker can send thousands of requests to /slow-endpoint, each consuming 10 seconds of event loop time. Fastapi will queue these requests, eventually exhausting memory and causing legitimate requests to time out.

Another Fastapi-specific vector targets the dependency injection system. Fastapi's Depends() creates new instances for each request, and if a dependency performs expensive operations, it becomes a bottleneck:

from fastapi import Depends

async def expensive_dependency():
    await asyncio.sleep(2)  # Expensive operation
    return "result"

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

Attackers can exploit this by repeatedly calling /vulnerable, causing the dependency to execute concurrently across thousands of requests, overwhelming the system.

Fastapi's background tasks also present a unique attack surface. Background tasks run after the response is sent, but if attackers trigger expensive background tasks, they can exhaust worker resources:

from fastapi import BackgroundTasks

async def expensive_background_task():
    await asyncio.sleep(30)  # Long-running task

@app.post("/trigger")
async def trigger_background(background_tasks: BackgroundTasks):
    background_tasks.add_task(expensive_background_task)
    return {"status": "triggered"}

A single endpoint call creates a 30-second background task. An attacker sending hundreds of requests creates hundreds of concurrent background tasks, potentially exhausting system resources.

Fastapi's streaming responses (StreamingResponse) can also be exploited. If an endpoint streams data slowly or performs expensive operations during streaming, attackers can keep connections open indefinitely, exhausting connection pools:

from fastapi.responses import StreamingResponse

@app.get("/stream")
async def stream_data():
    async def generator():
        for i in range(1000):
            await asyncio.sleep(1)  # Slow streaming
            yield f"data-{i}\
"
    return StreamingResponse(generator())

Each streaming connection holds resources for the duration of the stream, making it an effective DDoS vector when abused.

Detecting DDoS vulnerabilities in Fastapi requires understanding the framework's specific behaviors and monitoring the right metrics. middleBrick's black-box scanning approach is particularly effective for Fastapi applications because it tests the actual running API without requiring source code access.

middleBrick scans Fastapi endpoints for DDoS vulnerabilities by sending controlled request bursts and measuring response patterns. For async endpoints, it analyzes how the event loop handles concurrent requests by monitoring response times and error rates under load. The scanner identifies endpoints that exhibit slow degradation or timeouts when subjected to moderate request volumes.

For Fastapi's dependency injection system, middleBrick traces the dependency resolution process. It detects when dependencies create performance bottlenecks by measuring the time between request receipt and dependency resolution completion. If dependencies consistently take longer than 100ms to resolve, this indicates a potential DDoS vulnerability.

middleBrick also tests Fastapi's background task handling by triggering endpoints with background tasks and monitoring system resource usage. It identifies endpoints where background tasks accumulate faster than they complete, creating a growing backlog that can exhaust system resources.

The scanner examines Fastapi's streaming response implementations by measuring connection duration and resource consumption. It identifies endpoints where streaming connections remain open for extended periods or consume disproportionate resources relative to the data being transmitted.

middleBrick's LLM security checks are particularly relevant for Fastapi applications using AI features. The scanner tests for excessive agency patterns in Fastapi endpoints that interact with language models, ensuring that AI-powered endpoints don't become DDoS vectors through recursive or resource-intensive operations.

For Fastapi applications using OpenAPI specifications, middleBrick cross-references the spec with runtime findings. It identifies endpoints marked as summary or description that suggest resource-intensive operations, then tests these endpoints for DDoS vulnerabilities.

The scanner's rate limiting detection is crucial for Fastapi applications. It identifies endpoints that lack rate limiting headers (X-RateLimit-Limit, X-RateLimit-Remaining) or that don't implement proper rate limiting at the application level, leaving them vulnerable to request flooding.

middleBrick provides Fastapi-specific remediation guidance based on its findings. For async endpoints showing poor concurrency handling, it recommends Fastapi's BackgroundTask patterns or external rate limiting middleware. For dependency injection bottlenecks, it suggests caching strategies or dependency optimization techniques specific to Fastapi's architecture.

Remediating DDoS vulnerabilities in Fastapi requires leveraging the framework's native features and understanding async programming patterns. Here are Fastapi-specific solutions for common DDoS vectors.

For async endpoints vulnerable to request flooding, implement request limiting using Fastapi's middleware:

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded

app = FastAPI()

limiter = Limiter(key_func=lambda req: req.client.host)
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.state.limiter = limiter

@app.middleware("http")
async def add_rate_limit_headers(request: Request, call_next):
    if request.method != "OPTIONS":
        limit = app.state.limiter.request_limit
        remaining = app.state.limiter.remaining
        headers = {
            "X-RateLimit-Limit": str(limit),
            "X-RateLimit-Remaining": str(remaining),
        }
        response = await call_next(request)
        for key, value in headers.items():
            response.headers[key] = value
        return response

@app.get("/protected")
@limiter.limit("10/minute")
async def protected_endpoint():
    return {"message": "Protected resource"}

This middleware adds rate limiting headers and enforces limits using the slowapi package, which integrates seamlessly with Fastapi's async architecture.

For dependency injection bottlenecks, implement caching and async-aware dependency patterns:

from fastapi import Depends, FastAPI
from functools import lru_cache
import asyncio

app = FastAPI()

async def expensive_operation():
    await asyncio.sleep(2)
    return "cached-result"

@lru_cache(maxsize=32)
async def cached_expensive_operation():
    return await expensive_operation()

async def cached_dependency():
    return await cached_expensive_operation()

@app.get("/cached")
async def cached_endpoint(dep: str = Depends(cached_dependency)):
    return {"dep": dep}

This pattern caches expensive dependency results, reducing the impact of repeated requests. The lru_cache decorator works with async functions in Fastapi.

For background task vulnerabilities, implement task limiting and monitoring:

from fastapi import BackgroundTasks, FastAPI
from collections import deque
import asyncio

app = FastAPI()

class TaskLimiter:
    def __init__(self, max_concurrent: int):
        self.max_concurrent = max_concurrent
        self.active_tasks = 0
        self.queue = deque()
    
    async def run_task(self, task_func):
        if self.active_tasks >= self.max_concurrent:
            await asyncio.sleep(0.1)
            return await self.run_task(task_func)
        
        self.active_tasks += 1
        try:
            await task_func()
        finally:
            self.active_tasks -= 1

limiter = TaskLimiter(max_concurrent=10)

async def controlled_background_task():
    await asyncio.sleep(30)

@app.post("/trigger-limited")
async def trigger_limited(background_tasks: BackgroundTasks):
    await limiter.run_task(expensive_background_task)
    return {"status": "queued"}

This implementation limits concurrent background tasks, preventing resource exhaustion from task accumulation.

For streaming response vulnerabilities, implement connection limiting and timeout controls:

from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

class ConnectionLimiter:
    def __init__(self, max_connections: int):
        self.max_connections = max_connections
        self.active_connections = 0
    
    async def acquire(self):
        while self.active_connections >= self.max_connections:
            await asyncio.sleep(0.1)
        self.active_connections += 1
    
    async def release(self):
        self.active_connections -= 1

connection_limiter = ConnectionLimiter(max_connections=100)

@app.get("/stream-protected")
async def stream_protected(request: Request):
    await connection_limiter.acquire()
    try:
        async def limited_generator():
            for i in range(100):
                await asyncio.sleep(0.1)
                yield f"data-{i}\
"
        return StreamingResponse(limited_generator())
    finally:
        await connection_limiter.release()

This pattern limits concurrent streaming connections and controls the streaming rate to prevent resource exhaustion.

For comprehensive DDoS protection, combine Fastapi's native features with external services like Cloudflare or AWS WAF, which provide network-layer protection before requests reach your Fastapi application.