Log Injection in Fastapi with Jwt Tokens

Log injection occurs when untrusted input is written directly into log entries without sanitization, enabling an attacker to forge log lines, obscure real events, or trigger log-based attacks such as log poisoning or log injection into monitoring systems. In Fastapi applications that use JWT tokens for authentication, the combination of structured token handling, user-controlled claims, and verbose logging can unintentionally create injection opportunities.

When a Fastapi service decodes and validates JWT tokens—commonly via libraries such as python-jose or PyJWT—developers often log token metadata (e.g., subject, roles, issuer) for audit or debugging purposes. If any part of the token payload (e.g., the sub, name, or custom claims) is reflected in logs without escaping newlines or special characters, an attacker can craft a token containing carriage returns or line feeds to inject additional log lines. For example, a name claim like alice\n[WARN] admin privileges escalated can split a single logical log entry into multiple fabricated entries, potentially misleading operators or bypassing simple log filters that rely on line-based parsing.

In addition to payload injection, JWT tokens often carry scopes or roles that influence authorization decisions. If authorization failures are logged with insufficient sanitization, an attacker may forge log lines to imply elevated privileges or successful actions they did not perform. Consider a Fastapi route that logs User {username} accessed {resource} with scopes {scopes}. If scopes are sourced directly from the JWT without validation, an attacker presenting a token with a modified scopes claim can alter the apparent behavior of the system in logs. This does not bypass enforcement (assuming enforcement is implemented correctly), but it degrades trust in observability and incident response.

Another subtle risk arises when tokens are logged at the framework or middleware level. Fastapi dependency injection and middleware may log request paths, method, and extracted JWT claims as part of access logging. If these logs are aggregated into centralized systems that parse structured logs via regular expressions or simple delimiters, newline or control-character injection can break ingestion pipelines, cause parsing errors, or enable log injection into dashboards. Real-world attack patterns observed in OAuth and API abuse cases include using newline injection to obscure credential leakage or to manipulate SIEM correlation rules.

While JWT tokens themselves are cryptographically signed, the signature only guarantees integrity of the contents; it does not prevent an attacker from supplying a token with maliciously crafted claims if they possess a valid key or exploit a weak secret. Therefore, the vulnerability is not in the cryptographic verification but in the downstream handling of token data. Logging should treat all token-derived fields as untrusted input, applying the same sanitization applied to other user inputs.

Defensive handling of JWT tokens in Fastapi requires sanitizing any data extracted from tokens before it reaches logging, as well as enforcing strict validation and controlled formatting. Below are concrete, actionable code examples demonstrating secure practices.

1. Validate and sanitize claims before logging

Do not directly interpolate raw claims into log messages. Instead, normalize and escape special characters. For instance, replace newlines and carriage returns, and avoid logging sensitive claims unless strictly necessary.

import logging
from fastapi import FastAPI, Depends, HTTPException
from jose import jwt, JWTError
import re

app = FastAPI()
logger = logging.getLogger(__name__)

SECRET_KEY = "your-secret"
ALGORITHM = "HS256"

def sanitize_log_value(value: str) -> str:
    # Replace newlines and carriage returns to prevent log injection
    return re.sub(r'[\r\n]', ' ', value.strip())

def get_current_user(token: str = None):
    if not token:
        raise HTTPException(status_code=401, detail="Missing token")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        scopes: str = payload.get("scopes", "")
        # Sanitize before any logging
        safe_username = sanitize_log_value(username) if username else "unknown"
        safe_scopes = sanitize_log_value(scopes) if scopes else "none"
        logger.info(f"Authenticated user: {safe_username}, scopes: {safe_scopes}")
        return payload
    except JWTError as e:
        logger.warning(f"Invalid token: {str(e)}")
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/items/")
async def read_items(current_user: dict = Depends(get_current_user)):
    return {"message": "ok"}

2. Structured logging with controlled fields

Use structured logging (e.g., JSON logs) and explicitly select which fields to emit. This reduces the risk of injection disrupting log format and makes automated parsing safer.

import json
import logging
from fastapi import FastAPI, Depends
from jose import jwt

app = FastAPI()
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

SECRET_KEY = "your-secret"

def decode_token(token: str):
    return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

@app.get("/profile")
def profile(token: str = None):
    if not token:
        raise HTTPException(status_code=401, detail="Missing token")
    payload = decode_token(token)
    # Explicitly pick and sanitize fields
    log_entry = {
        "event": "profile_access",
        "sub": payload.get("sub", "").replace("\n", " ").replace("\r", " "),
        "aud": payload.get("aud", ""),
        "iat": payload.get("iat"),
    }
    logger.info(json.dumps(log_entry))
    return {"user": log_entry["sub"]}

3. Centralized logging middleware with sanitization

Apply sanitization consistently across all log statements by using middleware that processes token-derived fields before they are emitted by any route-specific logging.

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import logging
import re

app = FastAPI()
logger = logging.getLogger("access")

class LoggingSanitizationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Example: extract token from header and sanitize claims if logged
        authorization = request.headers.get("authorization", "")
        token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else ""
        # Perform sanitization where token claims are later used in logs
        response = await call_next(request)
        # Safe audit logging
        logger.info(f"Request {request.method} {request.url.path} completed")
        return response

app.add_middleware(LoggingSanitizationMiddleware)

These practices ensure that JWT-derived data does not corrupt logs or obscure real events, preserving the integrity of observability and incident response.