Ssrf in Fastapi

Server-Side Request Forgery (SSRF) in FastAPI applications typically occurs when the framework accepts user-controlled URLs for outbound requests and processes them without validation. FastAPI's async nature and modern Python features make it particularly susceptible to certain SSRF patterns that developers might overlook.

A common manifestation involves FastAPI endpoints that accept URLs for proxy functionality or external service integration. Consider this vulnerable pattern:

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.post('/proxy')
def proxy_image(url: str):
    response = httpx.get(url)
    return response.content

This endpoint allows any URL, enabling attackers to target internal services, cloud metadata endpoints, or perform port scanning. FastAPI's automatic request validation via Pydantic models doesn't prevent SSRF since it only validates data types, not URL safety.

Another FastAPI-specific pattern involves dependency injection combined with URL parameters:

from fastapi import Depends, FastAPI
import httpx

app = FastAPI()

async def get_url():
    return httpx.AsyncClient()

@app.post('/fetch')
def fetch_data(client: httpx.AsyncClient = Depends(), url: str):
    response = client.get(url)
    return response.json()

The dependency injection system makes this pattern cleaner but equally vulnerable. FastAPI's OpenAPI generation also exposes these endpoints clearly in the generated spec, making them easier targets for automated scanning.

Cloud-based FastAPI deployments face additional risks through environment-specific endpoints. Attackers often target:

• Cloud metadata services (169.254.169.254 for AWS, 169.254.169.254 for Azure)
• Database services on common ports (5432 for PostgreSQL, 3306 for MySQL)
• Internal API endpoints (localhost:8000, 127.0.0.1:8000)
• Container orchestration APIs (localhost:2375 for Docker)

FastAPI's async/await pattern can mask SSRF latency issues, making detection harder. An attacker might chain multiple SSRF requests through async endpoints, creating timing-based side-channel attacks that bypass simple rate limiting.

Detecting SSRF in FastAPI requires understanding both the framework's patterns and the specific attack vectors. Static analysis can identify vulnerable code patterns, but runtime detection provides more comprehensive coverage.

Code-level detection should flag these FastAPI patterns:

# Vulnerable pattern - user-controlled URL in async client
@app.post('/download')
async def download_file(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.content

# Dependency injection vulnerability
@app.post('/api-call')
async def api_call(client: httpx.AsyncClient = Depends(), url: str):
    return await client.get(url)

Dynamic detection through middleBrick scanning specifically tests FastAPI applications by:

• Submitting known SSRF payloads to all POST/GET endpoints accepting URL parameters
• Checking responses for internal service indicators (error messages, headers)
• Testing against common cloud metadata endpoints and internal IPs
• Analyzing OpenAPI specs generated by FastAPI for exposed URL parameters
• Attempting protocol smuggling with file://, gopher://, and other non-HTTP schemes

middleBrick's FastAPI-specific detection includes LLM/AI security checks that are particularly relevant since FastAPI is commonly used for AI/ML APIs. The scanner tests for:

- System prompt extraction via crafted URLs
- Prompt injection through URL parameters
- Excessive agency detection in AI endpoints
- Unauthenticated LLM endpoint discovery

Runtime monitoring can complement scanning by logging outbound requests from FastAPI applications and flagging unusual patterns like requests to private IP ranges or uncommon ports.

Remediating SSRF in FastAPI requires defense-in-depth approaches combining input validation, allowlisting, and architectural changes. Here are FastAPI-specific remediation patterns:

Input validation using Pydantic models with custom validators:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
import httpx
import re

class UrlRequest(BaseModel):
    url: str
    
    @validator('url')
    def validate_url(cls, v):
        # Allowlist HTTP/HTTPS only
        if not v.lower().startswith(('http://', 'https://')):
            raise ValueError('Only HTTP/HTTPS URLs are allowed')
        
        # Block private IP ranges
        private_ips = [
            r'10\\.\\d+\\.\\d+\\.\\d+',        # 10.0.0.0/8
            r'172\\.(1[6-9]|2[0-9]|3[0-1])\\.\\d+\\.\\d+',  # 172.16.0.0/12
            r'192\\.168\\.\\d+\\.\\d+',       # 192.168.0.0/16
            r'127\\.\\d+\\.\\d+\\.\\d+',      # loopback
            r'0\\.\\d+\\.\\d+\\.\\d+',        # 0.0.0.0/8
            r'169\\.254\\.\\d+\\.\\d+',       # link-local
            r'192\\.0\\.2\\.\\d+',             # documentation
            r'255\\.255\\.255\\.255',           # broadcast
        ]
        
        for pattern in private_ips:
            if re.match(pattern, v):
                raise ValueError('Private IP addresses are not allowed')
        
        return v

app = FastAPI()

@app.post('/safe-proxy')
async def safe_proxy(request: UrlRequest):
    async with httpx.AsyncClient() as client:
        response = await client.get(request.url)
        return response.content

Service-specific allowlisting for known external services:

from fastapi import FastAPI
from pydantic import BaseModel
import httpx

class ExternalServiceRequest(BaseModel):
    service: str
    path: str
    
    @validator('service')
    def validate_service(cls, v):
        allowed_services = ['github', 'api', 'cdn']
        if v not in allowed_services:
            raise ValueError('Unknown service')
        return v

SERVICES = {
    'github': 'https://api.github.com',
    'api': 'https://api.example.com',
    'cdn': 'https://cdn.example.com'
}

app = FastAPI()

@app.post('/external-service')
async def external_service(request: ExternalServiceRequest):
    base_url = SERVICES[request.service]
    full_url = f'{base_url}{request.path}'
    
    async with httpx.AsyncClient() as client:
        response = await client.get(full_url)
        return response.json()

Network-layer controls complement application-level fixes:

# Using middleware for additional filtering
from fastapi import Request, Response
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

class SSRFMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, request: Request, call_next):
        # Check for suspicious patterns in request body
        body = await request.body()
        if b'localhost' in body or b'127.0.0.1' in body:
            return Response(status_code=400, content='Invalid URL')
        
        return await call_next(request)

app.add_middleware(SSRFMiddleware)

For AI/ML endpoints built with FastAPI, implement additional LLM-specific protections:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx

class LLMRequest(BaseModel):
    prompt: str
    
    @validator('prompt')
    def validate_prompt(cls, v):
        # Block system prompt extraction patterns
        if '### System:' in v or 'System:' in v:
            raise ValueError('Invalid prompt format')
        return v

app = FastAPI()

@app.post('/llm/generate')
async def generate_text(request: LLMRequest):
    # Additional LLM-specific SSRF protections
    if 'url=' in request.prompt.lower():
        raise HTTPException(status_code=400, detail='URL parameters not allowed in prompts')
    
    # LLM processing logic here
    return {'response': 'Generated text'}