Http Request Smuggling in Fastapi

Http Request Smuggling exploits inconsistencies in how HTTP message parsing is implemented across different servers, proxies, or application layers. In Fastapi applications, this vulnerability often appears through subtle interactions between Fastapi's request handling and the underlying ASGI server (typically Uvicorn or Hypercorn).

The most common Fastapi-specific smuggling vector involves Fastapi's handling of multipart form data combined with Fastapi's automatic request body parsing. When a client sends a request with carefully crafted Content-Length and Transfer-Encoding headers, Fastapi's dependency injection system can be tricked into processing the wrong request body.

Consider this Fastapi endpoint that accepts file uploads:

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {"size": len(content)}

The vulnerability arises because Fastapi's UploadFile dependency automatically reads the request body, but if the request contains conflicting length headers, the ASGI server might parse the body differently than Fastapi expects. An attacker could send:

POST /upload/ HTTP/1.1
Host: example.com
Content-Length: 100
Transfer-Encoding: chunked

0

POST /admin/ HTTP/1.1
Host: example.com
Content-Type: application/json
Content-Length: 23

{"admin": true}

In this scenario, the ASGI server might interpret the chunked encoding and pass the second request to Fastapi's routing system, potentially bypassing authentication middleware that sits above Fastapi in the stack.

Another Fastapi-specific vector involves Fastapi's background task system. When using @app.on_event("startup") or background tasks with request-scoped dependencies, smuggling can cause background tasks to execute with malformed request data:

@app.post("/process/")
async def process_data(data: dict = Body(...)):
    app.add_background_task(process_background, data)
    return {"status": "accepted"}

async def process_background(data: dict):
    # This might receive corrupted data from smuggling
    process_sensitive_operation(data)

The background task receives the data object that was parsed from the potentially corrupted request, leading to unexpected behavior in what should be a trusted execution context.

Detecting HTTP Request Smuggling in Fastapi requires understanding both the application layer and the transport layer where the ASGI server operates. The most effective approach combines automated scanning with manual testing specific to Fastapi's architecture.

middleBrick's black-box scanning approach is particularly effective for Fastapi applications because it tests the unauthenticated attack surface without requiring access to source code. The scanner sends requests with varying Content-Length and Transfer-Encoding combinations to identify inconsistencies in how the Fastapi application and any reverse proxies handle the requests.

For manual detection in Fastapi applications, start by examining your middleware stack. Fastapi applications often sit behind multiple layers including Nginx, Cloudflare, or other CDNs. Each layer might handle request parsing differently:

from fastapi import FastAPI, Request
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# Common middleware stack
app.add_middleware(HTTPSRedirectMiddleware)
app.add_middleware(CustomAuthMiddleware)

@app.middleware("http")
async def smuggling_protection(request: Request, call_next):
    # Check for conflicting headers
    content_length = request.headers.get("content-length")
    transfer_encoding = request.headers.get("transfer-encoding")
    
    if content_length and transfer_encoding:
        # Potential smuggling attempt - log and reject
        raise HTTPException(status_code=400, detail="Conflicting headers detected")
    
    return await call_next(request)

middleBrick specifically tests Fastapi applications by sending CL.TE (Content-Length header vs Transfer-Encoding chunked), TE.CL (Transfer-Encoding: chunked with Content-Length), and ambiguous requests that could be interpreted differently by Fastapi versus the ASGI server.

Another detection technique involves monitoring request processing times. Smuggled requests often cause the application to hang or process requests incorrectly, leading to timeout patterns that can be detected through logging and monitoring.

Remediating HTTP Request Smuggling in Fastapi applications requires a defense-in-depth approach that addresses both the Fastapi application layer and the deployment infrastructure. The most effective strategy combines Fastapi-native protections with proper server configuration.

At the Fastapi application level, implement strict header validation middleware that rejects requests with conflicting or ambiguous headers:

from fastapi import FastAPI, Request, HTTPException
from starlette.requests import Request

app = FastAPI()

class SmugglingProtectionMiddleware:
    async def __call__(self, request: Request, call_next):
        headers = request.headers
        
        # Check for Content-Length and Transfer-Encoding conflict
        if "content-length" in headers and "transfer-encoding" in headers:
            raise HTTPException(
                status_code=400, 
                detail="Conflicting transfer headers detected"
            )
        
        # Check for multiple Content-Length headers (case-insensitive)
        content_lengths = [h for h in headers.keys() if h.lower() == "content-length"]
        if len(content_lengths) > 1:
            raise HTTPException(
                status_code=400, 
                detail="Multiple content-length headers detected"
            )
        
        # Validate Content-Length is a valid integer
        try:
            if "content-length" in headers:
                int(headers["content-length"])
        except ValueError:
            raise HTTPException(
                status_code=400, 
                detail="Invalid content-length header"
            )
        
        return await call_next(request)

app.add_middleware(SmugglingProtectionMiddleware)

For file upload endpoints, which are common targets for smuggling attacks in Fastapi, implement strict size limits and content validation:

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

app = FastAPI()

@app.post("/upload/", timeout=10.0)
async def upload_file(
    file: UploadFile = File(
        default=None, 
        description="Upload a file",
        example={}
    )
):
    if file is None:
        raise HTTPException(status_code=400, detail="No file provided")
    
    # Validate file size before reading
    if int(file.headers.get("content-length", 0)) > 10 * 1024 * 1024:  # 10MB limit
        raise HTTPException(status_code=413, detail="File too large")
    
    # Read file with timeout protection
    try:
        content = await asyncio.wait_for(file.read(), timeout=5.0)
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="Request timeout")
    
    return JSONResponse(content={"size": len(content)})

On the infrastructure side, configure your ASGI server (Uvicorn/Hypercorn) and any reverse proxies to normalize request parsing. For example, in Nginx in front of a Fastapi application:

server {
    listen 80;
    server_name example.com;
    
    # Reject requests with both Content-Length and Transfer-Encoding
    if ($http_content_length != "" && $http_transfer_encoding != "") {
        return 400;
    }
    
    # Only allow specific methods
    if ($request_method !~ ^(GET|POST|PUT|DELETE|OPTIONS)$) {
        return 405;
    }
    
    location / {
        proxy_pass http://fastapi_app;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        
        # Critical: ensure proxy doesn't re-add removed headers
        proxy_set_header Content-Length "";
        proxy_set_header Transfer-Encoding "";
    }
}

For Fastapi applications using dependency injection with request body parsing, ensure that all dependencies properly validate their input:

from pydantic import BaseModel
from fastapi import Body, HTTPException
from typing import Optional

class UploadRequest(BaseModel):
    file: bytes
    metadata: Optional[dict] = None
    
    @classmethod
    def validate_payload(cls, payload: bytes, content_length: int):
        if len(payload) != content_length:
            raise ValueError("Payload length mismatch")
        return cls.parse_obj(payload)

@app.post("/api/upload")
async def upload_endpoint(
    data: UploadRequest = Body(...)
):
    try:
        # The validation happens in the Pydantic model
        return {"status": "success", "size": len(data.file)}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))