Log Injection in Fastapi

Log injection in Fastapi occurs when untrusted user input flows directly into log statements without proper sanitization. Fastapi's async nature and dependency injection system create unique injection points that developers often overlook.

The most common Fastapi-specific vector involves dependency injection parameters. Consider this vulnerable pattern:

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_user_agent(headers: dict = Depends()) -> str:
    return headers.get('user-agent', 'unknown')

@app.get('/api/data')
async def read_data(user_agent: str = Depends(get_user_agent)):
    logger.info(f'User agent: {user_agent}')
    return {'data': 'sensitive info'}

An attacker can craft a User-Agent header containing newline characters and log levels:

User-Agent: 
INFO - User agent: normal
ERROR - Critical error occurred

This injects a fake INFO log followed by an ERROR log, polluting the log stream and potentially triggering false alerts.

Fastapi's exception handlers create another injection point. When exceptions occur, Fastapi automatically logs the request body and headers:

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    logger.error(f'ValueError: {exc}')
    return JSONResponse(
        status_code=400,
        content={"message": str(exc)}
    )

If the exception message contains user input with newline characters, it breaks the log structure. Fastapi's automatic request logging also captures the full request body, which could contain malicious log injection payloads.

Background tasks in Fastapi introduce timing-based injection risks. When user input flows into background tasks that log asynchronously:

@app.post('/upload')
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    
    async def process_and_log():
        logger.info(f'Processing file: {contents.decode()}')
    
    await asyncio.create_task(process_and_log())
    return {'status': 'uploaded'}

The async logging can interleave with other log entries, making injection payloads harder to detect and correlate with the original request.

Detecting log injection in Fastapi requires examining both the application code and runtime behavior. Static analysis should focus on these Fastapi-specific patterns:

First, scan for direct string interpolation in log statements where user input flows through dependency injection or request parameters. middleBrick's Fastapi scanner automatically detects these patterns by analyzing the dependency injection graph and identifying parameters that could contain untrusted data.

middleBrick tests for log injection by sending payloads containing newline characters and log levels through all request parameters, headers, and body fields. The scanner monitors the application's log output (if accessible) or looks for indicators like:

• Unexpected log levels in the response headers
• Time-based discrepancies suggesting async log injection
• Corrupted log structure in any accessible log endpoints

For Fastapi applications using structured logging (like structlog or Loguru), middleBrick analyzes the logging configuration to identify vulnerable format strings. The scanner specifically checks for Fastapi's built-in logging behavior, including:

# Fastapi's default access log - vulnerable if user input flows here
# Example: INFO:     127.0.0.1:54321 - "GET /api/data HTTP/1.1" status: 200

middleBrick's LLM security module adds another layer of detection for Fastapi applications that serve AI endpoints. It tests for prompt injection that could manipulate logging behavior in LLM responses, a unique Fastapi vulnerability when using async AI libraries.

Runtime detection should monitor for:

• Log entries with unexpected timestamps or source IPs
• Log levels appearing in user-controlled fields
• Structured log corruption (missing fields, extra levels)

The middleBrick CLI tool can scan a running Fastapi instance with: middlebrick scan https://your-fastapi-app.com, providing a security score and specific findings about log injection vulnerabilities.

Remediating log injection in Fastapi requires both input sanitization and architectural changes. The most effective approach combines Fastapi's type system with explicit sanitization.

For dependency injection parameters, use Pydantic models with custom validators:

from pydantic import BaseModel, validator
import re

class SafeUserAgent(BaseModel):
    user_agent: str
    
    @validator('user_agent')
    def sanitize_user_agent(cls, v):
        # Remove newlines and carriage returns
        v = re.sub(r'[\n\r]', ' ', v)
        # Remove ANSI color codes
        v = re.sub(r'\x1B\[[0-9;]*[mK]', '', v)
        return v

async def get_safe_user_agent(headers: dict = Depends()) -> str:
    return SafeUserAgent(**headers).user_agent

For request body validation, Fastapi's Pydantic integration provides automatic sanitization:

from typing import Dict
from pydantic import BaseModel

class LogSafeData(BaseModel):
    message: str
    
    @validator('message')
    def sanitize_message(cls, v):
        return re.sub(r'[\n\r]', ' ', v)

@app.post('/api/log')
async def log_message(data: LogSafeData):
    logger.info(f'Message: {data.message}')
    return {'status': 'logged'}

Fastapi's exception handlers should sanitize exception messages before logging:

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    sanitized_msg = re.sub(r'[\n\r]', ' ', str(exc))
    logger.error(f'ValueError: {sanitized_msg}')
    return JSONResponse(
        status_code=400,
        content={"message": sanitized_msg}
    )

For background tasks, use Fastapi's built-in task system with proper context passing:

from fastapi import BackgroundTasks

@app.post('/upload')
async def upload_file(
    file: UploadFile = File(...),
    background_tasks: BackgroundTasks = Depends()
):
    contents = await file.read()
    
    def process_and_log():
        sanitized = re.sub(r'[\n\r]', ' ', contents.decode())
        logger.info(f'Processing file: {sanitized}')
    
    background_tasks.add_task(process_and_log)
    return {'status': 'uploaded'}

Implement structured logging with Fastapi middleware to ensure consistent log formatting:

from fastapi.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
import structlog

class LogSanitizingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        
        # Sanitize any user data before it reaches structured logging
        if 'user-agent' in request.headers:
            ua = request.headers['user-agent']
            sanitized_ua = re.sub(r'[\n\r]', ' ', ua)
            request.headers['user-agent'] = sanitized_ua
        
        return response

Finally, configure Fastapi's logging to use structured formatters that escape special characters:

import logging
import structlog

structlog.configure(
    processors=[
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

formatter = structlog.stdlib.ProcessorFormatter(
    processor=structlog.dev.ConsoleRenderer(),
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)