Email Injection in Fastapi with Dynamodb

Email injection occurs when user-controlled data is concatenated into email headers or commands without validation, enabling attackers to inject additional headers such as CC, BCC, or Subject. In a Fastapi application that stores or processes user input destined for email workflows and uses DynamoDB as a persistence layer, the risk arises when untrusted data is written to DynamoDB and later retrieved for inclusion in email operations.

Consider a Fastapi endpoint that accepts a user profile including an email address and stores it in DynamoDB using the AWS SDK for Python (Boto3). If the application later reads the item from DynamoDB and uses the email field in an SMTP or email-sending library without sanitization, newlines or control characters in the stored email can be leveraged to inject malicious headers. For example, a value like [email protected]\r\nCC: [email protected] stored in DynamoDB may appear harmless at storage time but becomes dangerous when passed verbatim to an email library that interprets line breaks as header separators.

DynamoDB itself does not introduce injection in the SQL sense, but it can inadvertently store attacker-controlled payloads that later surface in unsafe contexts. Because Fastapi often serializes DynamoDB items into JSON responses or passes them to downstream services, any unchecked usage of these fields in email composition can lead to header injection. This combination therefore exposes a classic injection vector through persistence (DynamoDB), transport (application logic in Fastapi), and side-channel execution (email delivery), violating proper separation of data and commands.

Real-world email injection can enable spoofed recipients, header smuggling, or phishing payloads. While this is not a remote code execution on the database, it can compromise trust in communication channels and facilitate social engineering. middleBrick scans this attack surface by checking input validation and data exposure across the stack, highlighting where data from DynamoDB may reach email-generation code without adequate sanitization or encoding.

To mitigate email injection when using DynamoDB with Fastapi, validate and sanitize data at the point of entry and again before using stored values in email contexts. Use allowlists for email formats, enforce length limits, and avoid directly interpolating user input into email headers. Below are concrete code examples that demonstrate a safe pattern.

1. Store validated data in DynamoDB

Validate the email with a strict regex and use parameterized DynamoDB operations. This example uses Boto3 with a typed DynamoDB resource.

import re
from typing import Dict
import boto3
from pydantic import BaseModel, EmailStr, ValidationError

class UserProfile(BaseModel):
    user_id: str
    email: EmailStr  # Pydantic ensures a valid email format

def store_user_profile(profile: UserProfile) -> None:
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.Table('UserProfiles')
    item = {
        'user_id': profile.user_id,
        'email': profile.email,
        'created_at': datetime.utcnow().isoformat()
    }
    table.put_item(Item=item)

# In your Fastapi route:
from fastapi import Fastapi, HTTPException
app = Fastapi()

@app.post('/users/')
def create_user(body: UserProfile):
    try:
        store_user_profile(body)
        return {'status': 'ok'}
    except ValidationError:
        raise HTTPException(status_code=422, detail='Invalid email format')

2. Retrieve and use data safely for email operations

When reading from DynamoDB for email sending, re-validate and encode headers. Use a dedicated email library that separates headers from content, and never trust stored values.

import smtplib
from email.message import EmailMessage
import boto3

def send_welcome_email(user_id: str) -> None:
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.Table('UserProfiles')
    resp = table.get_item(Key={'user_id': user_id})
    item = resp.get('Item')
    if not item:
        raise ValueError('User not found')

    # Re-validate on retrieval
    email = item.get('email', '')
    if '@' not in email or '\r' in email or '\n' in email:
        raise ValueError('Invalid email for sending')

    msg = EmailMessage()
    msg['Subject'] = 'Welcome'
    msg['From'] = '[email protected]'
    msg['To'] = email
    msg.set_content('Welcome to our service.')

    with smtplib.SMTP('localhost') as smtp:
        smtp.send_message(msg)

3. Defense-in-depth measures

• Use middleware in Fastapi to reject requests containing newlines in email fields.
• Apply DynamoDB ConditionExpressions to enforce format constraints at the database layer when possible.
• Ensure email-sending libraries are configured to use strict header parsing and do not allow raw header injection.

middleBrick can help by scanning your Fastapi endpoints and DynamoDB interaction patterns for input validation gaps and data exposure risks, providing prioritized findings and remediation guidance aligned with frameworks such as OWASP API Top 10.