Cache Poisoning in Fastapi

Cache poisoning in Fastapi applications occurs when untrusted user input influences cached responses, causing subsequent users to receive manipulated or incorrect data. Fastapi's async nature and integration with caching backends like Redis or in-memory caches creates specific attack vectors that differ from traditional web frameworks.

The most common Fastapi cache poisoning scenario involves query parameters and path parameters being used as cache keys without proper validation. Consider this vulnerable endpoint:

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import redis

app = FastAPI()
redis_client = redis.Redis(host='redis', port=6379, db=0)

@app.get("/api/users/{user_id}")
async def get_user(user_id: str):
    cache_key = f"user:{user_id}"
    cached = redis_client.get(cache_key)
    if cached:
        return JSONResponse(content=cached)
    
    # Simulate database query
    user_data = await get_user_from_db(user_id)
    redis_client.set(cache_key, user_data)
    return JSONResponse(content=user_data)

An attacker can craft malicious user_id values like ../../../../etc/passwd or SQL injection payloads. If the application doesn't properly sanitize these inputs before using them as cache keys, the attacker's response gets cached and served to other users.

Fastapi's dependency injection system creates another attack surface. When dependencies are cached without considering their state or authentication context:

@app.get("/api/data")
async def get_sensitive_data(
    db: Database = Depends(get_database)
):
    # Database connection cached without auth context
    return await db.query("SELECT * FROM sensitive_data")

If the database dependency is cached per request rather than per authenticated user, one user's query results could be served to another user.

Header-based cache poisoning is particularly problematic in Fastapi. Attackers can manipulate cache keys through custom headers:

@app.get("/api/cache-me")
async def cache_with_headers(
    x_user_id: str = Header(...),
    x_session: str = Header(...)
):
    cache_key = f"data:{x_user_id}:{x_session}"
    # No validation of header values
    return await expensive_operation()

Malicious headers like X-User-Id: admin or X-Session: .. can poison the cache with privileged responses.

Detecting cache poisoning in Fastapi requires examining both the application code and runtime behavior. middleBrick's black-box scanning approach is particularly effective because it tests the actual attack surface without requiring source code access.

middleBrick scans Fastapi endpoints for cache poisoning by testing parameter manipulation patterns. The scanner attempts to inject special characters, path traversal sequences, and SQL-like syntax into path parameters and query strings, then verifies if these manipulated responses get cached and served to other users.

For Fastapi applications using Redis or similar backends, middleBrick can detect improper cache key construction by:

1. Analyzing the OpenAPI specification to understand parameter types and expected formats
2. Crafting boundary case inputs that would create predictable cache key collisions
3. Verifying if manipulated responses are served from cache to subsequent requests
4. Checking for missing input validation on parameters used in cache keys

The scanner also examines Fastapi's dependency injection patterns. It looks for dependencies that might be cached without proper context isolation, particularly database connections and authentication services.

middleBrick's LLM security module adds another layer of detection for Fastapi applications using AI features. It tests for system prompt leakage and prompt injection that could manipulate cached AI responses, a growing concern as Fastapi becomes popular for AI API development.

Runtime monitoring with middleBrick provides continuous detection by scanning your Fastapi APIs on a configurable schedule. The Pro plan's continuous monitoring can alert you when new cache poisoning vulnerabilities are introduced through code changes.

Fastapi provides several native features to prevent cache poisoning. The most effective approach combines input validation, proper cache key construction, and context-aware caching.

First, always validate and sanitize parameters before using them in cache keys:

from fastapi import HTTPException
from pydantic import BaseModel, Field
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend

class UserID(BaseModel):
    user_id: str = Field(..., regex=r'^[a-zA-Z0-9_-]{3,20}$')

@app.get("/api/users/{user_id}")
async def get_user(user_id: str = Path(..., regex=r'^[a-zA-Z0-9_-]{3,20}$')):
    # Validate input before cache operations
    if not re.match(r'^[a-zA-Z0-9_-]{3,20}$', user_id):
        raise HTTPException(status_code=400, detail="Invalid user ID format")
    
    cache_key = f"user:{user_id}"
    cached = await FastAPICache.get(cache_key)
    if cached:
        return cached
    
    user_data = await get_user_from_db(user_id)
    await FastAPICache.set(cache_key, user_data, expire=3600)
    return user_data

The regex validation ensures only alphanumeric user IDs with specific characters are accepted, preventing path traversal and injection attacks.

For dependency injection, use Fastapi's per-request scope to avoid caching across user contexts:

@app.get("/api/data")
async def get_sensitive_data(
    db: Database = Depends(get_database, scope="request")
):
    # Database connection created per request, not cached across users
    return await db.query("SELECT * FROM sensitive_data WHERE user_id = $1", user_id)

Fastapi's built-in dependency scopes prevent the caching issues that plague other frameworks.

For header-based caching, implement strict header validation and use Fastapi's Header validator:

from fastapi import Header, HTTPException

@app.get("/api/cache-me")
async def cache_with_headers(
    x_user_id: str = Header(..., regex=r'^[a-f0-9]{24}$'),
    x_session: str = Header(..., regex=r'^session_[a-zA-Z0-9]{16}$')
):
    if not validate_session(x_session):
        raise HTTPException(status_code=401, detail="Invalid session")
    
    cache_key = f"data:{x_user_id}:{x_session}"
    return await expensive_operation()

The regex patterns ensure headers contain only expected formats, blocking malicious input.

For applications using Fastapi's caching middleware, configure proper cache invalidation policies:

from fastapi_cache import FastAPICache
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    await FastAPICache.init(
        RedisBackend(),
        prefix="ep"
    )

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    response = await call_next(request)
    # Add cache control headers to prevent client-side poisoning
    response.headers["Cache-Control"] = "private, no-store"
    return response

These patterns, combined with middleBrick's scanning capabilities, provide comprehensive protection against Fastapi cache poisoning attacks.