Cache Poisoning in Fastapi (Python)

Cache poisoning in a Fastapi application written in Python occurs when an attacker causes the application or an intermediary cache to store malicious or incorrect responses that are later served to other users. This typically arises when input that is not sufficiently validated or normalized is used to construct cache keys, or when responses vary by user context but are cached as if they were public and identical for all users.

Fastapi, which is built on Starlette for HTTP handling and supports both async and standard Python functions, often uses dependencies, path parameters, and query strings to influence responses. If these inputs are reflected into cache keys without proper sanitization, an attacker may manipulate them to poison cached data. For example, a cache layer that uses the request path and selected query parameters as the key might store user-specific data under a key derived from a manipulated parameter, leading to BOLA/IDOR-like exposure where one user receives another user’s data.

Python-specific factors can exacerbate the issue. Dynamic typing and runtime evaluation, such as careless use of eval, exec, or serialization formats like pickle, may allow injected content to be interpreted as code or structured data. If Fastapi endpoints construct cache keys by concatenating strings from user-supplied inputs without strict validation or encoding, Python code can inadvertently produce keys that collide or overwrite legitimate entries. Additionally, Python libraries used for caching may retain headers or fragments of responses that include sensitive information when cache rules are misconfigured.

Real-world attack patterns relevant to this combination include injection of newline characters to create alternate cache entries, manipulation of content negotiation headers, and exploitation of template fragments that are cached at the application level. These techniques map to the broader OWASP API Top 10 category of Broken Object Level Authorization (BOLA) when cached resources are not properly scoped to the requesting identity. A thorough security scan, such as those performed by tools that test unauthenticated attack surfaces and check for BOLA/IDOR, can surface misconfigured caching behavior before it is exploited in production.

To mitigate cache poisoning in Fastapi with Python, focus on strict input validation, canonicalization of cache keys, and ensuring that user-specific data is never inadvertently shared. Below are concrete, Python-specific fixes and code examples you can apply in your Fastapi services.

• Validate and normalize all inputs used in cache keys: Use Pydantic models to enforce types and constraints, and normalize inputs before using them in cache logic. For example, coerce query parameters to a canonical form and reject unexpected parameters.

from fastapi import FastAPI, Depends, HTTPException, Query
from pydantic import BaseModel, validator
import hashlib

app = Fastapi()

class CacheKeyParams(BaseModel):
    user_id: str
    resource_id: str

    @validator("user_id")
    def validate_user_id(cls, v):
        if not v.isalnum():
            raise ValueError("user_id must be alphanumeric")
        return v

    @validator("resource_id")
    def validate_resource_id(cls, v):
        if not v.isalnum():
            raise ValueError("resource_id must be alphanumeric")
        return v

def make_cache_key(params: CacheKeyParams) -> str:
    canonical = f"{params.user_id}:{params.resource_id}"
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

@app.get("/items/{user_id}")
async def read_item(
    user_id: str,
    resource_id: str = Query(..., min_length=1)
):
    params = CacheKeyParams(user_id=user_id, resource_id=resource_id)
    key = make_cache_key(params)
    # Use key with your caching backend
    return {"cache_key": key, "user_id": user_id, "resource_id": resource_id}

• Scope cached responses to the correct identity: Include tenant or user identifiers in the cache key and avoid caching sensitive responses at shared keys. Do not cache responses that contain private data under a global key.

import functools
from fastapi import Request

def user_scoped_cache(key_prefix: str):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            user_id = request.session.get("user_id") if request.session else None
            if not user_id:
                raise HTTPException(status_code=401, detail="Unauthorized")
            safe_key = f"{key_prefix}:{user_id}"
            # Replace with actual cache lookup/store using safe_key
            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

@app.get("/profile")
@user_scoped_cache("profile")
async def get_profile(request: Request):
    return {"profile_for": request.session.get("user_id")}

• Avoid unsafe deserialization and be cautious with eval/exec: Never use Python’s eval or exec on inputs that could reach cache logic. Prefer safe serialization such as JSON with typed schemas.

import json
from fastapi import Body

@app.post("/settings")
async def update_settings(payload: str = Body(...)):
    try:
        data = json.loads(payload)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")
    # Validate and use data safely; do not eval or exec
    return {"status": "ok"}

• Leverage middleware or dependencies for canonicalization: Normalize headers and paths consistently so that cache keys remain stable across equivalent requests.

from fastapi.middleware import Middleware
from fastapi.middleware.base import BaseHTTPMiddleware
import re

class PathNormalizerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Collapse multiple slashes and remove dot-segments safely
        clean_path = re.sub(r"/+", "/", request.url.path)
        # Reconstruct a normalized request environment if needed
        response = await call_next(request)
        return response

app.add_middleware(PathNormalizerMiddleware)