Coverage for src/local_deep_research/security/ssrf_validator.py: 98%
149 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 # remain blocked unconditionally.
296 if nat64_allowed and blocked_range in NAT64_PREFIXES:
297 # The opt-in permits reaching only what a DIRECT connection
298 # to the embedded IPv4 would be allowed to reach — it is not
299 # a blanket bypass of the private / loopback / link-local /
300 # metadata policy. Re-apply the same policy to the embedded
301 # IPv4: block if the direct form would be blocked, otherwise
302 # allow the NAT64 reach. This closes the gap where
303 # allow_nat64=True ALONE made NAT64-wrapped RFC1918 /
304 # loopback reachable regardless of allow_private_ips.
305 # (Metadata and, when block_link_local is set, link-local
306 # wraps are already blocked above.) The embedded IPv4 is
307 # extracted per RFC 6052 §2.2 (byte positions vary by prefix
308 # length) so a crafted /48 form cannot hide a blocked target
309 # at the RFC positions behind a decoy in the trailing 32
310 # bits; fail closed if extraction is impossible. Recursion
311 # terminates at depth 1: the embedded value is plain IPv4,
312 # never in a NAT64 prefix, and allow_nat64=False prevents
313 # re-entry.
314 embedded_v4 = nat64_embedded_ipv4(ip)
315 if embedded_v4 is None or is_ip_blocked(
316 str(embedded_v4),
317 allow_localhost=allow_localhost,
318 allow_private_ips=allow_private_ips,
319 allow_nat64=False,
320 block_link_local=block_link_local,
321 ):
322 return True
323 continue
324 # If allow_private_ips is True, skip blocking for private + loopback
325 if allow_private_ips:
326 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
327 is_private = any(ip in pr for pr in PRIVATE_RANGES)
328 # Notification path: link-local stays blocked even under
329 # the private-IP opt-in (metadata lives here beyond the
330 # always-blocked literals; no legitimate self-hosted
331 # notifier does). Fires before the private/loopback skip so
332 # it cannot be un-blocked by it. Metadata literals already
333 # returned True above, so this only governs the rest of the
334 # link-local range.
335 if block_link_local and any(
336 ip in llr for llr in LINK_LOCAL_RANGES
337 ):
338 return True
339 if is_loopback or is_private:
340 continue
341 # If allow_localhost is True, skip blocking for loopback only
342 elif allow_localhost:
343 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
344 if is_loopback:
345 continue
346 return True
348 return False
350 except ValueError:
351 # Invalid IP address
352 return False
355def validate_url(
356 url: str,
357 allow_localhost: bool = False,
358 allow_private_ips: bool = False,
359 block_link_local: bool = False,
360) -> bool:
361 """
362 Validate URL to prevent SSRF attacks.
364 ``block_link_local`` keeps the whole link-local range -- IPv4
365 ``169.254.0.0/16`` and IPv6 ``fe80::/10``, plus the NAT64-wrapped form --
366 blocked EVEN under ``allow_private_ips=True``. It is a DEFENCE-IN-DEPTH
367 control: callers that set it are narrowing an intentionally permissive
368 ``allow_private_ips`` so cloud instance metadata cannot hide inside it. It is forwarded to
369 ``is_ip_blocked`` at both the literal-IP and the resolved-hostname check,
370 so a DNS name pointing at link-local is caught too. Off by default: callers
371 that legitimately accept private targets keep today's behaviour unless they
372 opt in.
374 Checks:
375 1. URL scheme is allowed (http/https only)
376 2. Hostname is not an internal/private IP address
377 3. Hostname does not resolve to an internal/private IP
379 Args:
380 url: URL to validate
381 allow_localhost: Whether to allow localhost/loopback addresses.
382 Set to True for trusted internal services like self-hosted
383 search engines (e.g., searxng). Default False.
384 allow_private_ips: Whether to allow all private/internal IPs plus localhost.
385 This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
386 used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
387 private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
388 like SearXNG or Ollama in containerized environments.
389 Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
390 (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
391 are ALWAYS blocked regardless of these flags.
393 Returns:
394 True if URL is safe, False otherwise
395 """
396 if not isinstance(url, str):
397 return False
398 try:
399 url = url.strip()
400 # Layer 1: reject RFC-illegal characters that drive parser-differential
401 # attacks (backslash, whitespace, control bytes). The URL is omitted
402 # from this log line because userinfo (RFC 3986 §3.2.1) may contain
403 # credentials and rejected URLs are by definition adversarial-shaped.
404 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
405 logger.warning("Blocked URL containing RFC-illegal characters")
406 return False
408 parsed = urlparse(url)
410 # Check scheme
411 if parsed.scheme.lower() not in ALLOWED_SCHEMES:
412 logger.warning(
413 f"Blocked URL with invalid scheme: {parsed.scheme} - {redact_url_for_log(url)}"
414 )
415 return False
417 # Layer 2: extract host using urllib3, the same parser ``requests``
418 # uses internally. ``urlparse`` and urllib3 disagree on URLs like
419 # ``http://127.0.0.1\@1.1.1.1`` — urlparse says ``1.1.1.1``,
420 # urllib3 says ``127.0.0.1``. Validating against urllib3 means the
421 # validator and the HTTP client cannot disagree on destination.
422 try:
423 u3 = parse_url(url)
424 except LocationParseError:
425 logger.warning("Blocked URL: urllib3 parser rejected it")
426 return False
427 hostname = u3.host
428 # Authority must be ASCII printable. urllib3 currently rejects
429 # non-ASCII via LocationParseError, but this guard keeps us
430 # independent of that staying constant — CVE-2019-9636 showed
431 # Python's stdlib loosened a similar restriction previously.
432 # Brackets/colon used in IPv6 hosts are within 0x20-0x7e, so this
433 # runs cleanly before bracket-strip.
434 if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname): 434 ↛ 435line 434 didn't jump to line 435 because the condition on line 434 was never true
435 logger.warning("Blocked URL with non-ASCII / control bytes in host")
436 return False
437 # Strip IPv6 brackets so ipaddress.ip_address can parse the host.
438 if hostname and hostname.startswith("[") and hostname.endswith("]"):
439 hostname = hostname[1:-1]
440 # rstrip(".") matches getaddrinfo behaviour — trailing dots are
441 # ignored at resolution time.
442 if hostname:
443 hostname = hostname.rstrip(".")
444 if not hostname:
445 logger.warning(
446 f"Blocked URL with no hostname: {redact_url_for_log(url)}"
447 )
448 return False
450 # Check if hostname is an IP address
451 try:
452 ip = ipaddress.ip_address(hostname)
453 if is_ip_blocked(
454 str(ip),
455 allow_localhost=allow_localhost,
456 allow_private_ips=allow_private_ips,
457 block_link_local=block_link_local,
458 ):
459 logger.warning(
460 f"Blocked URL with internal/private IP: {hostname} - {redact_url_for_log(url)}"
461 )
462 return False
463 except ValueError:
464 if is_percent_encoded_numeric_ipv4_host(hostname):
465 logger.warning("Blocked URL with encoded numeric IPv4 host")
466 return False
467 if is_ambiguous_numeric_ipv4_host(hostname):
468 logger.warning("Blocked URL with ambiguous numeric IPv4 host")
469 return False
471 # Resolve hostname to IP and check.
472 #
473 # NOTE: this is the validation-time check. On the ``safe_requests``
474 # path it is now backed by ``security.dns_pinning``, which pins the
475 # address validated here (or re-resolved+re-validated at connect
476 # time) so requests/urllib3 connect to exactly that address rather
477 # than re-resolving the hostname independently — closing the
478 # resolve-vs-connect gap for that path. Callers that hand the URL to
479 # an external client that re-resolves on its own (e.g. an LLM SDK,
480 # Apprise) still carry the residual window; see SECURITY.md.
481 try:
482 # Get all IP addresses for hostname
483 # nosec B104 - DNS resolution is intentional for SSRF prevention (checking if hostname resolves to private IP)
484 addr_info = socket.getaddrinfo(
485 hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
486 )
488 for info in addr_info:
489 ip_str = str(
490 info[4][0]
491 ) # Extract IP address from addr_info tuple
493 if is_ip_blocked(
494 ip_str,
495 allow_localhost=allow_localhost,
496 allow_private_ips=allow_private_ips,
497 block_link_local=block_link_local,
498 ):
499 logger.warning(
500 f"Blocked URL - hostname {hostname} resolves to "
501 f"internal/private IP: {ip_str} - {redact_url_for_log(url)}"
502 )
503 return False
505 except socket.gaierror:
506 logger.warning(f"Failed to resolve hostname {hostname}")
507 return False
508 except Exception:
509 logger.exception("Error during hostname resolution")
510 return False
512 # URL passes all checks
513 return True
515 except Exception:
516 logger.exception(f"Error validating URL {redact_url_for_log(url)}")
517 return False
520def assert_base_url_safe(base_url: str, *, setting_key: str) -> str:
521 """Validate an LLM provider base_url. Raises ValueError on SSRF.
523 Args:
524 base_url: The URL to validate.
525 setting_key: The settings dot-path that produced this URL
526 (e.g. ``"llm.ollama.url"``). Embedded into the error message
527 so operators know which setting to fix. Pass ``cls.url_setting``
528 from the OpenAI-compat parent or ``"llm.ollama.url"`` from
529 the Ollama provider — NEVER ``cls.provider_name`` which is a
530 display string ("xAI Grok", "llama.cpp") not a settings key.
532 Uses ``allow_localhost=True, allow_private_ips=True`` because the
533 legitimate destinations for LLM SDKs are localhost (Ollama, LM Studio,
534 llama.cpp) and RFC1918 (Docker / private network deployments). The
535 ``ALWAYS_BLOCKED_METADATA_IPS`` set still fires under those flags and
536 prevents the auth-gated SSRF that would otherwise reach cloud-credential
537 endpoints (AWS IMDS / ECS, Azure, OCI, DigitalOcean, AlibabaCloud,
538 Tencent).
540 Residual risk (documented, not closed here): this guard validates once
541 at provider construction, and the LLM SDK re-resolves the hostname on
542 every inference call through its own HTTP client. Unlike the
543 ``safe_requests`` path — which pins the validated address via
544 ``security.dns_pinning`` so the connection cannot be re-steered — the
545 SDK exposes no resolver/adapter seam to pin without patching its
546 internals, so the resolve-vs-connect window remains. Egress restriction
547 at the firewall is the operator-side mitigation. See SECURITY.md
548 ("LLM Provider URL Validation").
549 """
550 if not validate_url(base_url, allow_localhost=True, allow_private_ips=True):
551 raise ValueError(
552 f"base_url failed SSRF validation: refusing to send "
553 f"inference traffic. Check {setting_key} config."
554 )
555 return base_url
558def get_safe_url(
559 url: Optional[str], default: Optional[str] = None
560) -> Optional[str]:
561 """
562 Get URL if it's safe, otherwise return default.
564 Args:
565 url: URL to validate
566 default: Default value if URL is unsafe
568 Returns:
569 URL if safe, default otherwise
570 """
571 if not url:
572 return default
574 if validate_url(url):
575 return url
577 logger.warning(f"Unsafe URL rejected: {redact_url_for_log(url)}")
578 return default
581def redact_url_for_log(url: str) -> str:
582 """Return ``scheme://host:port`` (no userinfo, path, query, fragment).
584 For log output only. Drops everything except scheme + authority host
585 + port to minimise the chance of leaking credentials, tokens, or
586 sensitive paths into logs while still giving operators enough to
587 distinguish ``http://10.0.0.1:80`` from ``https://10.0.0.1:443``.
589 RFC 3986 §3.2.1 allows credentials in URL userinfo
590 (``http://user:pass@host/``). A rejected URL is by definition
591 adversarial-shaped, but it may still carry the operator's real
592 credentials if a misconfiguration produced it.
593 """
594 try:
595 u = parse_url(url)
596 scheme = u.scheme or "?"
597 host = u.host or "<no-host>"
598 host_port = f"{host}:{u.port}" if u.port else host
599 return f"{scheme}://{host_port}"
600 except (LocationParseError, ValueError):
601 return "<unparseable>"