Overview
In real-world deployments, SSRF can enable an attacker to use your FastAPI service as a proxy to reach internal networks, cloud metadata endpoints, or other services not publicly exposed. If an endpoint accepts a URL and fetches it server-side, the server may inadvertently access internal resources from the client's perspective. This class of vulnerability can lead to data exfiltration, unauthorized access to services, and potential lateral movement within a private network.
In FastAPI, SSRF typically manifests when an API endpoint accepts a URL parameter and uses an HTTP client (such as httpx or requests) to fetch the resource. If the developer does not validate the URL, restrict domains, or properly handle internal addresses, an attacker can direct the server to contact internal services (e.g., 169.254.169.254, 127.0.0.1, or private VPC endpoints) or metadata services, potentially leaking credentials or information. Even when some controls exist, misconfigurations or overly permissive allowlists can still enable exploitation.
Remediation patterns revolve around restricting outbound fetches: implement an explicit allowlist of domains, validate and normalize URLs, ensure only trusted protocols are allowed, and prevent DNS rebinding. When possible, perform server-side fetches only to a narrow set of trusted hosts, or avoid URL-based retrieval entirely by moving to a controlled data source or background task. Use timeouts, limit response data, and consider proxying requests through a centralized service that enforces network policies.
Security hygiene includes logging SSRF attempts, testing endpoints with targeted SSRF tests, and applying defense-in-depth with network egress controls and container runtime restrictions. Regularly review dependencies (e.g., httpx, requests) for SSRF-related fixes and exercise strict input validation.
Code Fix Example
FastAPI API Security Remediation
from fastapi import FastAPI, HTTPException
import httpx
from urllib.parse import urlparse
app = FastAPI()
# Vulnerable: fetch arbitrary URL provided by client
@app.get("/vuln-fetch")
async def vuln_fetch(url: str):
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(url)
return {"status": resp.status_code, "content": resp.text[:200]}
# Safe: allowlist-based fetching
ALLOWED_HOSTS = {"example.com", "api.example.org"}
def is_allowed(u: str) -> bool:
try:
p = urlparse(u)
host = p.hostname
if not host:
return False
if host.endswith(".example.com") or host in ALLOWED_HOSTS:
return True
return False
except Exception:
return False
@app.get("/safe-fetch")
async def safe_fetch(url: str):
if not is_allowed(url):
raise HTTPException(status_code=400, detail="URL not allowed")
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(url)
return {"status": resp.status_code, "content": resp.text[:200]}