Email Injection in Fastapi

Email injection in Fastapi applications typically occurs when user input is incorporated into email headers without proper validation. Fastapi's async nature and dependency injection system create specific attack vectors that developers must understand.

The most common manifestation appears in contact forms and user registration endpoints. Consider a Fastapi endpoint that accepts a contact form submission:

from fastapi import FastAPI, HTTPException
from email.mime.text import MIMEText
import smtplib

app = FastAPI()

@app.post("/contact")
async def contact_form(data: dict):
    name = data.get('name', '')
    email = data.get('email', '')
    message = data.get('message', '')
    
    # Vulnerable: direct header injection
    headers = f"From: {name} <{email}>\r\n"
    msg = MIMEText(message)
    msg['Subject'] = 'Contact Form'
    msg['From'] = email
    msg['To'] = 'admin@example.com'
    
    with smtplib.SMTP('localhost') as server:
        server.sendmail(email, 'admin@example.com', msg.as_string())
    
    return {"status": "message sent"}

The vulnerability here is that an attacker can inject newline characters into the email field, allowing them to add arbitrary headers. For example, submitting:

{
    "name": "Attacker",
    "email": "attacker@example.com\r\nBcc: victim@example.com\r\n",
    "message": "Hello"
}

This would result in the email being sent to both the admin and the victim without the admin's knowledge. Fastapi's Pydantic models don't automatically sanitize these inputs, making this a common oversight.

Another Fastapi-specific pattern involves dependency injection. Developers often create reusable email services as dependencies:

from fastapi import Depends

class EmailService:
    def __init__(self):
        self.server = smtplib.SMTP('localhost')
    
    async def send_contact_email(self, name: str, email: str, message: str):
        # Vulnerable to header injection
        msg = MIMEText(message)
        msg['From'] = email
        msg['To'] = 'admin@example.com'
        msg['Subject'] = 'Contact Form'
        
        self.server.sendmail(email, 'admin@example.com', msg.as_string())

async def get_email_service():
    return EmailService()

@app.post("/contact")
async def contact_form(data: dict, email_service: EmailService = Depends(get_email_service)):
    await email_service.send_contact_email(
        data.get('name', ''),
        data.get('email', ''),
        data.get('message', '')
    )
    return {"status": "message sent"}

The dependency injection pattern makes the vulnerability reusable across multiple endpoints, potentially amplifying the impact.

Detecting email injection in Fastapi applications requires both static analysis and runtime scanning. The vulnerability manifests in specific patterns that security scanners must recognize.

Static analysis should look for these Fastapi-specific patterns:

# Pattern 1: Direct header construction from user input
headers = f"From: {user_input}\r\n"

# Pattern 2: Pydantic model fields used in email headers
class ContactForm(BaseModel):
    name: str
    email: EmailStr  # Still vulnerable despite type validation
    message: str

# Pattern 3: Dependency injection of email services
class EmailService:
    async def send(self, to: str, subject: str, body: str):
        # Vulnerable if any parameter contains \r or \n
# Pattern 4: Dynamic header construction
msg['From'] = f"{name} <{email}>"

Runtime scanning with middleBrick specifically targets these Fastapi patterns. The scanner tests for header injection by sending payloads containing newline characters and observing whether additional headers are processed:

# middleBrick test payload for email injection
{
    "name": "Test",
    "email": "test@example.com\r\nBcc: bcc@example.com\r\n",
    "message": "Test message"
}

The scanner analyzes the response to determine if the email was sent to unintended recipients. middleBrick's black-box scanning approach is particularly effective here because it tests the actual runtime behavior without requiring source code access.

Fastapi's async nature creates additional detection challenges. Email injection might occur in async functions where the email sending happens in the background:

async def send_email_background(to: str, subject: str, body: str):
    # This async function might mask injection vulnerabilities
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'noreply@example.com'
    msg['To'] = to
    
    with smtplib.SMTP('localhost') as server:
        await asyncio.to_thread(server.sendmail, 'noreply@example.com', to, msg.as_string())

middleBrick's scanning engine detects these patterns by analyzing the API's behavior under various input conditions, identifying when newline characters in email fields lead to unexpected email routing.

Remediating email injection in Fastapi requires a defense-in-depth approach that leverages Fastapi's features and Python's email handling capabilities.

The most effective approach is input sanitization before header construction. Fastapi developers should create reusable sanitization utilities:

import re
from fastapi import HTTPException

def sanitize_email_input(value: str) -> str:
    """Remove any newline characters and potential header injection attempts"""
    if '\r' in value or '\n' in value:
        raise HTTPException(
            status_code=400,
            detail="Invalid characters in email field"
        )
    return value.strip()

def sanitize_name_input(value: str) -> str:
    """Remove any characters that could be used for header injection"""
    if '\r' in value or '\n' in value:
        raise HTTPException(
            status_code=400,
            detail="Invalid characters in name field"
        )
    # Remove any non-printable characters
    return re.sub(r'[\x00-\x1F\x7F]', '', value)

Integrate these sanitizers into your Fastapi endpoints:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr

app = FastAPI()

class ContactForm(BaseModel):
    name: str
    email: EmailStr
    message: str

@app.post("/contact")
async def contact_form(data: ContactForm):
    # Sanitize inputs
    name = sanitize_name_input(data.name)
    email = sanitize_email_input(data.email)
    message = data.message
    
    # Use email.utils to properly format addresses
    from_email = email.utils.formataddr((name, email))
    
    msg = MIMEText(message)
    msg['Subject'] = 'Contact Form'
    msg['From'] = from_email
    msg['To'] = 'admin@example.com'
    
    with smtplib.SMTP('localhost') as server:
        server.sendmail(from_email, 'admin@example.com', msg.as_string())
    
    return {"status": "message sent"}

For dependency injection scenarios, create a secure email service wrapper:

from email.utils import formataddr

class SecureEmailService:
    def __init__(self):
        self.server = smtplib.SMTP('localhost')
    
    async def send_contact_email(self, name: str, email: str, message: str):
        name = sanitize_name_input(name)
        email = sanitize_email_input(email)
        
        from_email = formataddr((name, email))
        
        msg = MIMEText(message)
        msg['Subject'] = 'Contact Form'
        msg['From'] = from_email
        msg['To'] = 'admin@example.com'
        
        self.server.sendmail(from_email, 'admin@example.com', msg.as_string())

Fastapi's exception handling can be used to provide consistent error responses:

from fastapi import HTTPException

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail}
    )

For production applications, consider using email service libraries that automatically handle header sanitization, such as SendGrid or Amazon SES, which provide Fastapi integrations and handle these security concerns at the API level.