Parth
e7d75c3b05
* Rewrite comments in Google developer documentation style Rewrite the comments and docstrings across the backend core modules so they read plainly. The previous prose was accurate but dense and figurative, which made it slow to skim. Applies the Google developer documentation style guide: short sentences, active voice, present tense, American spelling, and no metaphors, idioms, or rhetorical asides. Replaces em-dash chains with separate sentences.
53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
"""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 guard, a hosted user could point `endpoint_url` at an internal
|
|
service or at the cloud metadata endpoint, 169.254.169.254, and have the server
|
|
fetch it. The connection test even returns part of the response. The guard
|
|
therefore refuses any URL that resolves to a non-public address.
|
|
|
|
The guard does nothing in local mode. A local install talking to
|
|
http://localhost:11434, which is Ollama, is the intended case. The guard applies
|
|
only to a 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
|