Out Of Bounds Write in Fastapi
How Out Of Bounds Write Manifests in Fastapi
Out Of Bounds Write (OOBW) in Fastapi applications typically occurs when code writes data beyond the allocated boundaries of buffers, arrays, or data structures. Fastapi's asynchronous nature and heavy use of Pydantic models creates unique attack surfaces for OOBW vulnerabilities.
One common pattern involves Pydantic model validation with list fields. Consider this vulnerable endpoint:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class UserInput(BaseModel):
items: list[int] = []
@app.post("/process")
async def process_items(input: UserInput):
buffer = [0] * 10 # Fixed-size buffer
for i, item in enumerate(input.items):
buffer[i] = item # OOBW if input.items has >10 items
return bufferThis code allows attackers to write beyond the 10-element buffer by sending more than 10 items, causing memory corruption or crashes.
Another Fastapi-specific scenario involves response models with dynamic sizing. When using response_model and dependency injection:
from fastapi import Depends
async def get_user_data() -> dict:
return {'scores': [0]*5} # Returns 5-element list
@app.get("/user/leaderboard", response_model=Leaderboard)
async def get_leaderboard(user_data: dict = Depends(get_user_data)):
leaderboard = Leaderboard()
for i, score in enumerate(user_data['scores']):
leaderboard.scores[i] = score # OOBW if scores list grows unexpectedly
return leaderboardFastapi's background tasks can also introduce OOBW risks when processing large datasets asynchronously:
from fastapi import BackgroundTasks
@app.post("/upload")
async def upload_data(data: list[int], background_tasks: BackgroundTasks):
background_tasks.add_task(process_large_dataset, data)
return {"status": "processing"}
async def process_large_dataset(data: list[int]):
buffer = bytearray(1024)
for i, byte in enumerate(data):
buffer[i] = byte # OOBW if data exceeds 1024 bytesThe asynchronous nature means the vulnerability might not manifest immediately, making detection harder.
Fastapi-Specific Detection
Detecting OOBW in Fastapi requires both static analysis and runtime scanning. Static analysis tools like bandit can catch some patterns:
pip install bandit
bandit -r app.py -s B605,B608
However, static analysis misses Fastapi's dynamic aspects. Runtime detection with middleBrick provides comprehensive coverage:
npm install -g middlebrick
middlebrick scan https://your-fastapi-app.com/docs
middleBrick's black-box scanning tests the unauthenticated attack surface by sending crafted payloads to Fastapi endpoints. For OOBW detection, it specifically targets:
- POST endpoints with list/array parameters, sending oversized payloads
- Endpoints using Pydantic models with unbounded list fields
- Background task endpoints that process data asynchronously
- Response models that might have size mismatches
The scanner tests each endpoint with progressively larger inputs, monitoring for crashes, memory errors, or unexpected behavior. For Fastapi applications, middleBrick also examines OpenAPI specs to understand expected input sizes and types.
Manual testing with tools like httpie or curl can complement automated scanning:
# Test with oversized list
http POST :8000/process items:='[1,2,3,4,5,6,7,8,9,10,11,12]'
# Test with large binary data
http POST :8000/upload @large_file.bin
Watch for HTTP 500 errors, timeouts, or memory usage spikes when testing with large inputs.
Fastapi-Specific Remediation
Fastapi provides several native mechanisms to prevent OOBW vulnerabilities. The most effective approach is using Pydantic's validation features:
from pydantic import BaseModel, conlist
class SafeInput(BaseModel):
items: conlist(int, min_items=1, max_items=10) # Constrained list with bounds
@app.post("/process-safe")
async def process_safe(input: SafeInput):
buffer = [0] * 10
for i, item in enumerate(input.items):
buffer[i] = item # Now safe - input validated to max 10 items
return bufferThe conlist type enforces size constraints at the model level, rejecting oversized inputs before they reach your business logic.
For scenarios requiring dynamic sizing, use safe iteration patterns:
from fastapi import HTTPException
@app.post("/process-dynamic")
async def process_dynamic(input: UserInput):
if len(input.items) > 1000:
raise HTTPException(status_code=413, detail="Payload too large")
result = []
for item in input.items[:1000]: # Safe slicing
result.append(item * 2)
return resultFastapi's dependency injection system can also enforce size limits:
from fastapi import Header, HTTPException
def validate_size_limit(x_total_count: int = Header(...)):
if x_total_count > 1000:
raise HTTPException(status_code=413)
return True
@app.post("/upload-validated", dependencies=[Depends(validate_size_limit)])
async def upload_validated(data: UploadFile):
return {"status": "uploaded"}For background tasks, implement size validation before task creation:
from fastapi import BackgroundTasks
@app.post("/upload-safe")
async def upload_safe(data: list[int], background_tasks: BackgroundTasks):
if len(data) > 10000:
raise HTTPException(status_code=413)
background_tasks.add_task(process_large_dataset, data)
return {"status": "processing"}Combine these approaches with Fastapi's middleware for comprehensive protection:
from fastapi.middleware import Middleware
from fastapi import FastAPI
class SizeLimitingMiddleware:
def __init__(self, app, max_size=1000000):
self.app = app
self.max_size = max_size
async def __call__(self, scope, receive, send):
if scope['type'] == 'http' and 'content-length' in scope:
if int(scope['content-length']) > self.max_size:
async def mock_send(message):
if message['type'] == 'http.response.start':
message['status'] = 413
return await mock_send({'type': 'http.response.start'})
return await self.app(scope, receive, send)
app.add_middleware(SizeLimitingMiddleware)