Insecure Design in Fastapi with Dynamodb

An insecure design in a FastAPI service that uses DynamoDB often stems from modeling data and access patterns that make authorization mistakes easy to introduce. DynamoDB’s key-based model encourages designs where a single partition key (for example, PK = USER#123) serves as the primary shard for a user’s records. If FastAPI endpoints derive that key from request input without strict ownership checks, the design can conflate identity with authorization, enabling BOLA/IDOR across what appears to be a simple key-value lookup.

Consider a design where the client supplies an item_id and the server constructs a DynamoDB key such as PK = ITEM#{item_id} and SK = OWNER#{user_id}. If the server neglects to validate that the authenticated user matches the SK prefix, the endpoint may unintentionally expose items belonging to other users. This is a classic BOLA flaw embedded in the data model: the primary key structure makes it trivial to request any item by ID, assuming a weak or missing authorization step.

Another insecure pattern arises when FastAPI query parameters are directly mapped to DynamoDB filter expressions without canonical normalization. A design that allows free-text filtering on attributes like status or tenant_id can lead to privilege escalation if the tenant boundary is enforced only by application logic rather than by the key structure itself. For example, an attacker could supply a filter[tenant_id] that references another tenant’s data, and a permissive scan may not flag that the backend issues a Query without a strict partition-key equality condition.

DynamoDB’s sparse index design can also encourage insecure designs. If a table uses a Global Secondary Index (GSI) to support queries by non-key attributes, and FastAPI relies on the GSI without ensuring that the base table’s partition key still enforces ownership, the index can return items across partitions in ways that bypass intended tenant isolation. This becomes especially risky when combined with unauthenticated or weakly authenticated LLM endpoints that expose query-building functionality, increasing the attack surface for data exposure through indirect paths.

Additionally, the combination of FastAPI’s automatic request binding and DynamoDB’s conditional writes can create subtle authorization gaps if optimistic concurrency checks are omitted. A design that updates an item based solely on user input version tokens without re-validating ownership on read-modify-write cycles may allow overwriting or deleting another user’s data. Each of these patterns illustrates how an insecure design couples data modeling choices in DynamoDB with endpoint behaviors in FastAPI, turning what appears to be a straightforward key-based store into a vector for IDOR and privilege escalation.

Remediation centers on enforcing ownership at the data-access layer and normalizing inputs before constructing DynamoDB keys and expressions. Always derive partition and sort keys from authenticated identity, and avoid passing user-controlled IDs directly into query key structures without ownership validation.

1. Enforce ownership via composite keys

Design your DynamoDB key schema so that the partition key includes the user scope. For example, use PK = USER#<user_id> and SK = ITEM#<item_id>. This ensures that a query for items must include the user prefix, making it difficult to accidentally fetch another user’s items.

from fastapi import Depends, HTTPException, status
import boto3
from pydantic import BaseModel

class Item(BaseModel):
    item_id: str
    data: str

def get_current_user() -> str:
    # Replace with your auth logic (e.g., JWT, session)
    return "user-abc-123"

def get_ddb_client():
    return boto3.client("dynamodb", region_name="us-east-1")

@app.get("/items/{item_id}")
def read_item(item_id: str, user_id: str = Depends(get_current_user), ddb=get_ddb_client()):
    pk = f"USER#{user_id}"
    sk = f"ITEM#{item_id}"
    resp = ddb.get_item(
        TableName="AppTable",
        Key={"PK": {"S": pk}, "SK": {"S": sk}}
    )
    item = resp.get("Item")
    if not item:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found or access denied")
    return {"item_id": item_id, "data": item.get("data", {}).get("S")}

2. Use KeyConditionExpression with partition-key equality

When querying, always include the authenticated user as the partition key equality condition. Do not allow user input to dictate the partition key value.

@app.get("/items")
def list_items(user_id: str = Depends(get_current_user), ddb=get_ddb_client()):
    pk = f"USER#{user_id}"
    resp = ddb.query(
        TableName="AppTable",
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={
            ":pk": {"S": pk}
        }
    )
    items = resp.get("Items", [])
    return {"items": [{"item_id": i["SK"]["S"].replace("ITEM#", "")} for i in items]}

3. Normalize and validate inputs before use

Never directly interpolate user input into key names or filter expressions. Validate and sanitize values, and prefer parameterized expressions over string concatenation to avoid injection and malformed keys.

from typing import Optional

@app.get("/search")
def search_items(
    user_id: str = Depends(get_current_user),
    status: Optional[str] = None,
    ddb=get_ddb_client()
):
    pk = f"USER#{user_id}"
    filter_exp = None
    expr_vals = {":pk": {"S": pk}}
    if status:
        # Validate status against an allowlist in production
        filter_exp = "status = :status"
        expr_vals[":status"] = {"S": status}
    resp = ddb.scan(
        TableName="AppTable",
        FilterExpression=filter_exp,
        ExpressionAttributeValues=expr_vals
    )
    return {"items": resp.get("Items", [])}

4. Enforce tenant boundaries with attribute checks

If multi-tenancy is required, include the tenant ID in the key and enforce it on every request. Do not rely solely on client-supplied tenant identifiers for filtering.

def get_current_tenant() -> str:
    return "tenant-xyz"

@app.get("/admin/reports")
def admin_reports(
    user_id: str = Depends(get_current_user),
    tenant_id: str = Depends(get_current_tenant),
    ddb=get_ddb_client()
):
    # Ensure the caller belongs to the requested tenant
    pk = f"TENANT#{tenant_id}"
    sk = f"REPORT#METADATA"
    resp = ddb.get_item(
        TableName="AppTable",
        Key={"PK": {"S": pk}, "SK": {"S": sk}}
    )
    if not resp.get("Item"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Tenant access denied")
    # Proceed with tenant-scoped operations
    return {"tenant_id": tenant_id}

5. Avoid unauthenticated endpoints for sensitive operations

Unauthenticated or LLM-facing endpoints that construct DynamoDB queries should be limited to non-sensitive data or require additional authorization. Prefer authenticated flows and apply the same key design principles even in helper utilities.

6. Use condition expressions for safe updates

When modifying items, include a version attribute and re-validate ownership on write to prevent race conditions and accidental cross-user modifications.

class UpdateItem(BaseModel):
    item_id: str
    data: str
    version: int

@app.put("/items/{item_id}")
def update_item(
    item_id: str,
    payload: UpdateItem,
    user_id: str = Depends(get_current_user),
    ddb=get_ddb_client()
):
    pk = f"USER#{user_id}"
    sk = f"ITEM#{item_id}"
    resp = ddb.put_item(
        TableName="AppTable",
        Item={
            "PK": {"S": pk},
            "SK": {"S": sk},
            "data": {"S": payload.data},
            "version": {"N": str(payload.version)}
        },
        ConditionExpression="attribute_not_exists(#v) OR #v = :version",
        ExpressionAttributeNames={"#v": "version"},
        ExpressionAttributeValues={
            ":version": {"N": str(payload.version)}
        }
    )
    return {"status": "updated"}

These patterns emphasize that remediation in this context is about design discipline: constrain keys to user scope, validate inputs, and avoid implicit trust in client-supplied identifiers.