Timing Attack in Fastapi with Dynamodb

A timing attack in a Fastapi application that uses DynamoDB can occur when response times vary based on secret-dependent branches or operations. In Fastapi, endpoint handlers are often written to compare user-supplied tokens or query parameters against stored values using Python comparisons. If these comparisons are not constant-time—for example, using == on strings or short-circuiting logical checks—an attacker can measure round-trip times to infer information such as a valid user ID prefix or a HMAC signature byte by byte.

DynamoDB itself does not introduce a timing side channel for read operations like GetItem or Query; latency is generally driven by network RTT, provisioned capacity, and item size. However, the application layer in Fastapi can introduce variability before issuing the DynamoDB request. For instance, if the code first performs a string-based comparison on a token to decide whether to call DynamoDB, the time taken before the call varies with how many initial bytes match. This creates a leak: an attacker sending many crafted requests can infer the correct token structure by observing response time distributions.

Consider a Fastapi endpoint that retrieves a user by an API key stored in DynamoDB. If the code retrieves a candidate item and then compares the provided key with the stored key using a naive equality check, the comparison may exit early on mismatch. Although DynamoDB read latency is relatively stable, the added Python comparison time adds jitter correlated with the match length. Repeated measurements allow an attacker to perform a statistical analysis to recover the key or its prefix. This is a classic timing attack vector enabled by application logic, not by DynamoDB’s wire protocol or response consistency.

Another realistic scenario involves paginated or filtered queries where the presence or absence of items affects timing. For example, a Fastapi handler might iterate over a result set and break early when a condition is met. The time to return differs depending on whether the item is found early or only at the end of a scan. While DynamoDB’s Scan or Query response times are influenced by data size and indexing, the added conditional break in Python code introduces a timing signal. Attackers can exploit this by issuing queries designed to observe timing differences, potentially inferring data existence or ordering.

To detect this class of issue, scans such as those performed by middleBrick evaluate the unauthenticated attack surface of Fastapi endpoints that interact with DynamoDB. They do not inspect internal code but measure whether response times correlate with attacker-controlled inputs, flagging endpoints where timing-sensitive logic may exist. Findings include severity assessments and remediation guidance mapped to frameworks such as OWASP API Top 10, helping teams prioritize fixes.

Remediation focuses on ensuring that all operations that could depend on secret values complete in constant time relative to those values, and that DynamoDB interactions do not introduce observable timing variations via application logic. Below are concrete, realistic code examples for Fastapi that implement safe patterns.

1. Constant-time comparison for tokens or keys

Instead of using Python’s ==, use a constant-time comparison routine when comparing secrets or identifiers that could be subject to inference. For example:

import secrets
def safe_compare(val1: str, val2: str) -> bool:
    # secrets.compare_digest performs a constant-time comparison
    return secrets.compare_digest(val1, val2)

# Usage in a Fastapi handler
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()

def get_user_by_key(api_key: str):
    # Assume client is a DynamoDB resource
    table = client.Table("users")
    response = table.scan(
        FilterExpression=Attr("api_key").eq(api_key)
    )
    return response.get("Items", [])

@app.get("/user")
async def read_user(input_key: str):
    candidates = get_user_by_key(input_key)
    if not candidates:
        raise HTTPException(status_code=404, detail="Not found")
    stored = candidates[0]["api_key"]
    if not safe_compare(input_key, stored):
        raise HTTPException(status_code=403, detail="Forbidden")
    return {"user": candidates[0]}

2. Avoid early branching based on presence

Ensure that control flow does not reveal item existence via timing differences. Instead of breaking early, process items in a uniform manner or use existence checks that do not short-circuit in a way dependent on data values.

import boto3
from boto3.dynamodb.conditions import Attr

client = boto3.resource("dynamodb")

def has_item_with_key(table_name: str, key_name: str, key_value: str) -> bool:
    table = client.Table(table_name)
    response = table.scan(
        FilterExpression=Attr(key_name).eq(key_value)
    )
    # Always iterate to consume capacity and keep timing stable
    count = 0
    for _ in response.get("Items", []):
        count += 1
    # Return boolean without branching on data-dependent paths
    return count > 0

# In Fastapi, use with constant-time handling
@app.get("/check")
async def check_item(key: str):
    exists = has_item_with_key("items", "sku", key)
    # Perform a dummy constant-time operation to mask timing
    dummy = secrets.compare_digest(key, key[::-1]) if not exists else True
    return {"exists": exists}

3. Parameterized queries and prepared statements

Use DynamoDB’s expression attribute values to avoid injecting variable-length strings into key conditions that could affect serialization timing. Construct queries with placeholders and bind values separately to keep the request path stable.

import boto3

client = boto3.client("dynamodb")

def get_item_by_id(table_name: str, partition_key: str, sort_key: str = None):
    key = {
        "pk": {"S": partition_key}
    }
    if sort_key is not None:
        key["sk"] = {"S": sort_key}
    response = client.get_item(
        TableName=table_name,
        Key=key
    )
    return response.get("Item")

# Fastapi route using stable key construction
@app.get("/item")
async def get_item(pk: str, sk: str = None):
    item = get_item_by_id("mytable", pk, sk)
    if not item:
        raise HTTPException(status_code=404, detail="Not found")
    return item

4. Consistent error handling and response shapes

Return uniform response times and shapes for success and error cases where possible. Avoid raising different exception types or varying serialization workloads based on secret values. This reduces timing signals available to an attacker.

from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    # Ensure error responses take comparable time to serialize
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail}
    )