Overview
SSRF vulnerabilities in FastAPI can occur when an endpoint accepts user-supplied URLs to fetch resources. An attacker can trick the server into contacting internal services, cloud metadata endpoints, or other protected resources, potentially leaking data or causing actions to be performed on behalf of the server. This risk is particularly acute in microservice architectures where services talk to each other using user-provided URLs. If such requests are not properly restricted, the server effectively becomes an agent that can be driven to the attacker’s targets.
CVE-2021-32677 demonstrates how insecure request handling in FastAPI can open the door to CSRF: versions earlier than 0.65.2 read the request payload as JSON even when the content-type was not set to a JSON media type, enabling cross-site requests that carry cookies. The fix in 0.65.2 restricted parsing to application/json and similar media types. While this CVE is CSRF-related, it underscores the broader principle of strict input handling and the risk surface from poorly validated request data.
To defend against SSRF in FastAPI, upgrade to the patched release (0.65.2+). If upgrading is not possible, implement a middleware or validation that enforces strict content-type handling and robust outbound URL controls. In addition to patching, apply URL validation, host allowlists, network egress controls, timeouts, and safe-fetch patterns to minimize exposure to internal resources.
This guide provides concrete patterns and a remediation checklist you can apply to FastAPI projects to reduce SSRF risk while aligning with the lessons from CVE-2021-32677 about strict input handling and defense-in-depth.
Affected Versions
FastAPI < 0.65.2
Code Fix Example
FastAPI API Security Remediation
Vulnerable:
from fastapi import FastAPI
import requests
app = FastAPI()
@app.post('/fetch')
async def fetch(url: str):
resp = requests.get(url, timeout=5)
return {'status': resp.status_code, 'content': resp.text[:100]}
# Fixed:
from fastapi import FastAPI, HTTPException
from pydantic import HttpUrl, BaseModel
import requests
app = FastAPI()
class URLRequest(BaseModel):
url: HttpUrl
ALLOWED_DOMAINS = {'example.com','api.example.com'}
def is_allowed(url: str) -> bool:
from urllib.parse import urlparse
try:
p = urlparse(url)
except Exception:
return False
if p.scheme not in ('http','https'):
return False
host = p.hostname or ''
if host.startswith('127.') or host == 'localhost':
return False
if not any(host.endswith(d) for d in ALLOWED_DOMAINS):
return False
return True
@app.post('/fetch')
async def fetch(req: URLRequest):
url = str(req.url)
if not is_allowed(url):
raise HTTPException(status_code=400, detail='URL not allowed')
resp = requests.get(url, timeout=5)
return {'status': resp.status_code, 'content': resp.text[:100]}