Prototype Pollution in Flask with Dynamodb

Prototype pollution in a Flask application that uses DynamoDB typically occurs when user-controlled input is merged into server-side objects used to construct or manipulate DynamoDB requests. Because JavaScript prototypes are not a concern in Python, the risk in this stack centers on object injection into request building, query construction, or intermediate dictionaries that are later passed to the DynamoDB client. An attacker can inject keys that affect behavior elsewhere in the application, such as conditional logic, serialization, or schema assumptions, rather than directly corrupting JavaScript prototypes.

Consider a Flask route that builds a DynamoDB UpdateItem expression using user input without validation:

import boto3
from flask import Flask, request

app = Flask(__name__)
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')

@app.route('/update-profile', methods=['POST'])
def update_profile():
    user_id = request.json.get('user_id')
    updates = request.json.get('updates', {})  # attacker-controlled
    table = dynamodb.Table('users')

    # Unsafe: directly using attacker-controlled keys in expression attribute names/values
    update_expression = 'SET '
    expression_attrs = {}
    for key, value in updates.items():
        attr_name = f'#{key}'
        update_expression += f'{attr_name} = :{key}, '
        expression_attrs[f':{key}'] = value
        expression_attrs[attr_name] = key
    update_expression = update_expression.rstrip(', ')

    table.update_item(
        Key={'user_id': user_id},
        UpdateExpression=update_expression,
        ExpressionAttributeNames=expression_attrs
    )
    return {'status': 'ok'}

If the updates dictionary contains keys like __proto__, constructor, or other metadata fields, an attacker may attempt to influence application logic that depends on how these dictionaries are later interpreted. For example, if the application serializes objects or uses dynamic attribute access on merged dictionaries, injected keys can change behavior in unexpected ways. While DynamoDB itself does not interpret prototypes, the surrounding Python logic might—especially when using dynamic attribute access, **kwargs unpacking, or frameworks that rely on object inheritance chains.

Another common pattern involves using request data to build condition expressions or filter logic:

@app.route('/search', methods=['GET'])
def search_users():
    filters = request.args.to_dict()  # attacker-controlled query parameters
    # Unsafe: using untrusted keys in expression building
    filter_exp = ' AND '.join([f'{k} = :{k}' for k in filters.keys()])
    expression_attrs = {f':{k}': v for k, v in filters.items()}
    # Assume table.scan with FilterExpression built dynamically
    response = table.scan(FilterExpression=filter_exp, ExpressionAttributeValues=expression_attrs)
    return {'items': response['Items']}

Here, keys such as constructor or prototype do not affect DynamoDB but could affect downstream processing if the resulting expression strings are reused or logged in contexts where prototype-based attacks are possible (e.g., in a templating engine or JavaScript service consuming the data). The key takeaway is that prototype pollution in this stack is about unsafe object construction and dynamic expression building, not about JavaScript engine behavior.

Remediation focuses on strict input validation, avoiding dynamic expression construction, and isolating user input from control logic. Prefer static UpdateExpression templates and validate allowed fields against a whitelist. Never directly interpolate user keys into expression attribute names or values.

Safe example using a whitelist and parameterized updates:

ALLOWED_FIELDS = {'name', 'email', 'age', 'preferences'}

@app.route('/update-profile-safe', methods=['POST'])
def update_profile_safe():
    user_id = request.json.get('user_id')
    updates = request.json.get('updates', {})
    table = dynamodb.Table('users')

    # Validate and sanitize updates
    sanitized = {}
    expression_parts = []
    expression_attrs = {'#uid': 'user_id'}
    for idx, (key, value) in enumerate(updates.items()):
        if key not in ALLOWED_FIELDS:
            continue  # or return an error
        placeholder = f'#f{idx}'
        value_placeholder = f':v{idx}'
        expression_parts.append(f'{placeholder} = {value_placeholder}')
        expression_attrs[placeholder] = key
        expression_attrs[value_placeholder] = value

    if not expression_parts:
        return {'error': 'no valid fields'}, 400

    update_expression = 'SET ' + ', '.join(expression_parts)
    table.update_item(
        Key={'user_id': user_id},
        UpdateExpression=update_expression,
        ExpressionAttributeNames=expression_attrs,
        ExpressionAttributeValues=expression_attrs
    )
    return {'status': 'ok'}

For search and filter scenarios, avoid dynamic construction of expression strings from raw query parameters. Instead, map known filter keys to predefined field names and use parameterized values:

@app.route('/search-safe', methods=['GET'])
def search_users_safe():
    allowed_filters = {'name': 'name', 'email': 'email', 'age': 'age'}
    filters = request.args.to_dict()
    conditions = []
    attr_names = {'#name': 'name', '#email': 'email', '#age': 'age'}
    attr_values = {}
    for key, value in filters.items():
        if key not in allowed_filters:
            continue
        conditions.append(f'{attr_names["#" + key]} = :{key}')
        attr_values[f':{key}'] = value

    filter_expression = ' AND '.join(conditions) if conditions else None
    scan_kwargs = {}
    if filter_expression:
        scan_kwargs['FilterExpression'] = filter_expression
        scan_kwargs['ExpressionAttributeNames'] = attr_names
        scan_kwargs['ExpressionAttributeValues'] = attr_values

    response = table.scan(**scan_kwargs) if filter_expression else table.scan()
    return {'items': response['Items']}

Additional recommendations: use schema validation (e.g., Pydantic) for incoming JSON, enforce strict type checks, and avoid passing raw user input into any dynamic code generation or template rendering. Combine these measures with runtime security scanning to detect misconfigurations early.