Integrity Failures in Fastapi with Dynamodb

An integrity failure occurs when an application fails to prevent unauthorized changes to data, allowing attackers to modify, delete, or reorder information in ways that violate business rules. The combination of FastAPI and DynamoDB can inadvertently expose integrity risks when application-layer validation is incomplete and the database permissions are over-permissive.

In FastAPI, request bodies are often deserialized directly into Pydantic models. If these models do not enforce immutability of key fields such as identifiers, version numbers, or ownership attributes, an attacker can tamper with those values in transit. For example, changing an item ID in a URL parameter or modifying a user_id field in JSON to escalate privileges or access another user’s resources. Because DynamoDB is a NoSQL store, schema enforcement is the responsibility of the application; there is no native schema-level restriction that prevents an update to a primary key or an unexpected attribute change.

DynamoDB’s conditional writes provide a mechanism to enforce integrity, but they are only effective if the application uses them. Without a condition expression that checks the current state of an item—such as a version number or an expected owner—concurrent updates or malicious requests can overwrite valid data. A FastAPI route that builds an update expression without a condition is vulnerable to lost updates and race conditions. For instance, two clients reading the same item simultaneously may both apply their changes, with the second write silently overwriting the first.

Another vector arises from insufficient authorization checks before invoking DynamoDB operations. If a FastAPI endpoint derives identifiers (such as a resource ID) from user input and passes them directly to a DynamoDB update_item or delete_item call, an attacker can modify the ID to target another resource. Because DynamoDB does not understand application-level ownership, it will execute the operation as long as the key attributes match. This is a BOLA/IDOR pattern that integrity controls must explicitly address.

Integrity failures also manifest in how data is represented in responses. If a FastAPI serializer inadvertently exposes internal state or mutable fields, clients may rely on that data and submit it back unchanged, while an attacker alters it on the server. Proper read models should separate internal state from external representation, and writes should validate that the submitted state transitions are allowed.

When scanning this stack with middleBrick, findings often highlight missing condition expressions in DynamoDB update operations and missing ownership checks in FastAPI route handlers. These scans test the unauthenticated attack surface and can identify endpoints where integrity controls are absent, providing prioritized findings with severity levels and remediation guidance mapped to frameworks such as OWASP API Top 10.

Remediation centers on enforcing immutability of identifiers, using conditional writes, and validating ownership on every request. Below are concrete patterns for FastAPI with DynamoDB that preserve integrity.

1. Enforce immutable identifiers and ownership

Never allow client-supplied IDs to dictate which item to modify. Derive the key from authenticated context and validate ownership explicitly.

from fastapi import Depends, HTTPException, status
from pydantic import BaseModel
import boto3
from botocore.exceptions import ClientError

# Assume authentication sets current_user sub
class Item(BaseModel):
    item_id: str
    owner_id: str
    data: str

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table_name = 'Items'

def get_current_user_id() -> str:
    # In practice, this would decode a token
    return "user_abc_123"

async def get_item(item_id: str):
    table = dynamodb.Table(table_name)
    try:
        resp = table.get_item(Key={'item_id': item_id})
        item = resp.get('Item')
        if not item:
            raise HTTPException(status_code=404, detail="Item not found")
        return item
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))

async def ensure_owns_item(item_id: str, user_id: str):
    item = await get_item(item_id)
    if item.get('owner_id') != user_id:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden: ownership mismatch")

2. Use conditional writes for updates

Apply a version attribute or an expected state to prevent overwrites. This is a core defense against lost updates.

async def update_item_conditional(item_id: str, new_data: str, expected_version: int):
    table = dynamodb.Table(table_name)
    try:
        resp = table.update_item(
            Key={'item_id': item_id},
            UpdateExpression='SET data = :d, version = version + :inc',
            ConditionExpression='version = :expected_version',
            ExpressionAttributeValues={
                ':d': new_data,
                ':inc': 1,
                ':expected_version': expected_version
            },
            ReturnValues='UPDATED_NEW'
        )
        return resp['Attributes']
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            raise HTTPException(status_code=409, detail="Item was modified concurrently")
        raise HTTPException(status_code=500, detail=str(e))

3. Validate and sanitize inputs before DynamoDB operations

Treat all user input as untrusted. Use Pydantic to validate types and ranges, and avoid passing raw input into key expressions.

from fastapi import Query

async def safe_get(item_id: str = Query(..., min_length=1, max_length=36)):
    # Enforce format constraints before touching DynamoDB
    if not item_id.isalnum():
        raise HTTPException(status_code=400, detail="Invalid item_id format")
    return await get_item(item_id)

4. Use transactions for multi-item integrity

When multiple items must remain consistent, use DynamoDB transactions and include ownership checks in the condition expressions.

async def transfer_item(item_id: str, new_owner_id: str, requester_id: str):
    table = dynamodb.Table(table_name)
    try:
        # Ensure requester owns the item before transferring
        await ensure_owns_item(item_id, requester_id)
        response = table.transact_write_items(
            TransactItems=[
                {
                    'Update': {
                        'TableName': table_name,
                        'Key': {'item_id': {'S': item_id}},
                        'UpdateExpression': 'SET owner_id = :new_owner',
                        'ConditionExpression': 'owner_id = :current_owner',
                        'ExpressionAttributeValues': {
                            ':new_owner': {'S': new_owner_id},
                            ':current_owner': {'S': requester_id}
                        }
                    }
                }
            ]
        )
        return response
    except ClientError as e:
        raise HTTPException(status_code=409, detail=str(e))

By combining strict ownership validation, conditional writes, and transactions, the FastAPI and DynamoDB stack can maintain data integrity even under malicious input. middleBrick scans can verify that condition expressions exist and that endpoints check ownership, surfacing any integrity gaps with actionable remediation guidance.