Coverage for src/local_deep_research/security/ssrf_validator.py: 98%
125 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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
20# Cloud-provider metadata endpoints — always blocked, even with
21# allow_localhost=True or allow_private_ips=True. These IPs expose IAM /
22# instance-role credentials and are never legitimate destinations.
23# nosec B104 - Hardcoded IPs are intentional for SSRF prevention
24ALWAYS_BLOCKED_METADATA_IPS = frozenset(
25 {
26 "169.254.169.254", # AWS IMDSv1/v2, Azure, OCI, DigitalOcean
27 "169.254.170.2", # AWS ECS task metadata v3
28 "169.254.170.23", # AWS ECS task metadata v4
29 "169.254.0.23", # Tencent Cloud
30 "100.100.100.200", # AlibabaCloud
31 }
32)
34# Allowed URL schemes
35ALLOWED_SCHEMES = {"http", "https"}
38def is_nat64_wrapped_metadata_ip(ip: ipaddress._BaseAddress) -> bool:
39 """True iff ``ip`` is an IPv6 address inside a NAT64 prefix whose
40 embedded IPv4 (low 32 bits) is in ``ALWAYS_BLOCKED_METADATA_IPS``.
42 Both ``is_ip_blocked`` and ``NotificationURLValidator._ip_matches_blocked_range``
43 consult this before honoring the ``security.allow_nat64`` operator
44 opt-in, so cloud-metadata access cannot be re-opened through an
45 IPv6-wrapped destination on a NAT64-equipped host. Keeping the
46 extraction in one place prevents the two validators from drifting.
47 """
48 if not isinstance(ip, ipaddress.IPv6Address):
49 return False
50 for nat64_prefix in NAT64_PREFIXES:
51 if ip in nat64_prefix:
52 embedded_v4 = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
53 return str(embedded_v4) in ALWAYS_BLOCKED_METADATA_IPS
54 return False
57# RFC 3986 forbids these characters in URLs; their presence in a URL signals
58# a parser-differential attempt (GHSA-g23j-2vwm-5c25). \s covers space, \t,
59# \n, \r, \v, \f. Backslash is the load-bearing payload — Python's urlparse
60# treats it as a literal char while requests/urllib3 treat it as a path
61# delimiter, so a crafted URL like ``http://127.0.0.1\@1.1.1.1`` would
62# pass the urlparse-based hostname check but actually connect to 127.0.0.1.
63RFC_FORBIDDEN_URL_CHARS_RE = re.compile(r"[\\\s\x00-\x1f\x7f]")
66def is_ip_blocked(
67 ip_str: str,
68 allow_localhost: bool = False,
69 allow_private_ips: bool = False,
70 allow_nat64: Optional[bool] = None,
71) -> bool:
72 """
73 Check if an IP address is in a blocked range.
75 Args:
76 ip_str: IP address as string
77 allow_localhost: Whether to allow localhost/loopback addresses
78 allow_private_ips: Whether to allow all private/internal IPs plus localhost.
79 This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
80 used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
81 private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
82 like SearXNG or Ollama in containerized environments.
83 Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
84 (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
85 are ALWAYS blocked regardless of these flags.
86 allow_nat64: Override for the ``security.allow_nat64`` carve-out.
87 ``None`` (default) reads the env setting — the behavior every
88 existing caller relies on. An explicit ``bool`` answers a
89 hypothetical ("would enabling NAT64 unblock this?") without
90 mutating env; used by the notification "Test" admin hint to
91 decide whether to surface ``LDR_SECURITY_ALLOW_NAT64``. The
92 cloud-metadata always-block above fires first either way, so
93 this can never reopen IMDS.
95 Returns:
96 True if IP is blocked, False otherwise
97 """
98 # Loopback ranges that can be allowed for trusted internal services
99 # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
100 LOOPBACK_RANGES = [
101 ipaddress.ip_network("127.0.0.0/8"), # IPv4 loopback
102 ipaddress.ip_network("::1/128"), # IPv6 loopback
103 ]
105 # Private/internal network ranges - allowed with allow_private_ips=True
106 # nosec B104 - These hardcoded IPs are intentional for SSRF allowlist
107 PRIVATE_RANGES = [
108 # RFC1918 Private Ranges
109 ipaddress.ip_network("10.0.0.0/8"), # Class A private
110 ipaddress.ip_network("172.16.0.0/12"), # Class B private
111 ipaddress.ip_network("192.168.0.0/16"), # Class C private
112 # Container/Virtual Network Ranges
113 ipaddress.ip_network(
114 "100.64.0.0/10"
115 ), # CGNAT - used by Podman/rootless containers
116 ipaddress.ip_network(
117 "169.254.0.0/16"
118 ), # Link-local (cloud metadata IPs blocked separately via ALWAYS_BLOCKED_METADATA_IPS)
119 # IPv6 Private Ranges
120 ipaddress.ip_network("fc00::/7"), # IPv6 Unique Local Addresses
121 ipaddress.ip_network("fe80::/10"), # IPv6 Link-Local
122 ]
124 try:
125 ip = ipaddress.ip_address(ip_str)
127 # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1 → 127.0.0.1)
128 # These bypass IPv4 range checks if not converted.
129 if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
130 ip = ip.ipv4_mapped
132 # ALWAYS block cloud-metadata endpoints - critical SSRF target
133 # for credential theft (AWS IMDS/ECS, Azure, OCI, DigitalOcean,
134 # AlibabaCloud, Tencent Cloud). These are never legitimate
135 # destinations regardless of allow_localhost / allow_private_ips.
136 if str(ip) in ALWAYS_BLOCKED_METADATA_IPS:
137 return True
139 # Also block metadata IPs reached via NAT64 wrap. NAT64 prefixes
140 # embed the IPv4 destination in the low 32 bits; even when the
141 # operator has set LDR_SECURITY_ALLOW_NAT64=true the metadata
142 # block is "always" — an opt-in for IPv4 reachability does NOT
143 # license IMDS exposure.
144 if is_nat64_wrapped_metadata_ip(ip):
145 return True
147 # Operator escape hatch for IPv6-only deployments using DNS64+NAT64.
148 # Read lazily (not at import) so test monkeypatching works and so the
149 # value is not cached across env mutations. Cloud-metadata IPs are
150 # ALWAYS blocked above, so this carve-out cannot reopen IMDS via
151 # the IPv6-wrapped form.
152 #
153 # allow_nat64 overrides the env read: None (default) preserves every
154 # existing caller; an explicit bool lets the notification "Test"
155 # admin hint ask "would LDR_SECURITY_ALLOW_NAT64=true unblock this?"
156 # without touching process env.
157 if allow_nat64 is None:
158 from ..settings.env_registry import get_env_setting
160 nat64_allowed = bool(get_env_setting("security.allow_nat64", False))
161 else:
162 nat64_allowed = allow_nat64
164 # Check if IP is in any blocked range
165 for blocked_range in BLOCKED_IP_RANGES:
166 if ip in blocked_range:
167 # NAT64 carve-out: when the operator has opted in, the two
168 # NAT64 prefixes don't block. 6to4 / Teredo / discard remain
169 # blocked unconditionally.
170 if nat64_allowed and blocked_range in NAT64_PREFIXES:
171 continue
172 # If allow_private_ips is True, skip blocking for private + loopback
173 if allow_private_ips:
174 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
175 is_private = any(ip in pr for pr in PRIVATE_RANGES)
176 if is_loopback or is_private:
177 continue
178 # If allow_localhost is True, skip blocking for loopback only
179 elif allow_localhost:
180 is_loopback = any(ip in lr for lr in LOOPBACK_RANGES)
181 if is_loopback:
182 continue
183 return True
185 return False
187 except ValueError:
188 # Invalid IP address
189 return False
192def validate_url(
193 url: str,
194 allow_localhost: bool = False,
195 allow_private_ips: bool = False,
196) -> bool:
197 """
198 Validate URL to prevent SSRF attacks.
200 Checks:
201 1. URL scheme is allowed (http/https only)
202 2. Hostname is not an internal/private IP address
203 3. Hostname does not resolve to an internal/private IP
205 Args:
206 url: URL to validate
207 allow_localhost: Whether to allow localhost/loopback addresses.
208 Set to True for trusted internal services like self-hosted
209 search engines (e.g., searxng). Default False.
210 allow_private_ips: Whether to allow all private/internal IPs plus localhost.
211 This includes RFC1918 (10.x, 172.16-31.x, 192.168.x), CGNAT (100.64.x.x
212 used by Podman/rootless containers), link-local (169.254.x.x), and IPv6
213 private ranges (fc00::/7, fe80::/10). Use for trusted self-hosted services
214 like SearXNG or Ollama in containerized environments.
215 Note: cloud metadata endpoints in ``ALWAYS_BLOCKED_METADATA_IPS``
216 (AWS / Azure / OCI / DigitalOcean / AlibabaCloud / Tencent / ECS)
217 are ALWAYS blocked regardless of these flags.
219 Returns:
220 True if URL is safe, False otherwise
221 """
222 if not isinstance(url, str):
223 return False
224 try:
225 url = url.strip()
226 # Layer 1: reject RFC-illegal characters that drive parser-differential
227 # attacks (backslash, whitespace, control bytes). The URL is omitted
228 # from this log line because userinfo (RFC 3986 §3.2.1) may contain
229 # credentials and rejected URLs are by definition adversarial-shaped.
230 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
231 logger.warning("Blocked URL containing RFC-illegal characters")
232 return False
234 parsed = urlparse(url)
236 # Check scheme
237 if parsed.scheme.lower() not in ALLOWED_SCHEMES:
238 logger.warning(
239 f"Blocked URL with invalid scheme: {parsed.scheme} - {redact_url_for_log(url)}"
240 )
241 return False
243 # Layer 2: extract host using urllib3, the same parser ``requests``
244 # uses internally. ``urlparse`` and urllib3 disagree on URLs like
245 # ``http://127.0.0.1\@1.1.1.1`` — urlparse says ``1.1.1.1``,
246 # urllib3 says ``127.0.0.1``. Validating against urllib3 means the
247 # validator and the HTTP client cannot disagree on destination.
248 try:
249 u3 = parse_url(url)
250 except LocationParseError:
251 logger.warning("Blocked URL: urllib3 parser rejected it")
252 return False
253 hostname = u3.host
254 # Authority must be ASCII printable. urllib3 currently rejects
255 # non-ASCII via LocationParseError, but this guard keeps us
256 # independent of that staying constant — CVE-2019-9636 showed
257 # Python's stdlib loosened a similar restriction previously.
258 # Brackets/colon used in IPv6 hosts are within 0x20-0x7e, so this
259 # runs cleanly before bracket-strip.
260 if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname): 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 logger.warning("Blocked URL with non-ASCII / control bytes in host")
262 return False
263 # Strip IPv6 brackets so ipaddress.ip_address can parse the host.
264 if hostname and hostname.startswith("[") and hostname.endswith("]"):
265 hostname = hostname[1:-1]
266 # rstrip(".") matches getaddrinfo behaviour — trailing dots are
267 # ignored at resolution time.
268 if hostname:
269 hostname = hostname.rstrip(".")
270 if not hostname:
271 logger.warning(
272 f"Blocked URL with no hostname: {redact_url_for_log(url)}"
273 )
274 return False
276 # Check if hostname is an IP address
277 try:
278 ip = ipaddress.ip_address(hostname)
279 if is_ip_blocked(
280 str(ip),
281 allow_localhost=allow_localhost,
282 allow_private_ips=allow_private_ips,
283 ):
284 logger.warning(
285 f"Blocked URL with internal/private IP: {hostname} - {redact_url_for_log(url)}"
286 )
287 return False
288 except ValueError:
289 # Not an IP address, it's a hostname - need to resolve it
290 pass
292 # Resolve hostname to IP and check.
293 #
294 # NOTE: This is a best-effort, validation-time check. The caller
295 # (typically safe_requests) hands the URL to requests/urllib3
296 # afterwards, which resolves the hostname AGAIN at connect time --
297 # a DNS rebinding TOCTOU window. Closing it would require pinning
298 # the resolved IP into the outbound connection (HTTPAdapter shim
299 # with server_hostname for SNI), which is HTTPS-only and doesn't
300 # follow redirects cleanly. See SECURITY.md "Notification Webhook
301 # SSRF" subsection for the accepted-risk rationale (the same
302 # caveat applies here).
303 try:
304 # Get all IP addresses for hostname
305 # nosec B104 - DNS resolution is intentional for SSRF prevention (checking if hostname resolves to private IP)
306 addr_info = socket.getaddrinfo(
307 hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
308 )
310 for info in addr_info:
311 ip_str = str(
312 info[4][0]
313 ) # Extract IP address from addr_info tuple
315 if is_ip_blocked(
316 ip_str,
317 allow_localhost=allow_localhost,
318 allow_private_ips=allow_private_ips,
319 ):
320 logger.warning(
321 f"Blocked URL - hostname {hostname} resolves to "
322 f"internal/private IP: {ip_str} - {redact_url_for_log(url)}"
323 )
324 return False
326 except socket.gaierror:
327 logger.warning(f"Failed to resolve hostname {hostname}")
328 return False
329 except Exception:
330 logger.exception("Error during hostname resolution")
331 return False
333 # URL passes all checks
334 return True
336 except Exception:
337 logger.exception(f"Error validating URL {redact_url_for_log(url)}")
338 return False
341def assert_base_url_safe(base_url: str, *, setting_key: str) -> str:
342 """Validate an LLM provider base_url. Raises ValueError on SSRF.
344 Args:
345 base_url: The URL to validate.
346 setting_key: The settings dot-path that produced this URL
347 (e.g. ``"llm.ollama.url"``). Embedded into the error message
348 so operators know which setting to fix. Pass ``cls.url_setting``
349 from the OpenAI-compat parent or ``"llm.ollama.url"`` from
350 the Ollama provider — NEVER ``cls.provider_name`` which is a
351 display string ("xAI Grok", "llama.cpp") not a settings key.
353 Uses ``allow_localhost=True, allow_private_ips=True`` because the
354 legitimate destinations for LLM SDKs are localhost (Ollama, LM Studio,
355 llama.cpp) and RFC1918 (Docker / private network deployments). The
356 ``ALWAYS_BLOCKED_METADATA_IPS`` set still fires under those flags and
357 prevents the auth-gated SSRF that would otherwise reach cloud-credential
358 endpoints (AWS IMDS / ECS, Azure, OCI, DigitalOcean, AlibabaCloud,
359 Tencent).
361 Caveat: this guard validates once at provider construction. The SDK
362 re-resolves the hostname on every inference call, so a hostile DNS
363 authority can rebind between guard and connect — same TOCTOU as
364 ``validate_url``. See SECURITY.md for the accepted-risk rationale.
365 """
366 if not validate_url(base_url, allow_localhost=True, allow_private_ips=True):
367 raise ValueError(
368 f"base_url failed SSRF validation: refusing to send "
369 f"inference traffic. Check {setting_key} config."
370 )
371 return base_url
374def get_safe_url(
375 url: Optional[str], default: Optional[str] = None
376) -> Optional[str]:
377 """
378 Get URL if it's safe, otherwise return default.
380 Args:
381 url: URL to validate
382 default: Default value if URL is unsafe
384 Returns:
385 URL if safe, default otherwise
386 """
387 if not url:
388 return default
390 if validate_url(url):
391 return url
393 logger.warning(f"Unsafe URL rejected: {redact_url_for_log(url)}")
394 return default
397def redact_url_for_log(url: str) -> str:
398 """Return ``scheme://host:port`` (no userinfo, path, query, fragment).
400 For log output only. Drops everything except scheme + authority host
401 + port to minimise the chance of leaking credentials, tokens, or
402 sensitive paths into logs while still giving operators enough to
403 distinguish ``http://10.0.0.1:80`` from ``https://10.0.0.1:443``.
405 RFC 3986 §3.2.1 allows credentials in URL userinfo
406 (``http://user:pass@host/``). A rejected URL is by definition
407 adversarial-shaped, but it may still carry the operator's real
408 credentials if a misconfiguration produced it.
409 """
410 try:
411 u = parse_url(url)
412 scheme = u.scheme or "?"
413 host = u.host or "<no-host>"
414 host_port = f"{host}:{u.port}" if u.port else host
415 return f"{scheme}://{host_port}"
416 except (LocationParseError, ValueError):
417 return "<unparseable>"