Ssrf in Fastapi with Dynamodb

Server-Side Request Forgery (SSRF) in a FastAPI application that interacts with DynamoDB can occur when the application accepts attacker-controlled input and uses it to form requests to internal or external services before issuing DynamoDB operations. For example, an endpoint that takes a URL or host parameter to build a DynamoDB condition expression, construct a pre-signed URL, or select a DynamoDB table name can be abused to make the server reach unintended destinations. Because DynamoDB is often used as a primary data store, SSRF can lead to metadata service access (e.g., the instance metadata service on EC2), internal services not exposed publicly, or external services, potentially bypassing network-level segregation.

In FastAPI, routes that directly concatenate user input into boto3 calls or configuration used by boto3 can expose SSRF. A common pattern is an endpoint that accepts a table name or a condition value from the client and passes it to DynamoDB without strict allowlisting. If the application also uses dynamic endpoints, instance metadata, or Lambda environment variables that can be influenced by the request, SSRF may allow an attacker to pivot from the FastAPI layer to sensitive internal resources. Even when DynamoDB responses are returned to the client, SSRF remains relevant because the risk lies in the server-initiated outbound connections, not the data returned from DynamoDB itself.

An illustrative vulnerable FastAPI route:

from fastapi import FastAPI, Query
import boto3
from botocore.exceptions import ClientError

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

@app.get('/items/')
async def get_item(table_name: str = Query(...), key: str = Query(...)):
    table = dynamodb.Table(table_name)
    response = table.get_item(Key={'id': key})
    return response.get('Item', {})

In this example, an attacker can supply a table name that points to an AWS metadata service URL (e.g., http://169.254.169.254/latest/meta-data/iam/security-credentials/) if the application is misconfigured to allow such values, or abuse parameter parsing to trigger SSRF through custom logic that uses the input to form HTTP requests. Even when DynamoDB is the target, allowing arbitrary table names bypasses intended access controls and may expose sensitive configurations or enable data exfiltration via side channels.

To detect SSRF in this context, middleBrick scans the unauthenticated attack surface of the FastAPI endpoint, including OpenAPI/Swagger specifications with full $ref resolution, and cross-references runtime behavior with spec definitions. It flags parameters that can influence outbound requests made by the server, such as fields used to construct table names, object keys, or conditional expressions that may be interpreted as URLs or hostnames by underlying libraries.

Remediation focuses on strict input validation, allowlisting, and avoiding dynamic construction of resources from client input. For DynamoDB in FastAPI, never use raw user input as a table name or key without validation. Use an allowlist of known table names or map user-friendly identifiers to table names on the server. Also avoid using input to form hostnames or URLs that the server will request.

Secure FastAPI route with table name allowlist:

from fastapi import FastAPI, Query, HTTPException
import boto3
from botocore.exceptions import ClientError

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

ALLOWED_TABLES = {'users', 'orders', 'products'}

def get_table(table_name: str):
    if table_name not in ALLOWED_TABLES:
        raise HTTPException(status_code=400, detail='Invalid table name')
    return dynamodb.Table(table_name)

@app.get('/items/')
async def get_item(table_name: str = Query(...), key: str = Query(...)):
    table = get_table(table_name)
    try:
        response = table.get_item(Key={'id': key})
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
    item = response.get('Item')
    if not item:
        raise HTTPException(status_code=404, detail='Item not found')
    return item

When constructing keys, validate key formats instead of passing raw input. For example, if IDs must be integers, enforce conversion and range checks:

@app.get('/users/{user_id}')
async def get_user(user_id: int):
    if user_id <= 0:
        raise HTTPException(status_code=400, detail='Invalid user ID')
    table = dynamodb.Table('users')
    response = table.get_item(Key={'user_id': str(user_id)})
    return response.get('Item', {})

If you need to work with multiple logical data sets, use a mapping rather than trusting the client to specify the table:

TABLE_MAP = {
    'candidates': 'users',
    'orders': 'orders',
    'inventory': 'products'
}

def resolve_table(alias: str):
    real_name = TABLE_MAP.get(alias)
    if not real_name:
        raise HTTPException(status_code=400, detail='Unsupported dataset')
    return dynamodb.Table(real_name)

For operations that require condition expressions, avoid building expressions from raw strings. Use DynamoDB’s built-in condition expressions with placeholder values:

from boto3.dynamodb.conditions import Attr

@app.get('/secure-items/')
async def secure_list(table_name: str = Query(...), owner_id: str = Query(...)):
    table = get_table(table_name)
    try:
        response = table.scan(
            FilterExpression=Attr('owner_id').eq(owner_id)
        )
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))
    return response.get('Items', [])

middleBrick’s scans include checks for SSRF by analyzing parameter flows that could influence outbound requests and cross-referencing these with OpenAPI/Swagger definitions. Its findings map to frameworks such as OWASP API Top 10 and provide prioritized remediation guidance, helping teams address SSRF in DynamoDB-integrated FastAPI services without requiring an agent or credentials.