Beast Attack in Fastapi

Beast Attack in Fastapi applications typically exploits the framework's synchronous request handling and dependency injection system. When Fastapi processes requests synchronously, an attacker can craft requests that force the server to perform expensive operations, blocking worker threads and degrading performance.

A common Beast Attack pattern in Fastapi involves the Depends() system. Consider this vulnerable endpoint:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db

app = FastAPI()

@app.get('/users/{user_id}')
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail='User not found')
    return user

The vulnerability here is that get_db() creates a new database connection for every request, and if an attacker sends thousands of concurrent requests, they can exhaust the connection pool. Fastapi's default worker model means each blocked request ties up a worker thread.

Another Fastapi-specific manifestation occurs with background tasks. The BackgroundTasks class can be abused:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post('/process')
def process_data(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(expensive_operation, data)
    return {'status': 'processing'}

An attacker can trigger background tasks that consume excessive CPU or memory, and because Fastapi doesn't limit background task execution time, these can run indefinitely.

Rate limiting is another critical area. Fastapi's lack of built-in rate limiting makes it vulnerable to request flooding:

@app.get('/api/data')
def get_data():
    # No rate limiting - vulnerable to Beast Attack
    return {'data': expensive_query()}

Without rate limiting middleware, an attacker can send thousands of requests per second, overwhelming the server's ability to process legitimate traffic.

Detecting Beast Attack vulnerabilities in Fastapi requires both manual code review and automated scanning. middleBrick's API security scanner specifically tests for these Fastapi-related patterns.

middleBrick scans Fastapi endpoints by sending concurrent requests to identify synchronous blocking behavior. The scanner tests:

• Database connection exhaustion by simulating high-concurrency scenarios
• Background task abuse by triggering endpoints with BackgroundTasks parameters
• Rate limiting bypass attempts using multiple IP addresses
• Dependency injection abuse by testing Depends() chains

Here's how you can test for Beast Attack vulnerabilities manually in Fastapi:

import httpx
import asyncio

async def test_beast_attack(url, num_requests=1000):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for _ in range(num_requests)]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        successful = sum(1 for r in responses if isinstance(r, httpx.Response))
        failed = sum(1 for r in responses if isinstance(r, Exception))
        
        return {
            'successful_requests': successful,
            'failed_requests': failed,
            'error_rate': failed / num_requests
        }

# Test a Fastapi endpoint
result = asyncio.run(test_beast_attack('http://localhost:8000/api/endpoint'))
print(f"Success rate: {result['successful_requests'] / 1000 * 100:.1f}%")

middleBrick's scanner goes further by analyzing the OpenAPI spec generated by Fastapi to identify vulnerable patterns. It looks for:

• Endpoints with expensive database operations
• Background task endpoints without proper validation
• Missing rate limiting annotations
• Unbounded dependency chains

The scanner also tests for specific Fastapi behaviors like:

# Testing for synchronous blocking
@app.get('/sync-blocking')
def sync_blocking():
    # This will block the event loop
    time.sleep(10)
    return {'status': 'done'}

middleBrick identifies these patterns and reports them with specific remediation guidance for Fastapi applications.

Remediating Beast Attack vulnerabilities in Fastapi requires a multi-layered approach. The first line of defense is implementing proper rate limiting using Fastapi-compatible middleware.

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, default_limits=["100/minute"])napp.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.middleware("http")
async def add_rate_limit(request, call_next):
    if request.method == 'GET':
        return await call_next(request)
    return await limiter.limit("10/minute")(call_next)(request)

@app.get('/api/data')
@limiter.limit("50/minute")
def get_data():
    return {'data': expensive_query()}

For database connection management, use connection pooling with proper limits:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:pass@localhost/db"
engine = create_engine(DATABASE_URL, pool_size=20, max_overflow=10, pool_timeout=30)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Implement async operations to prevent blocking:

import asyncio
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def async_expensive_operation(data: dict):
    await asyncio.sleep(2)  # Non-blocking sleep
    # Perform actual async work
    return processed_data

@app.post('/process')
def process_data(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(async_expensive_operation, data)
    return {'status': 'processing'}

Add request timeouts to prevent long-running operations:

from fastapi import FastAPI, Request, HTTPException
from starlette.responses import JSONResponse

app = FastAPI()

@app.middleware("http")
async def timeout_middleware(request: Request, call_next):
    try:
        response = await call_next(request)
        return response
    except asyncio.TimeoutError:
        return JSONResponse(
            status_code=503,
            content={"detail": "Request timeout"}
        )

@app.get('/api/endpoint', timeout=5.0)
async def endpoint():
    # Operation that should complete within 5 seconds
    result = await some_async_operation()
    return result

middleBrick's Pro plan includes continuous monitoring that automatically scans your Fastapi endpoints on a schedule, alerting you if new Beast Attack vulnerabilities are introduced during development.