Overview
SSRF (Server-Side Request Forgery) vulnerabilities occur when a server makes HTTP requests to URLs supplied by a client without proper validation, allowing attackers to reach internal services, cloud metadata endpoints, or other restricted resources. In real-world usage, this can enable data exfiltration, port scanning of internal networks, or interactions with internal services not meant to be exposed publicly. The CVE-2024-40627 case demonstrates how insecure HTTP method handling within FastAPI ecosystem components can indirectly broaden an attack surface: the OpaMiddleware in the Fastapi OPA middleware allowed unauthenticated HTTP OPTIONS requests to pass through without policy evaluation, enabling information disclosure about entity existence. Although this CVE is not a direct SSRF, it underscores the broader risk of lax request handling and policy enforcement in a FastAPI deployment. When SSRF is possible, attackers can supply a server-side URL that the app fetches, potentially targeting internal or restricted resources. Upgrading vulnerable middleware and applying strict input validation are essential to mitigate these risks in FastAPI applications that fetch remote resources based on user input. Reference CVE-2024-40627 to understand the importance of enforcing proper authorization and policy checks for all HTTP methods and request flows. This guide combines lessons from that CVE with concrete SSRF remediations for FastAPI in Python.
Affected Versions
fastapi-opa OpaMiddleware: affected <= 2.0.0; fixed in 2.0.1
Code Fix Example
FastAPI API Security Remediation
VULNERABLE:
from fastapi import FastAPI, HTTPException
import httpx
app = FastAPI()
@app.get("/fetch")
async def fetch(url: str):
# WARNING: accepts user-provided URL and fetches it without validation
resp = httpx.get(url, timeout=5)
return {"status": resp.status_code, "body": resp.text[:200]}
FIX:
from fastapi import FastAPI, HTTPException
import httpx
import urllib.parse
APP_URLS_WHITELIST = {"example.org", "api.example.org"}
app = FastAPI()
def is_allowed_url(target: str) -> bool:
try:
parsed = urllib.parse.urlparse(target)
except Exception:
return False
if parsed.scheme not in {"http", "https"}:
return False
host = parsed.netloc.split(":")[0]
# basic hostname allowlist (supports subdomains)
if not any(host.endswith(d) for d in APP_URLS_WHITELIST):
return False
return True
@app.get("/fetch")
async def fetch(url: str):
if not is_allowed_url(url):
raise HTTPException(status_code=400, detail="URL not allowed")
resp = httpx.get(url, timeout=5, follow_redirects=False)
return {"status": resp.status_code, "body": resp.text[:200]}