Coverage for src/local_deep_research/security/notification_validator.py: 95%
172 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Security validation for notification service URLs.
4This module provides validation for user-configured notification service URLs
5to prevent Server-Side Request Forgery (SSRF) attacks and other security issues.
6"""
8import ipaddress
9import socket
10from concurrent.futures import ThreadPoolExecutor, TimeoutError
11from typing import List, Optional, Tuple, Union
12from urllib.parse import urlparse
13from loguru import logger
14from urllib3.exceptions import LocationParseError
15from urllib3.util import parse_url
17from .ip_ranges import PRIVATE_IP_RANGES as _PRIVATE_IP_RANGES
18from .ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE, redact_url_for_log
20# Type alias for resolved IP addresses (avoids referencing the private
21# ipaddress._BaseAddress API).
22ResolvedIP = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
25class NotificationURLValidationError(ValueError):
26 """Raised when a notification service URL fails security validation."""
28 pass
31class NotificationURLValidator:
32 """Validates notification service URLs to prevent SSRF and other attacks."""
34 # Dangerous protocols that should never be used for notifications
35 BLOCKED_SCHEMES = (
36 "file", # Local file access
37 "ftp", # FTP can be abused for SSRF
38 "ftps", # Secure FTP can be abused for SSRF
39 "data", # Data URIs can leak sensitive data
40 "javascript", # XSS/code execution
41 "vbscript", # XSS/code execution
42 "about", # Browser internal
43 "blob", # Browser internal
44 )
46 # Allowed protocols for notification services
47 ALLOWED_SCHEMES = (
48 "http", # Webhook services
49 "https", # Webhook services (preferred)
50 "mailto", # Email notifications
51 "discord", # Discord webhooks
52 "slack", # Slack webhooks
53 "telegram", # Telegram bot API
54 "gotify", # Gotify notifications
55 "pushover", # Pushover notifications
56 "ntfy", # ntfy.sh notifications (http)
57 "ntfys", # ntfy.sh notifications (https)
58 "signal", # Signal via signal-api-rest container
59 "matrix", # Matrix protocol
60 "mattermost", # Mattermost webhooks
61 "rocketchat", # Rocket.Chat webhooks
62 "teams", # Microsoft Teams
63 "json", # Generic JSON webhooks
64 "xml", # Generic XML webhooks
65 "form", # Form-encoded webhooks
66 )
68 # Reuse shared private IP range definitions
69 PRIVATE_IP_RANGES = _PRIVATE_IP_RANGES
71 # Prefix emitted by validate_service_url when an http(s) URL targets a
72 # private/internal IP. Pinned as a class-level constant so the
73 # validator, the hint logic, and the call site (service.py) share a
74 # single source of truth.
75 PRIVATE_IP_REJECTION_PREFIX = "Blocked private/internal IP address:"
77 @staticmethod
78 def _ip_matches_blocked_range(
79 ip, allow_private_ips: bool = False, allow_nat64: Optional[bool] = None
80 ) -> bool:
81 """Block-decision for a parsed IP, delegating to
82 ``ssrf_validator.is_ip_blocked`` so the two validators share a
83 single source of truth.
85 Honors:
86 - ALWAYS_BLOCKED_METADATA_IPS (cloud metadata, absolute)
87 - is_nat64_wrapped_metadata_ip (NAT64-wrapped IMDS, absolute)
88 - security.allow_nat64 env carve-out for the two NAT64 prefixes
89 (overridable via ``allow_nat64``: None reads env, an explicit
90 bool answers the "would NAT64 unblock this?" hint probe)
91 - allow_private_ips: when True, RFC1918 / CGNAT / loopback /
92 link-local / IPv6 ULA are allowed BUT the two absolute checks
93 above still fire. This closes the historical bypass where
94 ``allow_private_ips=True`` skipped the host check entirely
95 and let metadata IPs through the notification path.
96 """
97 from .ssrf_validator import is_ip_blocked
99 return is_ip_blocked(
100 str(ip),
101 allow_private_ips=allow_private_ips,
102 allow_nat64=allow_nat64,
103 )
105 @staticmethod
106 def _resolve_hostname_ips(
107 hostname: str,
108 ) -> Optional[List[ResolvedIP]]:
109 """Resolve hostname to a list of ipaddress objects, with a 5s
110 timeout. Returns None on resolution failure or timeout.
112 Extracted from ``_is_private_ip`` so multiple policy decisions
113 can share a single resolution pass — see
114 ``validate_service_url_with_hint`` for the single-pass call
115 site that closes a DNS-rebinding TOCTOU window between the
116 default-level and elevated-level decisions.
117 """
118 # NOTE: best-effort, validation-time check. Apprise re-resolves
119 # the hostname when it actually sends the request (via
120 # requests/urllib3), so an attacker controlling DNS can serve a
121 # public IP here and a private IP at send time — a classic DNS
122 # rebinding TOCTOU window. Apprise exposes no Session/adapter/
123 # DNS hook to close this in code without fragile monkey-patching
124 # of its plugin internals.
125 #
126 # Because the window cannot be closed cleanly in code, the
127 # whole outbound-notification path is gated behind an env-only
128 # master switch (LDR_NOTIFICATIONS_ALLOW_OUTBOUND, default off);
129 # turning it on is the operator's explicit risk-acceptance. See
130 # SECURITY.md "Notification Webhook SSRF". Operators wanting to
131 # avoid the window entirely should prefer plugin schemes
132 # (discord://, slack://, ntfy://, ntfys://, gotify://,
133 # telegram://, mattermost://, etc.) that hardcode their
134 # endpoints instead of raw http(s):// webhooks.
135 #
136 # concurrent.futures for thread-safe timeout instead of
137 # socket.setdefaulttimeout() which is process-global.
138 try:
139 executor = ThreadPoolExecutor(max_workers=1)
140 try:
141 future = executor.submit(
142 socket.getaddrinfo,
143 hostname,
144 None,
145 socket.AF_UNSPEC,
146 socket.SOCK_STREAM,
147 )
148 resolved_ips = future.result(timeout=5)
149 finally:
150 executor.shutdown(wait=False, cancel_futures=True)
151 return [
152 ipaddress.ip_address(sockaddr[0])
153 for _family, _, _, _, sockaddr in resolved_ips
154 ]
155 except (socket.gaierror, OSError, TimeoutError, ValueError):
156 logger.warning(
157 "DNS resolution failed for hostname {} — "
158 "allowing request (unable to determine if private)",
159 hostname,
160 )
161 return None
163 @staticmethod
164 def _host_blocks_at_level(
165 hostname: str,
166 allow_private_ips: bool = False,
167 allow_nat64: Optional[bool] = None,
168 resolved_ips: Optional[List[ResolvedIP]] = None,
169 ) -> bool:
170 """Decide whether ``hostname`` is blocked at the given
171 ``allow_private_ips`` / ``allow_nat64`` level, optionally reusing
172 a prior DNS resolution.
174 Pre-resolved IPs can be passed via ``resolved_ips`` to share a
175 single resolution across multiple policy decisions, closing
176 DNS-rebinding TOCTOU windows between calls. If ``resolved_ips``
177 is None and ``hostname`` is not an IP literal, DNS is resolved
178 via ``_resolve_hostname_ips``.
179 """
180 # Localhost-string shortcuts only apply when the operator hasn't
181 # opted into private-IP reachability. With allow_private_ips=True
182 # we let the IP path (DNS-resolved or literal) make the decision
183 # so metadata-IP literals like "169.254.169.254" still block.
184 if not allow_private_ips and hostname.lower() in (
185 "localhost",
186 "127.0.0.1",
187 "::1",
188 "0.0.0.0",
189 "::",
190 ):
191 return True
193 # Try to parse as IP address first.
194 try:
195 ip = ipaddress.ip_address(hostname)
196 return NotificationURLValidator._ip_matches_blocked_range(
197 ip,
198 allow_private_ips=allow_private_ips,
199 allow_nat64=allow_nat64,
200 )
201 except ValueError:
202 pass
204 # Hostname — use pre-resolved IPs if provided, otherwise resolve.
205 if resolved_ips is None:
206 resolved_ips = NotificationURLValidator._resolve_hostname_ips(
207 hostname
208 )
209 if resolved_ips is None:
210 # DNS failed — allow (matches the historical behaviour).
211 return False
212 for ip in resolved_ips:
213 if NotificationURLValidator._ip_matches_blocked_range(
214 ip,
215 allow_private_ips=allow_private_ips,
216 allow_nat64=allow_nat64,
217 ):
218 return True
219 return False
221 @staticmethod
222 def _extract_host(url: str) -> Optional[str]:
223 """Extract the raw hostname from ``url`` using urllib3's
224 ``parse_url`` (the same parser requests uses internally).
225 Returns None if urllib3 rejects the URL.
226 """
227 try:
228 u3 = parse_url(url)
229 except LocationParseError:
230 return None
231 return u3.host
233 @staticmethod
234 def _normalize_host(hostname: Optional[str]) -> str:
235 """Normalize a raw hostname from ``parse_url``: strip IPv6
236 bracket notation and trailing dots. Shared by
237 ``validate_service_url`` and ``validate_service_url_with_hint``
238 so the two call sites cannot drift.
239 """
240 if hostname and hostname.startswith("[") and hostname.endswith("]"):
241 hostname = hostname[1:-1]
242 if hostname:
243 hostname = hostname.rstrip(".")
244 return hostname or ""
246 @staticmethod
247 def _is_private_ip(
248 hostname: str,
249 allow_private_ips: bool = False,
250 allow_nat64: Optional[bool] = None,
251 _resolved_ips: Optional[List[ResolvedIP]] = None,
252 ) -> bool:
253 """
254 Check if hostname resolves to a private IP address.
256 Args:
257 hostname: Hostname to check
258 allow_private_ips: When True, RFC1918 / CGNAT / loopback /
259 link-local / IPv6 ULA are NOT considered private. Cloud
260 metadata IPs and NAT64-wrapped metadata IPs are blocked
261 regardless — the operator opt-in cannot license IMDS
262 exposure.
263 allow_nat64: Override for the ``security.allow_nat64`` carve-out.
264 None (default) reads the env setting; an explicit bool
265 answers the hint probe.
267 Returns:
268 True if hostname is a private IP or localhost (subject to
269 allow_private_ips), or wraps a metadata IP unconditionally
270 """
271 return NotificationURLValidator._host_blocks_at_level(
272 hostname,
273 allow_private_ips=allow_private_ips,
274 allow_nat64=allow_nat64,
275 resolved_ips=_resolved_ips,
276 )
278 @staticmethod
279 def validate_service_url(
280 url: str,
281 allow_private_ips: bool = False,
282 allow_nat64: Optional[bool] = None,
283 ) -> Tuple[bool, Optional[str]]:
284 """Validate a notification service URL for security issues.
286 Thin public wrapper around ``_validate_service_url_impl``.
287 Always performs its own DNS resolution (no caller-supplied
288 resolution state).
290 Args:
291 url: Service URL to validate (e.g., "discord://webhook_id/token")
292 allow_private_ips: Whether to allow private IPs (default: False)
293 Set to True for development/testing environments
294 allow_nat64: Override for the ``security.allow_nat64``
295 carve-out. None (default) reads the env setting;
296 an explicit bool overrides it.
298 Returns:
299 Tuple of (is_valid, error_message)
300 """
301 return NotificationURLValidator._validate_service_url_impl(
302 url,
303 allow_private_ips=allow_private_ips,
304 allow_nat64=allow_nat64,
305 )
307 @staticmethod
308 def _validate_service_url_impl(
309 url: str,
310 allow_private_ips: bool = False,
311 allow_nat64: Optional[bool] = None,
312 resolved_ips: Optional[List[ResolvedIP]] = None,
313 ) -> Tuple[bool, Optional[str]]:
314 """Internal validation with optional pre-resolved DNS state.
316 ``resolved_ips`` is trusted internal state used by
317 ``validate_service_url_with_hint`` to share a single DNS
318 resolution across multiple policy decisions. Must NEVER be
319 exposed on the public API — callers could pass ``[]`` to
320 bypass hostname resolution and private-IP checks.
322 Returns:
323 Tuple of (is_valid, error_message)
324 - is_valid: True if URL passes security checks
325 - error_message: None if valid, error description if invalid
327 Examples:
328 >>> validate_service_url("discord://webhook_id/token")
329 (True, None)
331 >>> validate_service_url("file:///etc/passwd")
332 (False, "Blocked unsafe protocol: file")
334 >>> validate_service_url("http://localhost:5000/webhook")
335 (False, "Blocked private/internal IP address: localhost")
337 Caller contract:
338 ``notifications.service.NotificationService.test_service``
339 calls ``validate_service_url_with_hint`` (single-pass sibling
340 below) to obtain the validation result AND, when the URL is
341 rejected, a ``hint_would_help`` flag indicating whether the
342 rejection targets a recoverable destination — one that
343 ``LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true`` (for RFC1918 /
344 CGNAT / loopback / link-local / IPv6 ULA, plus NAT64-wrapped
345 non-metadata destinations via the NAT64 carve-out) would
346 actually unblock. If ``hint_would_help`` is False the URL
347 targets an always-blocked category (cloud-metadata IPs, 6to4,
348 Teredo, discard prefix, IPv4-mapped IPv6 of metadata,
349 NAT64-wrapped metadata) and the hint is suppressed because
350 naming the env var would mislead. The parametrized
351 integration test ``test_test_service_ip_rejection_matrix``
352 in tests/web/services/test_notification_coverage.py locks
353 this contract end-to-end across every IP category — if the
354 wording here changes, that test fails and the call site
355 needs updating.
356 """
357 if not url or not isinstance(url, str):
358 return False, "Service URL must be a non-empty string"
360 # Strip whitespace (must run before the RFC-illegal char check
361 # so legitimate URLs with surrounding whitespace are not rejected).
362 url = url.strip()
364 # Reject URLs containing characters that drive parser-differential
365 # SSRF bypasses (backslash, whitespace, control bytes) — see
366 # GHSA-g23j-2vwm-5c25. The URL is omitted from the log line because
367 # userinfo (RFC 3986 §3.2.1) may contain credentials and rejected
368 # URLs are by definition adversarial-shaped.
369 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
370 logger.warning(
371 "Blocked notification URL containing RFC-illegal characters"
372 )
373 return (
374 False,
375 "URL contains characters that are not allowed (whitespace, backslash, or control bytes)",
376 )
378 # Parse URL
379 try:
380 parsed = urlparse(url)
381 except Exception:
382 # Never echo the parser exception back to the caller: this error
383 # string is surfaced to the user by the test-URL endpoint, and the
384 # exception text can carry parser internals / stack-trace fragments
385 # (CWE-209, py/stack-trace-exposure). Log at WARNING without a
386 # traceback to match the sibling LocationParseError handler below
387 # — a malformed URL is benign user input, not a server fault, so an
388 # ERROR-level stack trace would only add noise.
389 logger.warning("Failed to parse service URL")
390 return False, "Invalid URL format"
392 # Check for scheme
393 if not parsed.scheme:
394 return False, "Service URL must have a protocol (e.g., https://)"
396 scheme = parsed.scheme.lower()
398 # Check for blocked schemes
399 if scheme in NotificationURLValidator.BLOCKED_SCHEMES:
400 logger.warning(
401 f"Blocked unsafe notification protocol: {scheme} in URL: {redact_url_for_log(url)}"
402 )
403 return False, f"Blocked unsafe protocol: {scheme}"
405 # Check for allowed schemes
406 if scheme not in NotificationURLValidator.ALLOWED_SCHEMES:
407 logger.warning(
408 f"Unknown notification protocol: {scheme} in URL: {redact_url_for_log(url)}"
409 )
410 return (
411 False,
412 f"Unsupported protocol: {scheme}. "
413 f"Allowed: {', '.join(NotificationURLValidator.ALLOWED_SCHEMES[:5])}...",
414 )
416 # Extract the host for any allowed scheme. We use urllib3 (the
417 # parser ``requests`` uses internally) instead of urlparse —
418 # urlparse is vulnerable to parser-differential bypasses like
419 # ``http://127.0.0.1\@1.1.1.1`` (GHSA-g23j-2vwm-5c25).
420 #
421 # Per-scheme policy applied below:
422 # - http/https: full ``_is_private_ip`` check, honoring the
423 # operator ``allow_private_ips`` opt-in. RFC1918 / loopback
424 # are allowed through with the flag, but cloud-metadata and
425 # NAT64-wrapped metadata always block.
426 # - Apprise plugin schemes (discord, slack, signal, gotify,
427 # ntfy/ntfys, mattermost, rocketchat, matrix, teams, mailto,
428 # json, xml, form): private-IP reachability is intentionally
429 # allowed (these are typically self-hosted on a LAN), but the
430 # absolute cloud-metadata block still applies. Apprise
431 # translates these to HTTP requests against the URL host
432 # (e.g. ``signal://169.254.169.254/...`` → ``POST
433 # http://169.254.169.254/v2/send``), so without this guard
434 # the plugin schemes would bypass the IMDS protection that
435 # http/https has.
436 try:
437 u3 = parse_url(url)
438 except LocationParseError:
439 logger.warning(
440 "Blocked notification URL: urllib3 parser rejected it"
441 )
442 return False, "Invalid URL format (parser rejected)"
443 hostname = u3.host
444 # Authority must be ASCII printable (forward-defence vs urllib3
445 # ever loosening its IDN handling).
446 if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname): 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true
447 logger.warning(
448 "Blocked notification URL with non-ASCII / control bytes in host"
449 )
450 return False, "URL host contains disallowed characters"
451 hostname = NotificationURLValidator._normalize_host(hostname)
453 if scheme in ("http", "https"):
454 if hostname and NotificationURLValidator._is_private_ip(
455 hostname,
456 allow_private_ips=allow_private_ips,
457 _resolved_ips=resolved_ips,
458 allow_nat64=allow_nat64,
459 ):
460 logger.warning(
461 f"Blocked private/internal IP in notification URL: "
462 f"{hostname}"
463 )
464 return (
465 False,
466 f"{NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX} {hostname}",
467 )
468 else:
469 # Plugin-scheme IMDS guard. ``allow_private_ips=True`` leaves
470 # ALWAYS_BLOCKED_METADATA_IPS and NAT64-wrapped metadata as
471 # the only active blocks in ``_is_private_ip`` — exactly the
472 # set we want to enforce regardless of operator flags.
473 #
474 # resolved_ips is intentionally NOT forwarded here: the
475 # plugin-scheme guard must always perform its own resolution
476 # (it cannot rely on http(s)-centric pre-resolution).
477 if hostname and NotificationURLValidator._is_private_ip(
478 hostname,
479 allow_private_ips=True,
480 allow_nat64=allow_nat64,
481 ):
482 logger.warning(
483 f"Blocked cloud-metadata IP in notification URL: {hostname}"
484 )
485 return (
486 False,
487 f"Blocked cloud-metadata IP address: {hostname}",
488 )
490 # Passed all security checks
491 return True, None
493 @staticmethod
494 def _compute_hint(
495 hostname: str,
496 resolved_ips: Optional[List[ResolvedIP]],
497 ) -> bool:
498 """Return True iff the combined elevated policy
499 (LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true AND
500 LDR_SECURITY_ALLOW_NAT64=true) would unblock every resolved
501 IP. Probing both flags jointly — rather than each in
502 isolation — is required for mixed address lists where
503 neither flag alone unblocks every IP (e.g. 10.0.0.1 needs
504 the private-IPs flag, 64:ff9b::5db8:d822 needs the NAT64
505 flag).
506 """
507 hint = not NotificationURLValidator._host_blocks_at_level(
508 hostname,
509 allow_private_ips=True,
510 allow_nat64=True,
511 resolved_ips=resolved_ips,
512 )
513 if not hint:
514 logger.debug(
515 "hint suppressed: {} targets an always-blocked "
516 "category (metadata / 6to4 / Teredo / discard / "
517 "NAT64-wrapped metadata)",
518 hostname,
519 )
520 return hint
522 @staticmethod
523 def validate_service_url_with_hint(
524 url: str, allow_private_ips: bool = False
525 ) -> Tuple[bool, Optional[str], bool]:
526 """Validate URL and compute the admin-hint decision, using a
527 two-phase approach to ensure DNS is only queried for URLs that
528 pass ALL deterministic (non-network) checks.
530 Phase 1 — structural validation (no DNS):
531 Calls ``_validate_service_url_impl`` with ``resolved_ips=[]``, which
532 suppresses DNS for the http(s) private-IP check. IP literals
533 are still evaluated directly. Plugin-scheme IMDS guard resolves
534 internally. This means bad schemes, illegal characters, parse
535 errors, and malformed URLs are all rejected WITHOUT network
536 activity.
538 Phase 2 — DNS resolution + IP policy (http(s) hostnames only):
539 Only reached for http(s) URLs with non-literal hostnames that
540 passed Phase 1. Resolves DNS ONCE and shares the result across
541 the default-level rejection and the combined elevated-policy hint check,
542 closing the DNS-rebinding TOCTOU window. When DNS fails, the
543 failure is authoritative (``[]`` sentinel prevents retries).
545 For IP literals there is no DNS to rebind.
547 Args:
548 url: Service URL to validate
549 allow_private_ips: Operator-level flag. When True, the
550 private address categories (RFC1918 / CGNAT / loopback
551 / link-local / IPv6 ULA) are already permitted at this
552 level. ``hint_would_help`` may still be True for a
553 rejected NAT64-wrapped non-metadata destination,
554 because the NAT64 carve-out is governed by the separate
555 ``allow_nat64`` path rather than this flag.
557 Returns:
558 ``(is_valid, error_msg, hint_would_help)``:
560 * ``is_valid`` — True iff URL passes at the requested
561 ``allow_private_ips`` level.
562 * ``error_msg`` — None on success, the validator's reason
563 on rejection (identical to ``validate_service_url``).
564 * ``hint_would_help`` — True iff enabling the documented
565 escape hatches together
566 (``LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS`` for
567 RFC1918/CGNAT/loopback/link-local/IPv6 ULA, and
568 ``LDR_SECURITY_ALLOW_NAT64`` for NAT64-wrapped
569 non-metadata destinations) would unblock all resolved
570 addresses — covering cases needing one or both flags.
571 Always False when ``is_valid`` is True. May still be
572 True under ``allow_private_ips=True`` for a rejected
573 NAT64-wrapped non-metadata destination, because the
574 NAT64 carve-out is governed by the separate
575 ``allow_nat64`` path. Only meaningful when
576 ``error_msg`` starts with
577 ``PRIVATE_IP_REJECTION_PREFIX`` (i.e., the URL was
578 rejected for an http(s) private-IP target).
580 Caller contract:
581 ``notifications.service.NotificationService.test_service``
582 calls this to surface the validator's reason AND decide
583 whether to append the admin env-var hint in a single pass.
584 The previous design called ``validate_service_url`` two to
585 three times (once at the default level, then one or two
586 more via the former ``_admin_hint_would_help`` helper for
587 the private-IPs and NAT64 probes), opening a DNS-rebinding
588 window between the resolutions. The parametrized test
589 ``test_test_service_ip_rejection_matrix`` and the direct
590 contract test
591 ``test_validate_service_url_with_hint_contract`` in
592 tests/web/services/test_notification_coverage.py pin this
593 end-to-end.
594 """
595 if not url or not isinstance(url, str):
596 return False, "Service URL must be a non-empty string", False
598 url = url.strip()
600 if allow_private_ips and not url:
601 return (
602 False,
603 "Service URL must have a protocol (e.g., https://)",
604 False,
605 )
607 # ------------------------------------------------------------------
608 # Phase 1: structural validation (no DNS for http(s) hostnames).
609 #
610 # resolved_ips=[] suppresses DNS in the http(s) private-IP check
611 # (_host_blocks_at_level treats [] as "no blocked IPs"). IP
612 # literals are still checked directly. Plugin-scheme IMDS guard
613 # resolves internally (does not use resolved_ips). This means
614 # ALL deterministic rejections (bad scheme, illegal chars, parse
615 # errors, non-ASCII host, invalid structure) complete before any
616 # attacker-controlled DNS query.
617 # ------------------------------------------------------------------
618 is_valid, error_msg = (
619 NotificationURLValidator._validate_service_url_impl(
620 url,
621 allow_private_ips=allow_private_ips,
622 resolved_ips=[],
623 )
624 )
626 # If the URL was rejected in Phase 1, no DNS was needed.
627 if not is_valid:
628 if not error_msg or not error_msg.startswith(
629 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX
630 ):
631 logger.debug(
632 "hint suppressed: non-private-IP rejection ({})",
633 error_msg[:60] if error_msg else "<none>",
634 )
635 return False, error_msg, False
636 # Private-IP rejection — must be an IP literal (hostnames
637 # weren't resolved in Phase 1). Compute hint without DNS.
638 hostname = NotificationURLValidator._normalize_host(
639 NotificationURLValidator._extract_host(url)
640 )
641 if not hostname: 641 ↛ 642line 641 didn't jump to line 642 because the condition on line 641 was never true
642 return False, error_msg, False
643 return (
644 False,
645 error_msg,
646 NotificationURLValidator._compute_hint(hostname, None),
647 )
649 # URL passed Phase 1. For non-http(s) URLs or IP literals,
650 # validation is complete.
651 try:
652 parsed_scheme = urlparse(url).scheme.lower()
653 except Exception:
654 parsed_scheme = ""
656 hostname = NotificationURLValidator._normalize_host(
657 NotificationURLValidator._extract_host(url)
658 )
660 is_http_hostname = False
661 if hostname and parsed_scheme in ("http", "https"):
662 try:
663 ipaddress.ip_address(hostname)
664 except ValueError:
665 is_http_hostname = True
667 if not is_http_hostname:
668 # Non-http(s) or IP literal — fully validated in Phase 1.
669 return True, None, False
671 # ------------------------------------------------------------------
672 # Phase 2: resolve DNS ONCE for the http(s) hostname and run the
673 # real IP check. resolved_ips is authoritative for the rejection
674 # and hint check — no additional DNS lookups.
675 #
676 # [] = DNS attempted and failed (fail-open, no retry).
677 # [...] = resolved IPs shared across all policy decisions.
678 # ------------------------------------------------------------------
679 dns_result = NotificationURLValidator._resolve_hostname_ips(hostname)
680 resolved_ips: Optional[List[ResolvedIP]] = (
681 dns_result if dns_result is not None else []
682 )
684 is_valid, error_msg = (
685 NotificationURLValidator._validate_service_url_impl(
686 url,
687 allow_private_ips=allow_private_ips,
688 resolved_ips=resolved_ips,
689 )
690 )
691 if is_valid:
692 return True, None, False
693 if not error_msg or not error_msg.startswith( 693 ↛ 696line 693 didn't jump to line 696 because the condition on line 693 was never true
694 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX
695 ):
696 return False, error_msg, False
698 return (
699 False,
700 error_msg,
701 NotificationURLValidator._compute_hint(hostname, resolved_ips),
702 )
704 @staticmethod
705 def validate_service_url_strict(
706 url: str, allow_private_ips: bool = False
707 ) -> bool:
708 """
709 Strict validation that raises an exception on invalid URLs.
711 Args:
712 url: Service URL to validate
713 allow_private_ips: Whether to allow private IPs (default: False)
715 Returns:
716 True if valid
718 Raises:
719 NotificationURLValidationError: If URL fails security validation
720 """
721 is_valid, error_message = NotificationURLValidator.validate_service_url(
722 url, allow_private_ips
723 )
725 if not is_valid:
726 raise NotificationURLValidationError(
727 f"Notification service URL validation failed: {error_message}"
728 )
730 return True
732 @staticmethod
733 def validate_multiple_urls(
734 urls: str, allow_private_ips: bool = False, separator: str = ","
735 ) -> Tuple[bool, Optional[str]]:
736 """
737 Validate multiple comma-separated service URLs.
739 Args:
740 urls: Comma-separated service URLs
741 allow_private_ips: Whether to allow private IPs (default: False)
742 separator: URL separator (default: ",")
744 Returns:
745 Tuple of (all_valid, error_message)
746 - all_valid: True if all URLs pass validation
747 - error_message: None if all valid, first error if any invalid
748 """
749 if not urls or not isinstance(urls, str):
750 return False, "Service URLs must be a non-empty string"
752 # Split by separator and strip whitespace
753 url_list = [url.strip() for url in urls.split(separator) if url.strip()]
755 if not url_list:
756 return False, "No valid URLs found after parsing"
758 # Validate each URL
759 for url in url_list:
760 is_valid, error_message = (
761 NotificationURLValidator.validate_service_url(
762 url, allow_private_ips
763 )
764 )
766 if not is_valid:
767 # Return first error found
768 return (
769 False,
770 f"Invalid URL '{redact_url_for_log(url)}': {error_message}",
771 )
773 # All URLs passed validation
774 return True, None