Dns Rebinding in Fastapi

DNS rebinding is a subtle but dangerous attack that exploits the trust relationship between a FastAPI application and the network it assumes it's running on. The attack works by manipulating DNS resolution so that a malicious domain resolves to a trusted internal IP address after an initial connection. FastAPI applications often make internal network calls without proper validation, making them vulnerable to this class of attack.

The most common manifestation in FastAPI occurs when applications accept URLs or hostnames from users and make outbound requests to those destinations. Consider a FastAPI endpoint that fetches external resources:

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.post("/fetch")
async def fetch_resource(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

An attacker could register a domain like malicious-domain.com that initially resolves to their own IP address. When the FastAPI application connects to this domain, the attacker responds with a short TTL and redirects the DNS to an internal IP like 192.168.1.1. The FastAPI application, trusting the original domain name, makes a request to what it believes is an external service but is actually an internal network endpoint.

FastAPI's async nature makes this particularly dangerous. The DNS resolution happens once at the start of the request, and the connection is maintained. If the DNS changes during this window, the application might unknowingly connect to internal services. This is especially problematic when FastAPI applications interact with microservices, databases, or internal APIs that assume they're only accessible from within the trusted network.

Another FastAPI-specific scenario involves WebSocket connections. FastAPI's WebSocket support makes it easy to create real-time applications that accept URLs for WebSocket connections:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws-connect")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    url = await websocket.receive_text()
    async with WebSocket(url) as ws:
        await ws.connect()
        # Proxy traffic between connections
        while True:
            data = await websocket.receive_bytes()
            await ws.send_bytes(data)

This pattern is vulnerable to DNS rebinding because the WebSocket URL is resolved once and the connection is maintained, potentially allowing an attacker to redirect traffic to internal services after the initial handshake.

Detecting DNS rebinding vulnerabilities in FastAPI applications requires both static analysis and runtime scanning. The most effective approach combines code review with automated scanning tools like middleBrick that understand FastAPI's specific patterns.

Static analysis should focus on identifying FastAPI endpoints that accept network destinations without validation. Look for patterns like:

# Vulnerable patterns
@app.post("/proxy")
def proxy_request(url: str):
    # No validation of destination
    
@app.get("/fetch")
def fetch_content(url: str = Query(...)):
    # Accepts any URL
    
@app.websocket("/ws-proxy")
async def ws_proxy(websocket: WebSocket):
    target_url = await websocket.receive_text()
    # No validation of WebSocket target

middleBrick's FastAPI-specific scanning identifies these patterns by analyzing your OpenAPI specification and runtime behavior. The scanner tests each endpoint by submitting controlled inputs and monitoring the application's response behavior. For DNS rebinding specifically, middleBrick:

• Analyzes OpenAPI specs to identify endpoints accepting URL parameters
• Tests with domains that have short TTL values to trigger potential rebinding
• Monitors for outbound connections to unexpected IP ranges
• Checks for WebSocket endpoints that could be used for rebinding attacks
• Validates that rate limiting and input validation are properly implemented

The scanner also examines your FastAPI application's dependency injection patterns. If you're using background tasks or async endpoints that make network calls, these need special attention. middleBrick can identify when your application uses httpx, aiohttp, or other HTTP clients in ways that might be vulnerable to DNS rebinding.

For comprehensive detection, run middleBrick with the --fastapi flag to enable FastAPI-specific checks:

middlebrick scan https://your-fastapi-app.com \
    --fastapi \
    --output json

This activates specialized detection for FastAPI patterns, including analysis of Pydantic models used for request validation, dependency injection patterns, and WebSocket endpoint configurations.

Remediating DNS rebinding vulnerabilities in FastAPI requires a defense-in-depth approach that combines input validation, network controls, and secure coding practices. The most effective solutions leverage FastAPI's built-in features and Python's ecosystem.

First, implement strict URL validation using Pydantic models with custom validators:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from urllib.parse import urlparse
import ipaddress

class ValidURL(BaseModel):
    url: str
    
    @validator('url')
    def validate_url(cls, v):
        parsed = urlparse(v)
        if not parsed.scheme in ['http', 'https']:
            raise ValueError('Only HTTP/HTTPS URLs are allowed')
        
        # Check if IP is private
        try:
            ip = ipaddress.ip_address(parsed.hostname)
            if ip.is_private or ip.is_reserved or ip.is_loopback:
                raise ValueError('Private IP addresses are not allowed')
        except ValueError:
            # Not an IP, check if it's a local domain
            if parsed.hostname in ['localhost', '127.0.0.1']:
                raise ValueError('Localhost is not allowed')
        
        return v

app = FastAPI()

@app.post("/safe-fetch")
async def safe_fetch(data: ValidURL):
    async with httpx.AsyncClient() as client:
        response = await client.get(data.url)
        return response.text

This approach validates both the URL structure and prevents connections to private IP ranges. The custom validator checks the hostname against private IP ranges using Python's ipaddress module, blocking attempts to connect to internal networks.

For WebSocket endpoints, implement similar validation and add connection timeouts:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import validator
import asyncio

class WebSocketTarget(BaseModel):
    url: str
    
    @validator('url')
    def validate_ws_url(cls, v):
        if not v.startswith('ws://') and not v.startswith('wss://'):
            raise ValueError('Only WebSocket URLs are allowed')
        return v

@app.websocket("/secure-ws")
async def secure_ws(websocket: WebSocket):
    await websocket.accept()
    try:
        data = await websocket.receive_text()
        target = WebSocketTarget(url=data)
        
        # Validate target before connecting
        async with WebSocket(target.url) as ws:
            # Set connection timeout
            await asyncio.wait_for(ws.connect(), timeout=5.0)
            
            # Proxy with timeout
            while True:
                data = await websocket.receive_bytes(timeout=30.0)
                await ws.send_bytes(data)
                
    except (ValueError, WebSocketDisconnect, asyncio.TimeoutError):
        await websocket.close(code=1000)

Network-layer controls provide additional protection. Use FastAPI's middleware to implement IP-based filtering:

from fastapi import Request, HTTPException
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

class IPFilterMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, request: Request):
        client_ip = request.client.host
        if client_ip:
            ip = ipaddress.ip_address(client_ip)
            if ip.is_private:
                raise HTTPException(status_code=403, detail="Private IP not allowed")
        
        return await self.app(request)

app.add_middleware(IPFilterMiddleware)

Finally, integrate middleBrick into your CI/CD pipeline to catch these issues before deployment:

name: API Security Scan
on: [push, pull_request]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Run middleBrick Scan
        run: |
          npm install -g middlebrick
          middlebrick scan https://staging.your-app.com \
            --fastapi \
            --fail-below B \
            --output json > security-report.json
      - name: Upload Report
        uses: actions/upload-artifact@v2
        with:
          name: security-report
          path: security-report.json

This combination of input validation, network controls, and automated scanning provides comprehensive protection against DNS rebinding attacks in FastAPI applications.