Auth Headers Logged Before Auth Check

What it is

This is a sensitive data exposure via log injection (CWE-532), where a FastAPI MCP endpoint passes the raw `request.headers` dict — including `Authorization`, `X-Api-Key`, and similar bearer tokens — directly into a logging statement before any authentication gate has been evaluated. The credential is written in plaintext to log files, stdout streams, and any downstream log aggregators before the server has confirmed the caller is legitimate.

Why it matters for MCP

MCP servers receive high-frequency, programmatically generated requests from LLM orchestrators that embed long-lived API keys and session tokens directly in headers, making every request a high-value credential exposure event. Unlike browser-facing APIs where human operators notice anomalous log volume, MCP endpoints are often headlessly deployed with log pipelines fed into observability platforms (Datadog, Splunk, CloudWatch) accessible to broad engineering teams, dramatically widening the blast radius of a single logging mistake. Additionally, prompt injection payloads embedded in request bodies may deliberately trigger verbose logging paths to exfiltrate credentials into logs the attacker can later read.

Vulnerable example

import logging

from fastapi import FastAPI, Request

app = FastAPI()

logger = logging.getLogger(__name__)

@app.post("/mcp")

async def handle_mcp_request(request: Request):

    body = await request.json()

    # VULNERABLE: entire headers dict logged before auth check

    logger.info(f"MCP request: {request.headers} | Body: {body}")

    auth_header = request.headers.get("Authorization")

    if not auth_header or not validate_token(auth_header):

        return JSONResponse(status_code=401, content={"error": "Unauthorized"})

    return JSONResponse(content={"status": "ok"})

Secure example

import logging

from fastapi import FastAPI, Request

app = FastAPI()

logger = logging.getLogger(__name__)

SAFE_HEADERS = {"content-type", "accept", "user-agent"}

@app.post("/mcp")

async def handle_mcp_request(request: Request):

    body = await request.json()

    auth_header = request.headers.get("Authorization")

    if not auth_header or not validate_token(auth_header):

        logger.warning("Unauthorized MCP request from %s", request.client.host)

        return JSONResponse(status_code=401, content={"error": "Unauthorized"})

    safe_headers = {k: v for k, v in request.headers.items() if k.lower() in SAFE_HEADERS}

    logger.info("MCP request headers=%s", safe_headers)

    return JSONResponse(content={"status": "ok"})

How MCPSafe detects this

MCPSafe performs taint analysis tracing `request.headers` (and full-header accessors such as `dict(request.headers)`) as taint sources, flagging any flow where the tainted value reaches a logging sink (`logger.debug/info/warning/error/critical`, `print`, `logging.log`) that appears in the AST before a conditional branch containing an authentication check (`validate_token`, `verify_jwt`, header presence guards, or HTTP 401 returns). Calls that log only allowlisted header keys via explicit dict-comprehension filtering or `request.headers.get()` with a non-sensitive key name are excluded from the finding.

See the full threat catalog for every documented detection.

Further reading

• CWE-532: Insertion of Sensitive Information into Log File
• OWASP Logging Cheat Sheet: Data to Exclude
• FastAPI Request Object — Headers
• NIST SP 800-92: Guide to Computer Security Log Management