Replay Attack in Fastapi

Replay attacks in FastAPI applications occur when an attacker captures valid HTTP requests and resends them to the server, potentially causing unauthorized actions, data theft, or service disruption. FastAPI's async nature and modern async/await patterns create specific vulnerabilities that attackers can exploit.

The most common FastAPI replay scenarios involve:

• Authentication token reuse - JWT tokens or session IDs captured via network sniffing are resent to authenticate as the original user
• State-changing operations - POST, PUT, DELETE requests that create, modify, or delete data are replayed to repeat those actions
• Payment processing - Financial transactions are captured and resent to charge customers multiple times
• API rate limit bypass - Valid requests are replayed to circumvent rate limiting mechanisms

FastAPI's dependency injection system creates specific replay vulnerabilities. Consider this common pattern:

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from datetime import datetime

app = FastAPI()

class Transaction(BaseModel):
    amount: float
    description: str

@app.post("/process-payment/")
def process_payment(transaction: Transaction, timestamp: datetime = datetime.now()):
    # Process payment logic here
    return {"status": "processed", "timestamp": timestamp}

This endpoint is vulnerable because the timestamp parameter uses a default value that can be bypassed. An attacker can capture a valid request and replay it, causing duplicate payments. The async nature means multiple requests can be processed concurrently, increasing the attack surface.

Another FastAPI-specific vulnerability appears in WebSocket connections:

from fastapi import WebSocket

@app.websocket("/ws/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        # Process data without validation
        await websocket.send_text(f"Message processed: {data}")

WebSocket replay attacks can cause state corruption, unauthorized data access, or service disruption since the connection remains open for extended periods.

Detecting replay attacks in FastAPI requires monitoring specific patterns and implementing detection mechanisms. middleBrick's black-box scanning approach tests your FastAPI endpoints without requiring access to source code, making it ideal for production environments.

Key detection strategies for FastAPI applications:

Request fingerprinting - Track unique request signatures including headers, body content, and timing. FastAPI's request object provides comprehensive access to all request components:

from fastapi import Request
from collections import defaultdict

request_hashes = defaultdict(int)

@app.middleware("http")
async def replay_detection(request: Request, call_next):
    request_data = await request.body()
    request_hash = hash((request.method, request.url, request.headers, request_data))
    request_hashes[request_hash] += 1
    
    if request_hashes[request_hash] > 1:
        # Potential replay detected
        print(f"Possible replay attack: {request.method} {request.url}")
    
    response = await call_next(request)
    return response

Timing analysis - FastAPI's async capabilities mean legitimate requests can arrive quickly, but replay attacks often show abnormal timing patterns. Monitor request intervals and flag suspicious sequences.

middleBrick scanning - The scanner tests your FastAPI endpoints for replay vulnerabilities by:

• Sending multiple identical requests to test rate limiting effectiveness
• Analyzing response consistency to detect missing anti-replay measures
• Checking for proper nonce implementation
• Testing timestamp validation logic

The scanner's 12 security checks include specific replay attack detection that examines your FastAPI application's authentication and authorization mechanisms. middleBrick can identify if your endpoints lack proper replay protection without requiring any configuration or credentials.

State validation - FastAPI's dependency injection makes it easy to add state validation:

from fastapi import Header, HTTPException
from typing import Optional

replay_tokens = set()

def validate_nonce(nonce: Optional[str] = Header(None)):
    if not nonce:
        raise HTTPException(status_code=400, detail="Nonce required")
    
    if nonce in replay_tokens:
        raise HTTPException(status_code=403, detail="Replay attack detected")
    
    replay_tokens.add(nonce)
    return nonce

middleBrick's scanning can verify if such protections are properly implemented by testing with various nonce values and checking for consistent rejection of replayed requests.

FastAPI provides several native mechanisms to prevent replay attacks. The key is implementing proper request validation and state tracking using FastAPI's dependency injection system.

Nonce-based protection - FastAPI's dependency injection makes nonce validation straightforward:

from fastapi import Depends, FastAPI, HTTPException
from uuid import uuid4
from datetime import datetime, timedelta

app = FastAPI()

class Nonce:
    def __init__(self):
        self.nonce = str(uuid4())
        self.timestamp = datetime.now()
    
    def is_valid(self):
        return datetime.now() - self.timestamp < timedelta(minutes=5)

async def get_nonce():
    return Nonce()

@app.post("/secure-endpoint/")
def secure_endpoint(
    transaction: dict,
    nonce: Nonce = Depends(get_nonce)
):
    if not nonce.is_valid():
        raise HTTPException(status_code=403, detail="Invalid or expired nonce")
    
    # Process transaction
    return {"status": "success", "nonce": nonce.nonce}

Timestamp validation - FastAPI's Pydantic models enable precise timestamp validation:

from pydantic import BaseModel, Field
from datetime import datetime, timedelta

class TimestampedRequest(BaseModel):
    data: dict
    timestamp: datetime = Field(default_factory=datetime.now)
    
    @validator('timestamp')
    def validate_timestamp(cls, v):
        if datetime.now() - v > timedelta(seconds=300):
            raise ValueError('Request too old')
        return v

@app.post("/timestamped/")
def timestamped_endpoint(request: TimestampedRequest):
    return {"status": "processed", "received": request.timestamp}

Rate limiting integration - FastAPI integrates well with rate limiting libraries:

from fastapi import FastAPI, Request
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: Request, call_next):
    if request.method == "POST" and "/process/" in request.url.path:
        key = f"replay_protection_{request.client.host}"
        if not limiter.is_allowed(key, limit_value=1, period=60):
            raise HTTPException(status_code=429, detail="Rate limit exceeded")
    return await call_next(request)

middleBrick integration - After implementing these protections, use middleBrick to verify effectiveness:

# Scan your FastAPI endpoint
middlebrick scan https://your-api.com/process-payment

# Check the replay attack detection results
# middleBrick will report if your endpoint:
# - Accepts replayed requests
# - Lacks proper nonce validation
# - Has insufficient rate limiting
# - Missing timestamp validation

The scanner's continuous monitoring (Pro plan) can automatically detect if replay protections degrade over time, alerting you when new vulnerabilities are discovered in your FastAPI application.