Cache Poisoning in Fastapi with Dynamodb

Cache poisoning in a FastAPI application that uses DynamoDB typically occurs when unvalidated or attacker-controlled data influences cache keys, cache entries, or downstream query parameters. Because FastAPI often caches HTTP responses or database query results to reduce DynamoDB read load, an attacker can manipulate inputs so that malicious or sensitive data is stored under legitimate keys, causing other users to receive incorrect responses.

Consider a FastAPI endpoint that uses a path parameter to query DynamoDB and caches the result. If the cache key is derived directly from user input without normalization or strict validation, an attacker can vary casing, whitespace, or encoding to poison the cache. For example, an endpoint like /users/{user_id} might compute a cache key as f"user:{user_id}". If the same logical user_id can be supplied as 123 or 123 (trailing space), separate cache entries are created, allowing an attacker to store a modified response under a variant key and have it returned to other users.

DynamoDB itself does not introduce cache poisoning, but its behavior can amplify risks when combined with caching layers. Because DynamoDB returns items by key, a poisoned cache key that does not match the expected key format can result in cache misses and repeated database queries. Worse, if the application caches query responses that include sensitive attributes (e.g., role or permissions fields), an attacker may be able to store crafted responses that expose or misrepresent data. For instance, an endpoint that includes query string parameters like ?role=admin and caches based on those parameters without canonicalization could allow an attacker to poison the cache with a modified role value, leading to privilege escalation for subsequent users who receive the poisoned cache entry.

Another common pattern involves caching results of DynamoDB scans or partially trusted queries. If input used to build a filter is reflected into cache keys, attackers can force storage of arbitrary or misleading data. Because FastAPI may reuse cached responses across requests, the poisoned entry can persist and affect multiple users. Additionally, if the caching layer is shared across tenants without proper isolation, cross-user cache poisoning can occur, where one tenant’s manipulated key or response affects another tenant’s cached data.

In this context, the combination of FastAPI’s dynamic routing and DynamoDB’s key-based access requires careful handling of inputs used to construct cache identifiers. Without canonicalization, strict validation, and separation of tenant or user contexts, the cache becomes a vector for storing and disseminating attacker-controlled content. This does not imply DynamoDB is at fault; rather, it highlights how integration choices in FastAPI can inadvertently expose the system to cache poisoning when user-influenced data reaches cache logic unchecked.

To detect such issues, scans should validate that cache keys are derived from canonical, normalized inputs and that sensitive data is not embedded in cached responses. The presence of unauthenticated endpoints that accept query parameters and cache results increases the attack surface. MiddleBrick’s LLM/AI Security checks and unauthenticated scan capabilities help surface these risks by analyzing the API surface and testing input handling without requiring credentials.

Remediation focuses on strict input validation, canonical cache keys, and isolating cached responses by tenant or user context. Below are concrete FastAPI examples that demonstrate safe patterns when working with DynamoDB.

• Validate and normalize identifiers before using them in cache keys:

from fastapi import FastAPI, HTTPException, Depends
import hashlib
import boto3
from pydantic import BaseModel

app = FastAPI()
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('users')

def normalize_user_id(user_id: str) -> str:
    # Remove whitespace and enforce expected format
    cleaned = user_id.strip().lower()
    if not cleaned.isalnum():
        raise ValueError('Invalid user_id')
    return cleaned

class User(BaseModel):
    id: str
    name: str
    role: str

def get_user_from_db(user_id: str) -> User:
    response = table.get_item(Key={'user_id': user_id})
    item = response.get('Item')
    if not item:
        raise HTTPException(status_code=404, detail='User not found')
    return User(**item)

@app.get('/users/{user_id}')
def read_user(user_id: str, use_cache: bool = True):
    normalized = normalize_user_id(user_id)
    cache_key = f'user:{hashlib.sha256(normalized.encode()).hexdigest()}'
    # pseudo-cache layer; ensure cache stores only normalized-keyed responses
    # response = cache.get(cache_key)
    # if not response:
    #     user = get_user_from_db(normalized)
    #     cache.set(cache_key, user.dict())
    # return response
    return get_user_from_db(normalized)

• Avoid caching responses that include sensitive or role-based data; if caching is required, exclude sensitive fields:

import json

def safe_cache_response(user: User) -> str:
    # Exclude role or other sensitive attributes from cached payload
    return json.dumps({'id': user.id, 'name': user.name})

# When storing in cache:
# cache.set(cache_key, safe_cache_response(user))

• Isolate cache entries by tenant or user context to prevent cross-user poisoning:

def get_tenant_cache_key(tenant_id: str, user_id: str) -> str:
    tenant_norm = normalize_user_id(tenant_id)
    user_norm = normalize_user_id(user_id)
    return f'tenant:{tenant_norm}:user:{hashlib.sha256(user_norm.encode()).hexdigest()}'

• Ensure query parameters that influence data selection are validated and canonicalized before inclusion in cache logic:

from typing import Optional

@app.get('/items/')
def list_items(
    role: Optional[str] = None,
    limit: int = 10,
    cache_bypass: bool = False
):
    canonical_role = None
    if role is not None:
        cleaned = role.strip().lower()
        if cleaned not in ('admin', 'user', 'guest'):
            raise HTTPException(status_code=400, detail='Invalid role')
        canonical_role = cleaned
    # Build deterministic cache key from validated inputs
    cache_key_parts = ['items', f'role:{canonical_role}', f'limit:{limit}']
    cache_key = ':'.join(cache_key_parts)
    # Use cache_key with DynamoDB query; ensure query parameters are parameterized to avoid injection
    # response = cache.get(cache_key) or query_dynamodb(canonical_role, limit)
    return {'role': canonical_role, 'limit': limit}

These patterns reduce the risk of cache poisoning by ensuring that cache keys are deterministic, validated, and isolated. They also demonstrate how to safely integrate DynamoDB with FastAPI while preserving security boundaries. Continuous scanning with tools that support unauthenticated checks and API-aware analysis can help verify that such controls are correctly implemented in practice.

When evaluating your API, prefer solutions that combine spec-driven analysis with runtime checks, including unauthenticated scanning and specialized LLM/AI Security tests. This helps identify input handling and caching logic issues that could lead to cache poisoning or exposure of sensitive data.