Overview
SSRF vulnerabilities in FastAPI apps occur when a server side endpoint accepts a URL from a client and fetches that URL without validating the destination. An attacker can use this to reach internal services, metadata endpoints, or other protected resources, potentially leading to data exfiltration, internal reconnaissance, or abuse of internal services. The real-world impact can be severe in cloud or container environments where outbound access to internal networks is possible. For context, CVE-2024-42816 describes an XSS vulnerability in fastapi-admin pro v0.1.4, illustrating how unsanitized user input in a FastAPI-based app can lead to harmful behavior if input handling is lax (CWE-79). While that CVE is about XSS, it underscores the broader risk of trusting untrusted user input in FastAPI applications, including SSRF scenarios where outbound requests are driven by user-supplied data. This guide references that CVE to demonstrate the importance of strict input validation and safe outbound request handling to prevent SSRF in FastAPI apps.
Affected Versions
fastapi-admin pro v0.1.4 (CVE-2024-42816)
Code Fix Example
FastAPI API Security Remediation
Vulnerable (SSRF surface):
from fastapi import FastAPI
import httpx
app = FastAPI()
@app.get("/vuln-fetch")
async def vuln_fetch(url: str):
# Vulnerable: no validation of 'url'; server will fetch any provided URL
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(url)
return {"status": resp.status_code, "body": resp.text[:200]}
# Fixed (SSRF mitigated): use a strict allowlist and controlled fetch
from urllib.parse import urlparse
from fastapi import HTTPException
ALLOWED_SCHEMES = {"http", "https"}
ALLOWED_HOSTS = {"example.org", "api.example.org"} # adjust per environment
def is_allowed_target(target_url: str) -> bool:
try:
p = urlparse(target_url)
if p.scheme not in ALLOWED_SCHEMES:
return False
if not p.hostname:
return False
if p.hostname not in ALLOWED_HOSTS and not p.hostname.endswith(".example.org"):
return False
return True
except Exception:
return False
@app.get("/vuln-fetch-secure")
async def vuln_fetch_secure(url: str):
if not is_allowed_target(url):
raise HTTPException(status_code=400, detail="URL not allowed for outbound fetch")
async with httpx.AsyncClient(timeout=5.0, follow_redirects=False) as client:
resp = await client.get(url)
return {"status": resp.status_code, "body": resp.text[:200]}