Prototype Pollution in Fastapi with Dynamodb

Prototype pollution in a FastAPI application that uses DynamoDB typically arises when user-controlled input is merged into server-side objects that later influence how data is stored or retrieved from DynamoDB. In JavaScript/TypeScript ecosystems, objects are often mutated by copying user-supplied properties into a target object without validation, which can overwrite constructor or prototype properties such as __proto__, constructor, or prototype. If a FastAPI service deserializes JSON into a dict and then passes that dict to code (for example via an embedded scripting layer or a downstream Node.js service), an attacker can inject properties that affect object behavior at runtime. Even when the data ultimately reaches DynamoDB, pollution can affect how items are modeled: attribute names may be altered, type expectations violated, or conditional logic subverted.

With DynamoDB, the impact is less about runtime code execution on the database and more about how polluted data leads to incorrect authorization checks, malformed queries, or unsafe deserialization when items are reconstructed in application code. For example, an item retrieved from DynamoDB may be merged into a Python dict that is later used to build another request; if polluted keys like __class__ or __mro__ are present when objects are reconstructed or serialized, unexpected behavior can occur in frameworks that rely on object identity or inheritance. In FastAPI, common patterns such as dict.update(user_input) or using **kwargs to construct models can propagate polluted fields into DynamoDB operations if input is not strictly validated. Moreover, if the service stores user-controlled keys directly as DynamoDB attribute names, an attacker can force creation of attributes that interfere with application-level assumptions about item structure, leading to privilege escalation or data exposure when combined with BOLA/IDOR checks that rely on key-based access patterns.

Consider an endpoint that accepts filter criteria to query a DynamoDB table. If user input is shallow-merged into a condition expression or key condition without normalization, polluted properties can change the semantics of the query or bypass intended filters. While DynamoDB itself does not execute code, the surrounding FastAPI logic may construct queries based on polluted dicts, causing the service to expose other users’ items or skip essential authorization. This illustrates why validating and sanitizing all incoming data in FastAPI — and ensuring that DynamoDB interactions use strict attribute whitelisting — is essential to mitigate prototype pollution in this specific stack.

Remediation focuses on strict input validation, avoiding unsafe merges, and ensuring DynamoDB interactions use explicit attribute handling. In FastAPI, prefer Pydantic models for deserialization and avoid passing raw user dicts directly into DynamoDB operations. When building DynamoDB expressions, use explicit attribute names and never directly interpolate user keys.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import boto3
from botocore.exceptions import ClientError

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

class ItemCreate(BaseModel):
    user_id: str = Field(..., min_length=1)
    name: str = Field(..., min_length=1, max_length=100)
    value: int = Field(..., ge=0)

@app.post('/items')
def create_item(data: ItemCreate):
    # Safe: explicit fields, no raw dict merge
    try:
        table.put_item(Item={
            'user_id': data.user_id,
            'name': data.name,
            'value': data.value
        })
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {'status': 'ok'}

@app.get('/items/{user_id}')
def list_items(user_id: str, name: str):
    # Safe: explicit key condition, no user-controlled attribute names
    try:
        resp = table.query(
            KeyConditionExpression='user_id = :uid AND begins_with(name, :nm)',
            ExpressionAttributeValues={':uid': user_id, ':nm': name}
        )
        return resp.get('Items', [])
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))

For operations that require dynamic attributes, validate attribute names against a strict allowlist and reject any keys that match prototype pollution patterns (e.g., __proto__, constructor, prototype). Avoid using dict.update() with untrusted data; instead, construct new dicts with known-safe keys.

ALLOWED_FILTERS = {'name', 'value', 'created_at'}

def safe_filter(user_filters: dict) -> dict:
    cleaned = {}
    for k, v in user_filters.items():
        if k in ALLOWED_FILTERS:
            cleaned[k] = v
    return cleaned

Log and monitor queries that include unexpected attributes as part of your security telemetry. By combining Pydantic validation, explicit DynamoDB key expressions, and allowlist-based filtering, you reduce the risk that prototype pollution can affect item modeling or query logic in a FastAPI + DynamoDB stack.