Coverage for src/local_deep_research/security/ssrf_validator.py: 98%
149 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-02 13:53 +0000
1"""
2URL Validator for SSRF Prevention
4Validates URLs to prevent Server-Side Request Forgery (SSRF) attacks
5by blocking requests to internal/private networks and enforcing safe schemes.
6"""
8import ipaddress
9import re
10import socket
11from urllib.parse import urlparse
12from typing import Optional
13from loguru import logger
14from urllib3.exceptions import LocationParseError
15from urllib3.util import parse_url
17from .ip_ranges import PRIVATE_IP_RANGES as BLOCKED_IP_RANGES
18from .ip_ranges import NAT64_PREFIXES
19from .legacy_ipv4 import (
20 is_ambiguous_numeric_ipv4_host,
21 is_percent_encoded_numeric_ipv4_host,
22)
24# Cloud-provider metadata endpoints — always blocked, even with
25# allow_localhost=True or allow_private_ips=True. These IPs expose IAM /
26# instance-role credentials and are never legitimate destinations.
27#
28# Entries are matched by canonical string form: is_ip_blocked parses the
29# candidate with ipaddress.ip_address first, so any textual variant
30# (uppercase / zero-padded / expanded IPv6) is normalized to the canonical
31# form below before the membership test.
32# nosec B104 - Hardcoded IPs are intentional for SSRF prevention
33ALWAYS_BLOCKED_METADATA_IPS = frozenset(
34 {
35 "169.254.169.254", # AWS IMDSv1/v2, Azure, OCI, DigitalOcean
36 "169.254.170.2", # AWS ECS task metadata v3
37 "169.254.170.23", # AWS ECS task metadata v4
38 "169.254.0.23", # Tencent Cloud
39 "100.100.100.200", # AlibabaCloud
40 # AWS native IPv6 IMDS endpoint. Documented by AWS as the IPv6
41 # instance-metadata address ([fd00:ec2::254]). It is a ULA
42 # (fc00::/7), NOT an IPv4-mapped / NAT64-wrapped form of
43 # 169.254.169.254, so the IPv4 entries above and the NAT64
44 # embedded-IPv4 check do not cover it — it must be listed
45 # explicitly or it stays reachable under allow_private_ips=True
46 # (which permits fc00::/7).
47 "fd00:ec2::254", # AWS IMDS over IPv6
48 }
49)
51# Link-local ranges. When ``block_link_local`` is set (notification path),
52# these stay blocked even under ``allow_private_ips=True`` — cloud-provider
53# metadata lives here beyond the always-blocked literals and no legitimate
54# self-hosted notifier does. Kept as a distinct list (a subset of the private
55# ranges) so the carve-out is explicit and testable. Module-level (built once
56# at import) rather than rebuilt on every ``is_ip_blocked`` call.
57# nosec B104 - Hardcoded ranges are intentional for SSRF prevention
58LINK_LOCAL_RANGES = [
59 ipaddress.ip_network("169.254.0.0/16"), # IPv4 link-local
60 ipaddress.ip_network("fe80::/10"), # IPv6 link-local
61]
63# Allowed URL schemes
64ALLOWED_SCHEMES = {"http", "https"}
67# RFC 6052 §2.2 places the embedded IPv4 at prefix-length-specific byte
68# positions — NOT always the trailing 32 bits. Only /96 puts the IPv4 in the
69# low 32 bits; shorter prefixes split it around the reserved octet at byte 8.
70# Reading the wrong bytes lets a crafted address (e.g. a /48 form) hide a real
71# metadata/internal target at the RFC positions while a decoy sits in the
72# trailing 32 bits. Map each supported prefix length to the four byte indices
73# (of the 16-byte address) that hold the embedded IPv4, in order.
74#
75# All six RFC 6052 lengths are mapped on purpose, though NAT64_PREFIXES
76# currently carries only /48 and /96 — so the 32/40/56/64 rows are unreachable
77# today and untested. That is deliberate: adding a prefix to NAT64_PREFIXES
78# should not also require editing this table, because a missing row here fails
79# CLOSED but silently (extraction returns None, the caller blocks, and a
80# legitimate NAT64 deployment simply stops resolving). A spec-complete table
81# makes that impossible. Anything added here must come from RFC 6052 §2.2, not
82# from inference.
83_NAT64_V4_BYTE_INDICES = {
84 32: (4, 5, 6, 7),
85 40: (5, 6, 7, 9),
86 48: (6, 7, 9, 10),
87 56: (7, 9, 10, 11),
88 64: (9, 10, 11, 12),
89 96: (12, 13, 14, 15),
90}
93def nat64_embedded_ipv4(
94 ip: ipaddress._BaseAddress,
95) -> Optional[ipaddress.IPv4Address]:
96 """Return the IPv4 embedded in a NAT64-prefixed IPv6 address per RFC 6052
97 §2.2 (keyed on the matched prefix's length), or ``None`` if ``ip`` is not
98 inside a known NAT64 prefix (or the prefix length has no defined embedding).
100 Using the trailing 32 bits for every length misreads a non-/96 embedding —
101 see ``_NAT64_V4_BYTE_INDICES``. Centralized so ``is_ip_blocked`` and the
102 two wrap checks below extract identically and cannot drift.
104 Byte 8 — the octet RFC 6052 §2.2 reserves as zero for every embedding
105 shorter than /96 — is deliberately NOT validated: a malformed embedding is
106 policy-CHECKED, not rejected. Extraction reads the RFC byte positions
107 either way, so a crafted address still has its real embedded target
108 policy-checked (it cannot under-block). Rejecting malformed embeddings
109 would also fail closed (``None`` → the caller blocks), but would over-block
110 anything a lenient translator would in fact route.
111 """
112 if not isinstance(ip, ipaddress.IPv6Address):
113 return None
114 packed = ip.packed
115 for nat64_prefix in NAT64_PREFIXES:
116 if ip in nat64_prefix:
117 idx = _NAT64_V4_BYTE_INDICES.get(nat64_prefix.prefixlen)
118 if idx is None: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 return None
120 return ipaddress.IPv4Address(bytes(packed[i] for i in idx))
121 return None
124def is_nat64_wrapped_metadata_ip(ip: ipaddress._BaseAddress) -> bool:
125 """True iff ``ip`` is an IPv6 address inside a NAT64 prefix whose embedded
126 IPv4 (extracted per RFC 6052 §2.2) is in ``ALWAYS_BLOCKED_METADATA_IPS``.
128 Both ``is_ip_blocked`` and ``NotificationURLValidator._ip_matches_blocked_range``
129 consult this before honoring the ``security.allow_nat64`` operator
130 opt-in, so cloud-metadata access cannot be re-opened through an
131 IPv6-wrapped destination on a NAT64-equipped host.
132 """
133 embedded_v4 = nat64_embedded_ipv4(ip)
134 return (
135 embedded_v4 is not None
136 and str(embedded_v4) in ALWAYS_BLOCKED_METADATA_IPS
137 )
140def is_nat64_wrapped_link_local_ip(ip: ipaddress._BaseAddress) -> bool:
141 """True iff ``ip`` is an IPv6 address inside a NAT64 prefix whose embedded
142 IPv4 (per RFC 6052 §2.2) is IPv4 link-local (``169.254.0.0/16``).
144 Mirrors ``is_nat64_wrapped_metadata_ip`` for the ``block_link_local``
145 notification guard. The ``security.allow_nat64`` opt-in re-opens general
146 IPv4 reachability via NAT64, but it must not re-open link-local — where
147 cloud-provider metadata lives beyond the always-blocked literals (e.g.
148 Scaleway's ``169.254.42.42``). Consulted only when ``block_link_local`` is
149 set, so non-notification callers (which permit link-local under
150 ``allow_private_ips``) keep the opt-in's reachability unchanged.
151 """
152 embedded_v4 = nat64_embedded_ipv4(ip)
153 return embedded_v4 is not None and any(
154 isinstance(r, ipaddress.IPv4Network) and embedded_v4 in r
155 for r in LINK_LOCAL_RANGES
156 )
159# RFC 3986 forbids these characters in URLs; their presence in a URL signals
160# a parser-differential attempt (GHSA-g23j-2vwm-5c25). \s covers space, \t,
161# \n, \r, \v, \f. Backslash is the load-bearing payload — Python's urlparse
162# treats it as a literal char while requests/urllib3 treat it as a path
163# delimiter, so a crafted URL like ``http://127.0.0.1\@1.1.1.1`` would
164# pass the urlparse-based hostname check but actually connect to 127.0.0.1.
165RFC_FORBIDDEN_URL_CHARS_RE = re.compile(r"[\\\s\x00-\x1f\x7f]")
168def is_ip_blocked(
169 ip_str: str,
170 allow_localhost: bool = False,
171 allow_private_ips: bool = False,
172 allow_nat64: Optional[bool] = None,
173 block_link_local: bool = False,
174) -> bool:
175 """
176 Check if an IP address is in a blocked range.
178 Args:
179 ip_str: IP address as string
180 allow_localhost: Whether to allow localhost/loopback addresses
181 allow_private_ips: Whether to allow all private/internal IPs plus localhost.
182 This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
183 used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
184 private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
185 like SearXNG or Ollama in containerized environments.
186 Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
187 (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
188 are ALWAYS blocked regardless of these flags.
189 block_link_local: When True, the entire link-local range — IPv4
190 ``169.254.0.0/16`` and IPv6 ``fe80::/10`` — is treated as blocked
191 EVEN under ``allow_private_ips=True``. Used by the notification
192 send path (see ``security.dns_pinning`` /
193 ``notification_validator``): the lenient plugin/raw-webhook
194 partition allows private LAN targets, but link-local is where
195 cloud-provider metadata lives beyond the six always-blocked
196 literals (e.g. Scaleway's ``169.254.42.42``) and is never a
197 legitimate self-hosted notifier, so it stays blocked there. Has no
198 effect unless ``allow_private_ips=True`` — without the opt-in
199 link-local is already blocked. Default False preserves the
200 behavior every non-notification caller relies on (RFC1918 /
201 loopback / non-link-local ULA remain allowed under the flag).
202 allow_nat64: Override for the ``security.allow_nat64`` carve-out.
203 ``None`` (default) reads the env setting — the behavior every
204 existing caller relies on. An explicit ``bool`` answers a
205 hypothetical ("would enabling NAT64 unblock this?") without
206 mutating env; used by the notification "Test" admin hint to
207 decide whether to surface ``LDR_SECURITY_ALLOW_NAT64``. The
208 cloud-metadata always-block above fires first either way, so
209 this can never reopen IMDS.
211 Returns:
212 True if IP is blocked, False otherwise
213 """
214 # Loopback ranges that can be allowed for trusted internal services
215 # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
216 LOOPBACK_RANGES = [
217 ipaddress.ip_network("127.0.0.0/8"), # IPv4 loopback
218 ipaddress.ip_network("::1/128"), # IPv6 loopback
219 ]
221 # Private/internal network ranges - allowed with allow_private_ips=True
222 # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
223 PRIVATE_RANGES = [
224 # RFC1918 Private Ranges
225 ipaddress.ip_network("10.0.0.0/8"), # Class A private
226 ipaddress.ip_network("172.16.0.0/12"), # Class B private
227 ipaddress.ip_network("192.168.0.0/16"), # Class C private
228 # Container/Virtual Network Ranges
229 ipaddress.ip_network(
230 "100.64.0.0/10"
231 ), # CGNAT - used by Podman/rootless containers
232 ipaddress.ip_network(
233 "169.254.0.0/16"
234 ), # Link-local (cloud metadata IPs blocked separately via ALWAYS_BLOCKED_METADATA_IPS)
235 # IPv6 Private Ranges
236 ipaddress.ip_network("fc00::/7"), # IPv6 Unique Local Addresses
237 ipaddress.ip_network("fe80::/10"), # IPv6 Link-Local
238 ]
240 try:
241 ip = ipaddress.ip_address(ip_str)
243 # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1 → 127.0.0.1)
244 # These bypass IPv4 range checks if not converted.
245 if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
246 ip = ip.ipv4_mapped
248 # ALWAYS block cloud-metadata endpoints - critical SSRF target
249 # for credential theft (AWS IMDS/ECS, Azure, OCI, DigitalOcean,
250 # AlibabaCloud, Tencent Cloud). These are never legitimate
251 # destinations regardless of allow_localhost / allow_private_ips.
252 if str(ip) in ALWAYS_BLOCKED_METADATA_IPS:
253 return True
255 # Also block metadata IPs reached via NAT64 wrap. NAT64 prefixes
256 # embed the IPv4 destination in the low 32 bits; even when the
257 # operator has set LDR_SECURITY_ALLOW_NAT64=true the metadata
258 # block is "always" — an opt-in for IPv4 reachability does NOT
259 # license IMDS exposure.
260 if is_nat64_wrapped_metadata_ip(ip):
261 return True
263 # For the notification send path (``block_link_local``), a NAT64-wrapped
264 # link-local address must also stay blocked even under the allow_nat64
265 # opt-in — otherwise the carve-out below `continue`s past the link-local
266 # check for the IPv6-wrapped form of e.g. Scaleway's 169.254.42.42.
267 # (When NAT64 is off the whole prefix is already blocked in the loop, so
268 # this only changes the opt-in case; gated on block_link_local so
269 # non-notification callers are unaffected.)
270 if block_link_local and is_nat64_wrapped_link_local_ip(ip):
271 return True
273 # Operator escape hatch for IPv6-only deployments using DNS64+NAT64.
274 # Read lazily (not at import) so test monkeypatching works and so the
275 # value is not cached across env mutations. Cloud-metadata IPs are
276 # ALWAYS blocked above, so this carve-out cannot reopen IMDS via
277 # the IPv6-wrapped form.
278 #
279 # allow_nat64 overrides the env read: None (default) preserves every
280 # existing caller; an explicit bool lets the notification "Test"
281 # admin hint ask "would LDR_SECURITY_ALLOW_NAT64=true unblock this?"
282 # without touching process env.
283 if allow_nat64 is None:
284 from ..settings.env_registry import get_env_setting
286 nat64_allowed = bool(get_env_setting("security.allow_nat64", False))
287 else:
288 nat64_allowed = allow_nat64
290 # Check if IP is in any blocked range
291 for blocked_range in BLOCKED_IP_RANGES:
292 if ip in blocked_range:
293 # NAT64 carve-out: when the operator has opted in, the two
294 # NAT64 prefixes don't block outright. 6to4 / Teredo / discard
295 # / IPv4-Compatible (::/96, except ::1) / IPv4-Translated
296 # (::ffff:0:0:0/96) remain blocked unconditionally.
297 if nat64_allowed and blocked_range in NAT64_PREFIXES:
298 # The opt-in permits reaching only what a DIRECT connection
299 # to the embedded IPv4 would be allowed to reach — it is not
300 # a blanket bypass of the private / loopback / link-local /
301 # metadata policy. Re-apply the same policy to the embedded
302 # IPv4: block if the direct form would be blocked, otherwise
303 # allow the NAT64 reach. This closes the gap where
304 # allow_nat64=True ALONE made NAT64-wrapped RFC1918 /
305 # loopback reachable regardless of allow_private_ips.
306 # (Metadata and, when block_link_local is set, link-local
307 # wraps are already blocked above.) The embedded IPv4 is
308 # extracted per RFC 6052 §2.2 (byte positions vary by prefix
309 # length) so a crafted /48 form cannot hide a blocked target
310 # at the RFC positions behind a decoy in the trailing 32
311 # bits; fail closed if extraction is impossible. Recursion
312 # terminates at depth 1: the embedded value is plain IPv4,
313 # never in a NAT64 prefix, and allow_nat64=False prevents
314 # re-entry.
315 embedded_v4 = nat64_embedded_ipv4(ip)
316 if embedded_v4 is None or is_ip_blocked(
317 str(embedded_v4),
318 allow_localhost=allow_localhost,
319 allow_private_ips=allow_private_ips,
320 allow_nat64=False,
321 block_link_local=block_link_local,
322 ):
323 return True
324 continue
325 # If allow_private_ips is True, skip blocking for private + loopback
326 if allow_private_ips:
327 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
328 is_private = any(ip in pr for pr in PRIVATE_RANGES)
329 # Notification path: link-local stays blocked even under
330 # the private-IP opt-in (metadata lives here beyond the
331 # always-blocked literals; no legitimate self-hosted
332 # notifier does). Fires before the private/loopback skip so
333 # it cannot be un-blocked by it. Metadata literals already
334 # returned True above, so this only governs the rest of the
335 # link-local range.
336 if block_link_local and any(
337 ip in llr for llr in LINK_LOCAL_RANGES
338 ):
339 return True
340 if is_loopback or is_private:
341 continue
342 # If allow_localhost is True, skip blocking for loopback only
343 elif allow_localhost:
344 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
345 if is_loopback:
346 continue
347 return True
349 return False
351 except ValueError:
352 # Invalid IP address
353 return False
356def validate_url(
357 url: str,
358 allow_localhost: bool = False,
359 allow_private_ips: bool = False,
360 block_link_local: bool = False,
361) -> bool:
362 """
363 Validate URL to prevent SSRF attacks.
365 ``block_link_local`` keeps the whole link-local range -- IPv4
366 ``169.254.0.0/16`` and IPv6 ``fe80::/10``, plus the NAT64-wrapped form --
367 blocked EVEN under ``allow_private_ips=True``. It is a DEFENCE-IN-DEPTH
368 control: callers that set it are narrowing an intentionally permissive
369 ``allow_private_ips`` so cloud instance metadata cannot hide inside it. It is forwarded to
370 ``is_ip_blocked`` at both the literal-IP and the resolved-hostname check,
371 so a DNS name pointing at link-local is caught too. Off by default: callers
372 that legitimately accept private targets keep today's behaviour unless they
373 opt in.
375 Checks:
376 1. URL scheme is allowed (http/https only)
377 2. Hostname is not an internal/private IP address
378 3. Hostname does not resolve to an internal/private IP
380 Args:
381 url: URL to validate
382 allow_localhost: Whether to allow localhost/loopback addresses.
383 Set to True for trusted internal services like self-hosted
384 search engines (e.g., searxng). Default False.
385 allow_private_ips: Whether to allow all private/internal IPs plus localhost.
386 This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
387 used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
388 private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
389 like SearXNG or Ollama in containerized environments.
390 Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
391 (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
392 are ALWAYS blocked regardless of these flags.
394 Returns:
395 True if URL is safe, False otherwise
396 """
397 if not isinstance(url, str):
398 return False
399 try:
400 url = url.strip()
401 # Layer 1: reject RFC-illegal characters that drive parser-differential
402 # attacks (backslash, whitespace, control bytes). The URL is omitted
403 # from this log line because userinfo (RFC 3986 §3.2.1) may contain
404 # credentials and rejected URLs are by definition adversarial-shaped.
405 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
406 logger.warning("Blocked URL containing RFC-illegal characters")
407 return False
409 parsed = urlparse(url)
411 # Check scheme
412 if parsed.scheme.lower() not in ALLOWED_SCHEMES:
413 logger.warning(
414 f"Blocked URL with invalid scheme: {parsed.scheme} - {redact_url_for_log(url)}"
415 )
416 return False
418 # Layer 2: extract host using urllib3, the same parser ``requests``
419 # uses internally. ``urlparse`` and urllib3 disagree on URLs like
420 # ``http://127.0.0.1\@1.1.1.1`` — urlparse says ``1.1.1.1``,
421 # urllib3 says ``127.0.0.1``. Validating against urllib3 means the
422 # validator and the HTTP client cannot disagree on destination.
423 try:
424 u3 = parse_url(url)
425 except LocationParseError:
426 logger.warning("Blocked URL: urllib3 parser rejected it")
427 return False
428 hostname = u3.host
429 # Authority must be ASCII printable. urllib3 currently rejects
430 # non-ASCII via LocationParseError, but this guard keeps us
431 # independent of that staying constant — CVE-2019-9636 showed
432 # Python's stdlib loosened a similar restriction previously.
433 # Brackets/colon used in IPv6 hosts are within 0x20-0x7e, so this
434 # runs cleanly before bracket-strip.
435 if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname): 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true
436 logger.warning("Blocked URL with non-ASCII / control bytes in host")
437 return False
438 # Strip IPv6 brackets so ipaddress.ip_address can parse the host.
439 if hostname and hostname.startswith("[") and hostname.endswith("]"):
440 hostname = hostname[1:-1]
441 # rstrip(".") matches getaddrinfo behaviour — trailing dots are
442 # ignored at resolution time.
443 if hostname:
444 hostname = hostname.rstrip(".")
445 if not hostname:
446 logger.warning(
447 f"Blocked URL with no hostname: {redact_url_for_log(url)}"
448 )
449 return False
451 # Check if hostname is an IP address
452 try:
453 ip = ipaddress.ip_address(hostname)
454 if is_ip_blocked(
455 str(ip),
456 allow_localhost=allow_localhost,
457 allow_private_ips=allow_private_ips,
458 block_link_local=block_link_local,
459 ):
460 logger.warning(
461 f"Blocked URL with internal/private IP: {hostname} - {redact_url_for_log(url)}"
462 )
463 return False
464 except ValueError:
465 if is_percent_encoded_numeric_ipv4_host(hostname):
466 logger.warning("Blocked URL with encoded numeric IPv4 host")
467 return False
468 if is_ambiguous_numeric_ipv4_host(hostname):
469 logger.warning("Blocked URL with ambiguous numeric IPv4 host")
470 return False
472 # Resolve hostname to IP and check.
473 #
474 # NOTE: this is the validation-time check. On the ``safe_requests``
475 # path it is now backed by ``security.dns_pinning``, which pins the
476 # address validated here (or re-resolved+re-validated at connect
477 # time) so requests/urllib3 connect to exactly that address rather
478 # than re-resolving the hostname independently — closing the
479 # resolve-vs-connect gap for that path. Callers that hand the URL to
480 # an external client that re-resolves on its own (e.g. an LLM SDK,
481 # Apprise) still carry the residual window; see SECURITY.md.
482 try:
483 # Get all IP addresses for hostname
484 # nosec B104 - DNS resolution is intentional for SSRF prevention (checking if hostname resolves to private IP)
485 addr_info = socket.getaddrinfo(
486 hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
487 )
489 for info in addr_info:
490 ip_str = str(
491 info[4][0]
492 ) # Extract IP address from addr_info tuple
494 if is_ip_blocked(
495 ip_str,
496 allow_localhost=allow_localhost,
497 allow_private_ips=allow_private_ips,
498 block_link_local=block_link_local,
499 ):
500 logger.warning(
501 f"Blocked URL - hostname {hostname} resolves to "
502 f"internal/private IP: {ip_str} - {redact_url_for_log(url)}"
503 )
504 return False
506 except socket.gaierror:
507 logger.warning(f"Failed to resolve hostname {hostname}")
508 return False
509 except Exception:
510 logger.exception("Error during hostname resolution")
511 return False
513 # URL passes all checks
514 return True
516 except Exception:
517 logger.exception(f"Error validating URL {redact_url_for_log(url)}")
518 return False
521def assert_base_url_safe(base_url: str, *, setting_key: str) -> str:
522 """Validate an LLM provider base_url. Raises ValueError on SSRF.
524 Args:
525 base_url: The URL to validate.
526 setting_key: The settings dot-path that produced this URL
527 (e.g. ``"llm.ollama.url"``). Embedded into the error message
528 so operators know which setting to fix. Pass ``cls.url_setting``
529 from the OpenAI-compat parent or ``"llm.ollama.url"`` from
530 the Ollama provider — NEVER ``cls.provider_name`` which is a
531 display string ("xAI Grok", "llama.cpp") not a settings key.
533 Uses ``allow_localhost=True, allow_private_ips=True`` because the
534 legitimate destinations for LLM SDKs are localhost (Ollama, LM Studio,
535 llama.cpp) and RFC1918 (Docker / private network deployments). The
536 ``ALWAYS_BLOCKED_METADATA_IPS`` set still fires under those flags and
537 prevents the auth-gated SSRF that would otherwise reach cloud-credential
538 endpoints (AWS IMDS / ECS, Azure, OCI, DigitalOcean, AlibabaCloud,
539 Tencent).
541 Residual risk (documented, not closed here): this guard validates once
542 at provider construction, and the LLM SDK re-resolves the hostname on
543 every inference call through its own HTTP client. Unlike the
544 ``safe_requests`` path — which pins the validated address via
545 ``security.dns_pinning`` so the connection cannot be re-steered — the
546 SDK exposes no resolver/adapter seam to pin without patching its
547 internals, so the resolve-vs-connect window remains. Egress restriction
548 at the firewall is the operator-side mitigation. See SECURITY.md
549 ("LLM Provider URL Validation").
550 """
551 if not validate_url(base_url, allow_localhost=True, allow_private_ips=True):
552 raise ValueError(
553 f"base_url failed SSRF validation: refusing to send "
554 f"inference traffic. Check {setting_key} config."
555 )
556 return base_url
559def get_safe_url(
560 url: Optional[str], default: Optional[str] = None
561) -> Optional[str]:
562 """
563 Get URL if it's safe, otherwise return default.
565 Args:
566 url: URL to validate
567 default: Default value if URL is unsafe
569 Returns:
570 URL if safe, default otherwise
571 """
572 if not url:
573 return default
575 if validate_url(url):
576 return url
578 logger.warning(f"Unsafe URL rejected: {redact_url_for_log(url)}")
579 return default
582def redact_url_for_log(url: str) -> str:
583 """Return ``scheme://host:port`` (no userinfo, path, query, fragment).
585 For log output only. Drops everything except scheme + authority host
586 + port to minimise the chance of leaking credentials, tokens, or
587 sensitive paths into logs while still giving operators enough to
588 distinguish ``http://10.0.0.1:80`` from ``https://10.0.0.1:443``.
590 RFC 3986 §3.2.1 allows credentials in URL userinfo
591 (``http://user:pass@host/``). A rejected URL is by definition
592 adversarial-shaped, but it may still carry the operator's real
593 credentials if a misconfiguration produced it.
594 """
595 try:
596 # requests strips leading whitespace before it fetches; parse_url does
597 # not, and would put the scheme in the host slot.
598 u = parse_url(url.lstrip() if isinstance(url, str) else url)
599 scheme = u.scheme or "?"
600 host = u.host or "<no-host>"
601 host_port = f"{host}:{u.port}" if u.port else host
602 return f"{scheme}://{host_port}"
603 except (LocationParseError, ValueError):
604 return "<unparseable>"