""" safe_fetch — fetch user-supplied URLs without opening an SSRF hole. The standard library will happily connect anywhere, including your own localhost, your cloud metadata endpoint, and every machine on your private network. If any part of your application fetches a URL a user gave you — a webhook tester, a link preview, an avatar importer, a site scanner — that is a Server-Side Request Forgery vulnerability unless you actively prevent it. This module is the prevention. Standard library only, no dependencies. from safe_fetch import fetch, SafeFetchError try: response = fetch("https://example.com") except SafeFetchError as e: print("refused:", e) MIT licensed. Extracted from https://issitesafe.com """ import ipaddress import socket import ssl import time import urllib.error import urllib.request from urllib.parse import urljoin, urlparse, urlunparse __version__ = "1.0.0" __all__ = ["fetch", "SafeFetchError", "is_blocked_ip", "normalise_url", "resolve_and_check"] DEFAULT_MAX_REDIRECTS = 5 DEFAULT_TIMEOUT = 8 DEFAULT_MAX_BODY = 64 * 1024 DEFAULT_USER_AGENT = "safe-fetch/%s" % __version__ ALLOWED_SCHEMES = ("http", "https") class SafeFetchError(Exception): """A request was refused, or could not be completed.""" # -------------------------------------------------------------------------- # Address filtering # -------------------------------------------------------------------------- def is_blocked_ip(ip_str): """ True if an address must never be connected to on a user's behalf. Python's ipaddress module already classifies every range we care about, so we lean on it rather than hand-writing CIDR lists that drift out of date: is_private 10/8, 172.16/12, 192.168/16, fc00::/7 and friends is_loopback 127/8, ::1 is_link_local 169.254/16 <-- cloud metadata lives here is_reserved future-use space is_multicast 224/4 is_unspecified 0.0.0.0, :: 169.254.169.254 deserves specific mention: on AWS, GCP and Azure that address serves instance metadata, often including credentials. It is the single most valuable target of an SSRF attack, and it is link-local, so is_link_local catches it. """ try: ip = ipaddress.ip_address(ip_str) except ValueError: # Not parseable as an address: refuse rather than guess. return True # Unwrap IPv4-mapped IPv6 (::ffff:127.0.0.1) and judge the real address. # Current CPython already delegates is_private/is_loopback to the mapped # address, so this is belt-and-braces rather than a fix — but it makes the # intent explicit and does not rely on that behaviour staying implicit. if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: ip = ip.ipv4_mapped return ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified ) def resolve_and_check(hostname): """ Resolve a hostname and refuse it if ANY returned address is internal. Checking every address matters. A hostname can resolve to several, and an attacker who controls DNS can return one public address and one private one, hoping you validate the first and connect to the second. This does not fully close DNS rebinding, where the record changes between your check and your connection. Closing that completely means connecting to a pinned IP and passing the hostname separately for TLS and Host. For most applications this check is the right amount of defence; if you are handling genuinely hostile input, pin the address too. """ try: infos = socket.getaddrinfo(hostname, None) except socket.gaierror: raise SafeFetchError("Could not resolve %r." % hostname) addresses = sorted({info[4][0] for info in infos}) if not addresses: raise SafeFetchError("Could not resolve %r." % hostname) for addr in addresses: if is_blocked_ip(addr): raise SafeFetchError( "%r resolves to %s, which is a private or internal address." % (hostname, addr) ) return addresses def normalise_url(raw): """Validate a URL and strip the parts we should never send onward.""" raw = (raw or "").strip() if not raw: raise SafeFetchError("No URL given.") if len(raw) > 2000: raise SafeFetchError("URL is too long.") if "://" not in raw: raw = "https://" + raw parsed = urlparse(raw) if parsed.scheme not in ALLOWED_SCHEMES: raise SafeFetchError( "Only http and https are allowed (got %r). Schemes such as file://, " "gopher:// and dict:// are classic SSRF escalation paths." % parsed.scheme ) if not parsed.hostname: raise SafeFetchError("No hostname in URL.") # Rebuild the authority from host and port ONLY. # # Using parsed.netloc here would keep any embedded credentials, so a URL of # the form https://user:pass@host/ would forward somebody's password to the # target. urlunparse does not strip them for you - the userinfo is part of # netloc, and it is easy to assume otherwise. host = parsed.hostname.rstrip(".") if not host: raise SafeFetchError("No hostname in URL.") # Internationalised domains have to be punycode on the wire, or a perfectly # real address such as munchen.de is reported as unresolvable. try: host = host.encode("idna").decode("ascii") except UnicodeError: raise SafeFetchError("Hostname contains characters that cannot be encoded.") try: netloc = "%s:%d" % (host, parsed.port) if parsed.port else host except ValueError: raise SafeFetchError("Invalid port.") return urlunparse((parsed.scheme, netloc, parsed.path or "/", "", "", "")) # -------------------------------------------------------------------------- # Fetching # -------------------------------------------------------------------------- class _NoRedirect(urllib.request.HTTPRedirectHandler): """Suppress automatic redirects so each hop can be re-validated.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None def fetch(url, timeout=DEFAULT_TIMEOUT, max_redirects=DEFAULT_MAX_REDIRECTS, max_body=DEFAULT_MAX_BODY, user_agent=DEFAULT_USER_AGENT, headers=None): """ Fetch a URL, validating every hop. Returns a dict: url, status, headers (lower-cased), body, chain, elapsed_ms. Raises SafeFetchError if the request is refused or fails. Redirects are followed manually and re-validated each time. This is the step most implementations miss: a public URL that 302s to http://169.254.169.254/ defeats any check performed only on the first URL. """ opener = urllib.request.build_opener(_NoRedirect) current = normalise_url(url) chain = [] request_headers = { "User-Agent": user_agent, "Accept": "*/*", "Accept-Language": "en", } if headers: request_headers.update(headers) for _ in range(max_redirects + 1): parsed = urlparse(current) resolve_and_check(parsed.hostname) # re-checked on every hop chain.append(current) req = urllib.request.Request(current, method="GET", headers=request_headers) started = time.time() try: resp = opener.open(req, timeout=timeout) body = resp.read(max_body) return { "url": current, "status": resp.status, "headers": {k.lower(): v for k, v in resp.headers.items()}, "body": body, "chain": chain, "elapsed_ms": int((time.time() - started) * 1000), } except urllib.error.HTTPError as e: if e.code in (301, 302, 303, 307, 308): location = e.headers.get("Location") if not location: raise SafeFetchError("Redirect with no Location header.") current = normalise_url(urljoin(current, location)) continue # 4xx and 5xx still carry useful headers. return { "url": current, "status": e.code, "headers": {k.lower(): v for k, v in e.headers.items()}, "body": b"", "chain": chain, "elapsed_ms": int((time.time() - started) * 1000), } except ssl.SSLError as e: raise SafeFetchError("TLS error: %s" % (getattr(e, "reason", None) or e)) except socket.timeout: raise SafeFetchError("Timed out after %ds." % timeout) except urllib.error.URLError as e: raise SafeFetchError("Could not connect: %s" % e.reason) raise SafeFetchError("Too many redirects (limit %d)." % max_redirects)