""" One guarded outbound HTTP client, for every request whose URL is chosen by data rather than by us. Three separate features fetch third-party-controlled URLs — the documentation scraper, its sitemap follower, and the customer-website monitor. Each did its own unguarded `requests.get`, and one disabled TLS verification entirely. From inside the VPS that reaches Postgres, Redis, the PrestaShop Docker containers, the Caddy admin API and any cloud metadata endpoint. `push_service.is_allowed_endpoint` solves a different problem (a fixed allowlist of four push relays) and cannot be reused here, because a doc scraper must be able to reach arbitrary public hosts. So the rule here is the inverse: allow any public host, deny anything that resolves to a non-public address. Use `safe_httpx_request()` for all such fetches. Do not add a bare client.get(). """ from __future__ import annotations import ipaddress import logging import socket from typing import Optional from urllib.parse import urlparse import httpx logger = logging.getLogger(__name__) #: Only these schemes may ever be fetched. Blocks file://, gopher://, ftp://, #: and the data:/blob: family. ALLOWED_SCHEMES = ("http", "https") #: Redirects are followed manually so every hop is re-validated. A permissive #: first hop that 302s to 169.254.169.254 is the classic bypass. MAX_REDIRECTS = 5 DEFAULT_TIMEOUT = 15 class SsrfBlocked(ValueError): """Raised when a URL resolves somewhere we refuse to send a request.""" # ── DNS rebinding (TOCTOU) — closed by pinning, see _pin_url ───────────────── # Validating a hostname and then handing that *hostname* to the HTTP client is not # enough: the client resolves it a second time when it opens the socket, so an # attacker controlling DNS can answer the check with a public IP and the connect # with 127.0.0.1. We therefore connect to the exact IP we validated. def _ip_is_public(ip: str) -> bool: """False for loopback, private, link-local, multicast, reserved and unspecified.""" try: addr = ipaddress.ip_address(ip) except ValueError: return False return not ( addr.is_private or addr.is_loopback or addr.is_link_local # 169.254.0.0/16 — cloud metadata lives here or addr.is_multicast or addr.is_reserved or addr.is_unspecified ) def resolve_public_ips(hostname: str) -> list[str]: """ Resolve `hostname` and return its addresses, or raise if ANY is non-public. Every resolved address must be public, not merely the first: a hostname with both a public and a 127.0.0.1 record would otherwise pass validation and then connect to the private one, which is DNS-rebinding in a single lookup. """ try: infos = socket.getaddrinfo(hostname, None) except socket.gaierror as exc: raise SsrfBlocked(f"cannot resolve host {hostname!r}: {exc}") from exc ips = sorted({info[4][0] for info in infos}) if not ips: raise SsrfBlocked(f"host {hostname!r} resolved to no addresses") for ip in ips: if not _ip_is_public(ip): raise SsrfBlocked(f"host {hostname!r} resolves to non-public address {ip}") return ips def _pin_url(url: str, ip: str) -> str: """Rewrite `url` to address `ip` directly, preserving scheme/port/path. The caller must also send `Host: ` and, for https, the `sni_hostname` request extension — otherwise the server can't route the request and TLS verification would be attempted against the bare IP. """ parsed = urlparse(url) host = ip if ":" not in ip else f"[{ip}]" # bracket IPv6 literals netloc = f"{host}:{parsed.port}" if parsed.port else host return parsed._replace(netloc=netloc).geturl() def validate_url(url: str) -> str: """Validate scheme + host and return the hostname. Raises SsrfBlocked.""" parsed = urlparse(url) if parsed.scheme not in ALLOWED_SCHEMES: raise SsrfBlocked(f"scheme {parsed.scheme!r} is not allowed") host = parsed.hostname if not host: raise SsrfBlocked("URL has no host") # Reject credentials in the URL — they are never needed here and are a # common way to smuggle a different effective host past naive parsers. if parsed.username or parsed.password: raise SsrfBlocked("credentials in URL are not allowed") resolve_public_ips(host) return host def safe_httpx_request(client, method: str, url: str, **kwargs): """ Perform an httpx request that cannot be pointed at a non-public address. Two things have to be true, and each was a real bug here at some point: 1. Every REDIRECT HOP is validated, not just the URL you were handed. A client with follow_redirects=True defeats the guard entirely: a public host 302s to 169.254.169.254 and only hop one was ever checked. The passed `client` MUST be built with follow_redirects=False, and any follow_redirects in kwargs is dropped so a caller cannot re-open it. 2. The connection goes to the IP we VALIDATED, not to the hostname. A hostname handed to httpx is resolved again when the socket opens, so an attacker controlling DNS answers the check with a public IP and the connect with 127.0.0.1 (DNS rebinding). We rewrite the URL to the validated IP and carry the original hostname in the Host header and, for https, in the `sni_hostname` extension — which drives both SNI and certificate hostname verification, so TLS stays fully verified. """ kwargs.pop("follow_redirects", None) # Popped ONCE, outside the loop: popping per-hop silently dropped the # caller's User-Agent on every redirect after the first. caller_headers = dict(kwargs.pop("headers", None) or {}) caller_extensions = dict(kwargs.pop("extensions", None) or {}) current = url for hop in range(MAX_REDIRECTS + 1): parsed = urlparse(current) host = parsed.hostname validate_url(current) # scheme + credentials checks ips = resolve_public_ips(host) # raises SsrfBlocked if any is private headers = dict(caller_headers) headers["Host"] = parsed.netloc extensions = dict(caller_extensions) if parsed.scheme == "https": extensions["sni_hostname"] = host resp = client.request( method, _pin_url(current, ips[0]), headers=headers, extensions=extensions, follow_redirects=False, **kwargs, ) if resp.is_redirect: location = resp.headers.get("Location") if not location: return resp # Resolve against the HOSTNAME url, never the pinned-IP one, or a # relative Location would silently stay pinned to the old address. current = str(httpx.URL(current).join(location)) logger.debug("safe_httpx_request: redirect hop %d -> %s", hop + 1, current) continue return resp raise SsrfBlocked(f"too many redirects (>{MAX_REDIRECTS}) starting at {url}")