"""SSRF guard for the one place the server makes an outbound request to a user-supplied address: the BYOK `endpoint_url` (connection test + turns/chat). Without this, a hosted user could point endpoint_url at an internal service or the cloud metadata endpoint (169.254.169.254) and have the server fetch it — the connection test even echoes part of the response back. We refuse any URL that resolves to a non-public address. No-op in local mode: a local install talking to http://localhost:11434 (Ollama) is the normal, intended case — the guard only applies to the hosted, multi-user deployment where the endpoint comes from an untrusted visitor. """ import ipaddress import socket from urllib.parse import urlparse from . import auth def endpoint_block_reason(url: str) -> str | None: """A human-readable reason this URL must NOT be fetched server-side, or None if it's allowed. Resolves the host and rejects it if any resulting address is non-public (private, loopback, link-local/metadata, reserved, …). Checking at request time (not just on save) is deliberate: it resists a DNS record that flips to a private IP after the value was stored. """ if not auth.MULTI_USER: return None parsed = urlparse(url) if parsed.scheme not in ("http", "https"): return "the endpoint URL must start with http:// or https://" host = parsed.hostname if not host: return "the endpoint URL has no host" port = parsed.port or (443 if parsed.scheme == "https" else 80) try: infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) except socket.gaierror: return "the endpoint host could not be resolved" for info in infos: try: ip = ipaddress.ip_address(info[4][0]) except ValueError: return "the endpoint host resolved to an unrecognized address" # is_global is the strict allowlist: private/loopback/link-local/CGNAT # all report False, so this one check covers the metadata IP too. if not ip.is_global or ip.is_multicast or ip.is_reserved: return "the endpoint URL resolves to a non-public address" return None