Overview
Real-world SSRF can allow an attacker to coerce your server into making requests to internal or protected resources. By abusing a parameter that accepts a URL, an attacker can reach internal services, cloud instance metadata endpoints, or other systems behind a firewall, potentially leaking credentials or enabling port scanning and further exploitation. Without proper controls, such requests can be performed with the same network privileges as your FastAPI process, dramatically expanding the attack surface.
In FastAPI, SSRF typically materializes when an endpoint fetches user-supplied URLs using httpx, requests, or urllib without validating or constraining the destination. Because FastAPI handlers run in your application process, the server effectively becomes a proxy for outbound calls to arbitrary targets, including internal networks, which can lead to data exfiltration, lateral movement, and access to protected services.
To mitigate SSRF, implement strict URL validation and an allowlist of trusted hosts, enforce network egress controls, and route outbound requests through a controlled gateway. Avoid using user-provided URLs directly; adopt a safe fetch utility, apply timeouts, disable redirects where appropriate, and log suspicious attempts to support detection and incident response.
Code Fix Example
FastAPI API Security Remediation
VULNERABLE PATTERN
from fastapi import FastAPI
import httpx
app = FastAPI()
@app.post('/fetch')
async def fetch(url: str):
resp = httpx.get(url, timeout=5.0)
return {'status': resp.status_code, 'body': resp.text[:200]}
# FIX: URL validation + allowlist
import urllib.parse
from fastapi import HTTPException
ALLOWED_SCHEMES = {'http','https'}
ALLOWED_HOSTS = {'example.com','api.example.com'}
def is_allowed(url: str) -> bool:
p = urllib.parse.urlparse(url)
if p.scheme not in ALLOWED_SCHEMES:
return False
if not p.hostname:
return False
if p.hostname not in ALLOWED_HOSTS:
return False
return True
@app.post('/fetch-safe')
async def fetch_safe(url: str):
if not is_allowed(url):
raise HTTPException(status_code=400, detail='URL not allowed')
resp = httpx.get(url, timeout=5.0)
return {'status': resp.status_code, 'body': resp.text[:200]}