Broken Access Control in Fastapi with Dynamodb

Broken Access Control occurs when authorization checks are missing or incorrectly enforced, allowing an authenticated user to access or modify resources that should be restricted. In a Fastapi application backed by DynamoDB, this often arises from trusting client-supplied identifiers (such as a resource ID or path parameter) without verifying that the authenticated subject owns or is permitted to operate on that resource. Because DynamoDB is a low-level, schema-flexible store, it does not enforce ownership or permissions by itself; it is the application layer in Fastapi that must implement and enforce these checks consistently.

Consider an endpoint designed to retrieve a user profile by ID. A vulnerable Fastapi route might read the ID directly from the request and query DynamoDB without confirming the ID belongs to the requesting user:

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

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

class UserProfile(BaseModel):
    user_id: str
    email: str
    display_name: str

@app.get('/profiles/{user_id}', response_model=UserProfile)
def get_profile(user_id: str, token_user_id: str = Depends(get_current_user_id)):
    response = table.get_item(Key={'user_id': user_id})
    item = response.get('Item')
    if not item:
        raise HTTPException(status_code=404, detail='Not found')
    # Missing: ensure token_user_id == user_id
    return UserProfile(**item)

In this example, an attacker can change the user_id path parameter to access any profile in the table, provided they know or can guess valid IDs. This is a classic BOLA (Broken Object Level Authorization) and IDOR pattern. DynamoDB does not raise an error for unauthorized reads; it simply returns the item if the key exists, so the responsibility to enforce ownership lies entirely with Fastapi code.

The same class of issue extends to write operations. An endpoint that updates or deletes a record based on a client-supplied key without validating that the record belongs to the caller can lead to privilege escalation or unintended data modification. For example, a Fastapi route that deletes a DynamoDB item by ID supplied in the body or URL, without checking that the item’s owner_id matches the authenticated subject, exposes a BFLA (Business Logic Flaw) and privilege escalation risk.

Additional risk patterns include missing authorization on related operations (e.g., listing resources the user shouldn’t see) and inconsistent enforcement where some endpoints check ownership and others do not. Because DynamoDB often stores denormalized or shared data models, it is easy to omit necessary checks when designing table structures, especially when composite keys are used. Without rigorous mapping between authentication subject and DynamoDB partition/sort keys in Fastapi, authorization gaps become likely.

Remediation centers on enforcing ownership and authorization on every request that interacts with DynamoDB, using the authenticated subject to constrain queries and updates. Below are concrete, secure patterns for Fastapi with DynamoDB resource and document models.

Secure read with ownership check

Always include the authenticated user identifier in the key condition and validate that the returned item matches the subject:

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

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

class UserProfile(BaseModel):
    user_id: str
    email: str
    display_name: str

def get_current_user_id() -> str:
    # implementation that extracts and validates identity
    return 'user-uuid-123'

@app.get('/profiles/me', response_model=UserProfile)
def get_my_profile(token_user_id: str = Depends(get_current_user_id)):
    response = table.get_item(Key={'user_id': token_user_id})
    item = response.get('Item')
    if not item:
        raise HTTPException(status_code=404, detail='Not found')
    return UserProfile(**item)

By using a scoped endpoint like /profiles/me and deriving the key from the authenticated subject, you avoid IDOR entirely. For multi-tenant or organization-based models, include the organization or tenant ID in both the key and the authentication context, and assert that they match before querying.

Secure write with ownership and existence checks

For updates and deletes, re-fetch the item and confirm ownership as part of a conditional write. Conditional writes in DynamoDB help prevent race conditions where permissions change between read and write:

from fastapi import Fastapi, Depends, HTTPException
import boto3
from boto3.dynamodb.conditions import Attr

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

class UpdateProfile(BaseModel):
    display_name: str

def get_current_user_id() -> str:
    return 'user-uuid-123'

@app.put('/profiles/me')
def update_my_profile(payload: UpdateProfile, token_user_id: str = Depends(get_current_user_id)):
    # Conditional update ensures item still belongs to the user at write time
    response = table.update_item(
        Key={'user_id': token_user_id},
        UpdateExpression='SET display_name = :val',
        ConditionExpression=Attr('user_id').eq(token_user_id),
        ExpressionAttributeValues={':val': payload.display_name},
        ReturnValues='UPDATED_NEW'
    )
    return {'message': 'updated', 'updated': response.get('Attributes')}

The ConditionExpression causes DynamoDB to reject the update if the item no longer has the expected owner, providing an additional safety net. For deletes, use the same pattern with table.delete_item(Key=..., ConditionExpression=...).

Organizational models and composite keys

When data is shared within teams or organizations, include a partition key that incorporates the organization or tenant ID, and enforce that the authenticated subject belongs to that tenant. For example, use a composite key like PK = ORG#org_id#USER#user_id and SK for sort attributes. In Fastapi, derive both parts from the token and validate that the requested resource’s key components match the subject’s organization membership before querying.

import boto3
from fastapi import Depends, HTTPException

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

def get_current_user() -> dict:
    # returns {'org_id': 'org-123', 'user_id': 'user-abc'}
    return {'org_id': 'org-123', 'user_id': 'user-abc'}

@app.get('/org/items/{item_id}')
def get_org_item(item_id: str, user: dict = Depends(get_current_user)):
    pk = f"ORG#{user['org_id']}#USER#{user['user_id']}"
    response = table.get_item(Key={'pk': pk, 'sk': item_id})
    item = response.get('Item')
    if not item:
        raise HTTPException(status_code=404, detail='Not found or access denied')
    return item

With this pattern, even if item_id is guessed, access is denied unless the item’s partition key matches the authenticated user’s organizational context. This aligns the DynamoDB data model with authorization boundaries enforced in Fastapi.