File size: 2,900 Bytes
cd0c7a9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """SSRF protection utilities.
Validates that user-supplied URLs point to allowed public hosts and
blocks requests to private/reserved IP ranges.
"""
from __future__ import annotations
import ipaddress
import logging
from urllib.parse import urlparse
from fastapi import HTTPException
logger = logging.getLogger(__name__)
ALLOWED_HOSTS: set[str] = {
"files.rcsb.org",
"data.rcsb.org",
"search.rcsb.org",
"www.rcsb.org",
"alphafold.ebi.ac.uk",
"rest.uniprot.org",
"www.uniprot.org",
}
PRIVATE_NETWORKS = [
ipaddress.ip_network("127.0.0.0/8"),
ipaddress.ip_network("10.0.0.0/8"),
ipaddress.ip_network("172.16.0.0/12"),
ipaddress.ip_network("192.168.0.0/16"),
ipaddress.ip_network("169.254.0.0/16"),
ipaddress.ip_network("::1/128"),
ipaddress.ip_network("fc00::/7"),
ipaddress.ip_network("fe80::/10"),
]
def validate_url(url: str, param_name: str = "url") -> None:
"""Validate a user-supplied URL against SSRF protections.
Checks:
1. URL is well-formed
2. Host is in the allowlist
3. Resolved IP is not in a private/reserved range
Raises HTTPException(400) on violation.
"""
if not url or not url.strip():
return
parsed = urlparse(url)
if parsed.scheme not in ("http", "https"):
raise HTTPException(status_code=400, detail=f"{param_name}: only http/https URLs are allowed")
hostname = parsed.hostname
if not hostname:
raise HTTPException(status_code=400, detail=f"{param_name}: invalid URL — no hostname")
# Check allowlist (suffix match to allow subdomains)
host_allowed = any(
hostname == allowed or hostname.endswith("." + allowed)
for allowed in ALLOWED_HOSTS
)
if not host_allowed:
raise HTTPException(
status_code=400,
detail=f"{param_name}: host '{hostname}' is not in the allowed list. "
f"Allowed: {', '.join(sorted(ALLOWED_HOSTS))}",
)
# Resolve IP and check for private ranges
try:
import socket
addrinfos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
for family, _, _, _, sockaddr in addrinfos:
ip = ipaddress.ip_address(sockaddr[0])
for net in PRIVATE_NETWORKS:
if ip in net:
raise HTTPException(
status_code=400,
detail=f"{param_name}: resolved to private IP {ip} — request blocked",
)
except HTTPException:
raise
except Exception as e:
logger.warning(f"SSRF DNS check failed for {hostname}: {e}")
# If DNS resolution fails, block the request rather than allowing it through
raise HTTPException(
status_code=400,
detail=f"{param_name}: could not resolve hostname — request blocked",
)
|