Dns Cache Poisoning in Fastapi with Api Keys

DNS cache poisoning (also known as DNS spoofing) occurs when an attacker injects a malicious DNS record into a resolver’s cache, causing a victim to be redirected to an attacker-controlled host. In a FastAPI application that relies on external service discovery via DNS—such as resolving a backend hostname used for outgoing requests—this can redirect traffic without the application ever using a fixed IP address. When API keys are used for authorization, the risk is compounded because the application may automatically forward those keys to the poisoned endpoint, thinking it is a trusted service.

Consider a FastAPI service that resolves a payment provider hostname at startup or lazily on each request. If the DNS response is poisoned, the hostname resolves to an attacker server that mimics the payment API. Because API keys are often stored in environment variables and passed with each request, the application will include them when communicating with the malicious host. middleBrick’s checks for SSRF and Data Exposure highlight this class of misconfiguration by correlating DNS-dependent request flows with outbound transmissions of credentials.

Even when API keys are rotated frequently, poisoning can cause immediate leakage or manipulation if the application does not validate the identity of the resolved host. For example, an attacker controlling a compromised subdomain or a vulnerable public resolver may serve a crafted response that maps a legitimate service name to a malicious IP. The FastAPI app, lacking hostname pinning or strict certificate validation, will connect to the malicious IP and present API keys as part of normal authentication headers. This pattern is surfaced by middleBrick’s checks for Encryption and Data Exposure, which inspect whether transmitted credentials are sent over unverified channels.

In architectures where service discovery is performed by querying an internal DNS-based service registry, unsafe consumption of DNS responses can also enable privilege escalation through BOLA/IDOR-like confusion: an attacker might register or compromise a low-privilege service entry so that its DNS record points to a malicious host. The FastAPI app then routes privileged traffic there, effectively chaining DNS cache poisoning with authorization weaknesses. middleBrick’s parallel security checks—particularly Inventory Management and Unsafe Consumption—help surface these cross-category risks by mapping spec-defined hosts to runtime observations.

middleBrick detects these scenarios by tracing how external hostnames are resolved and used, comparing OpenAPI/Swagger definitions with observed network behavior. When API keys are included in requests to dynamically resolved endpoints, the scanner highlights the absence of integrity controls on DNS resolution and the presence of credential transmission, aligning findings with relevant items from OWASP API Top 10 and common compliance mappings.

Remediation focuses on ensuring that API keys are only sent to verified endpoints and that DNS resolution is treated as an untrusted input. Below are concrete code examples that you can apply in FastAPI to reduce the risk when using API keys in conjunction with dynamic service discovery.

1. Pin hostnames and validate certificates

Use an HTTP client that supports certificate or hostname pinning, and avoid trusting arbitrary DNS responses. With httpx, you can pin a certificate fingerprint and enforce strict HTTPS verification.

import httpx
import ssl

ssl_context = ssl.create_default_context()
# Pin to a specific certificate fingerprint (SHA-256)
ssl_context.load_verify_locations(cafile="pinned-ca.pem")

client = httpx.AsyncClient(
    verify=ssl_context,
    headers={"Authorization": "ApiKey YOUR_API_KEY"}
)

async def call_payment(amount: float) -> dict:
    response = await client.post("https://api.payment.example.com/charge", json={"amount": amount})
    response.raise_for_status()
    return response.json()

2. Use an allowlist of known service endpoints

Maintain a strict allowlist of resolved endpoints and validate the target before sending sensitive data. This prevents requests from being redirected to attacker-controlled hosts due to poisoned DNS.

from fastapi import Depends, HTTPException, status
import httpx

ALLOWED_HOSTS = {"api.payment.example.com", "api.inventory.example.com"}

async def get_allowed_client(url: str):
    if url not in ALLOWED_HOSTS:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Endpoint not allowed"
        )
    return httpx.AsyncClient(headers={"Authorization": "ApiKey YOUR_API_KEY"})

@app.post("/charge")
async def charge(payload: dict, client: httpx.AsyncClient = Depends(get_allowed_client)):
    resp = await client.post("https://api.payment.example.com/charge", json=payload)
    return resp.json()

3. Rotate API keys and scope them per service

Ensure each integration uses its own API key with minimal permissions. If a key is associated with a specific service and host, compromise of one key does not grant access to others. Combine this with short lifetimes and automated rotation where feasible.

import os
from fastapi import Depends, Header, HTTPException, status

SERVICE_KEYS = {
    "api.payment.example.com": os.getenv("PAYMENT_API_KEY"),
    "api.inventory.example.com": os.getenv("INVENTORY_API_KEY"),
}

def get_api_key(host: str = Header(...)):
    key = SERVICE_KEYS.get(host)
    if not key:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid service")
    return {"Authorization": f"ApiKey {key}"}

@app.post("/charge")
async def charge(headers: dict = Depends(get_api_key)):
    client = httpx.AsyncClient(headers=headers)
    resp = await client.post("https://api.payment.example.com/charge", json={"amount": 100})
    return resp.json()

4. Validate server identity using public key or certificate pinning

When feasible, pin the server’s public key or certificate hash rather than relying on DNS alone. This ensures that even if DNS is poisoned, the TLS handshake will fail if the server cannot present the pinned credential.

import ssl
import httpx

# Pin by public key hash (SPKI)
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations(cafile="spki-pin.pem")

client = httpx.AsyncClient(
    verify=ssl_context,
    headers={"Authorization": "ApiKey YOUR_API_KEY"}
)

# This call will only succeed if the server presents the pinned key
resp = await client.get("https://api.payment.example.com/status")

5. Enforce strict transport and reject insecure resolutions

Ensure all outbound calls use HTTPS and reject any resolution that does not present a valid certificate chain. Avoid disabling verification for convenience, as this would nullify the protection provided by API keys.

import httpx

client = httpx.AsyncClient(
    verify=True,
    headers={"Authorization": "ApiKey YOUR_API_KEY"},
    timeout=10.0
)

# Example usage
async def safe_call():
    r = await client.get("https://api.payment.example.com/health")
    r.raise_for_status()

These patterns—pinning, allowlists, per-service keys, and strict verification—reduce the blast radius if DNS cache poisoning occurs. They ensure that API keys remain bound to trusted endpoints and are not automatically forwarded to attacker-controlled infrastructure.