Coverage for src/local_deep_research/security/notification_validator.py: 97%

1""" 

2Security validation for notification service URLs. 

3 

4This module provides validation for user-configured notification service URLs 

5to prevent Server-Side Request Forgery (SSRF) attacks and other security issues. 

6""" 

7 

8import ipaddress 

9import re 

10import socket 

11from concurrent.futures import ThreadPoolExecutor, TimeoutError 

12from typing import ClassVar, List, Optional, Tuple, Union 

13from urllib.parse import unquote, urlparse 

14 

15from loguru import logger 

16from urllib3.exceptions import LocationParseError 

17from urllib3.util import parse_url 

18 

19from .ip_ranges import PRIVATE_IP_RANGES as _PRIVATE_IP_RANGES 

20from .legacy_ipv4 import ( 

21 AMBIGUOUS_NUMERIC_IPV4_HOST_ERROR, 

22 ENCODED_NUMERIC_IPV4_HOST_ERROR, 

23 is_ambiguous_numeric_ipv4_host, 

24 is_percent_encoded_numeric_ipv4_host, 

25) 

26from .ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE, redact_url_for_log 

27 

28# Type alias for resolved IP addresses (avoids referencing the private 

29# ipaddress._BaseAddress API). 

30ResolvedIP = Union[ipaddress.IPv4Address, ipaddress.IPv6Address] 

31 

32_SCHEMELESS_URL_FRAGMENT_RE = re.compile( 

33 r"^(?:localhost|(?:\d{1,3}\.){3}\d{1,3}|" 

34 r"(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,})" 

35 r"(?::\d+)?(?:[/?#]|$)" 

36) 

37_URL_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]{0,31}://") 

38_URL_BOUNDARY_RE = re.compile(r"[,\s]+(?=[A-Za-z][A-Za-z0-9+.-]{0,31}://)") 

39_APPRISE_TELEGRAM_URL_RE = re.compile( 

40 r"^tgram://(?:bot)?[0-9]+(?::|%3A)[a-z0-9_-]+(?:/|\?|$)", 

41 re.IGNORECASE, 

42) 

43 

44 

45def parse_notification_url_list( 

46 urls: str, separator: str = "," 

47) -> tuple[list[str], Optional[str]]: 

48 """Partition notification URLs without dropping malformed fragments. 

49 

50 A boundary is a comma or whitespace sequence followed by a URL scheme. 

51 This preserves commas inside one Apprise service URL while keeping 

52 whitespace a hard separator. URL-like fragments without a scheme are 

53 returned separately so callers can fail closed instead of dispatching only 

54 the valid subset. 

55 

56 Returns ``(parsed_urls, invalid_fragment)``. Custom separators retain the 

57 historical split behavior. 

58 """ 

59 if separator != ",": 

60 return ( 

61 [entry.strip() for entry in urls.split(separator) if entry.strip()], 

62 None, 

63 ) 

64 

65 parsed_urls = [ 

66 entry.strip(" ,\t\r\n") 

67 for entry in _URL_BOUNDARY_RE.split(urls) 

68 if entry.strip(" ,\t\r\n") 

69 ] 

70 for entry in parsed_urls: 

71 if not _URL_SCHEME_RE.match(entry): 

72 return parsed_urls, entry 

73 for fragment in re.split(r"[,\s]+", entry)[1:]: 

74 if _SCHEMELESS_URL_FRAGMENT_RE.match(fragment): 

75 return parsed_urls, fragment 

76 return parsed_urls, None 

77 

78 

79class NotificationURLValidationError(ValueError): 

80 """Raised when a notification service URL fails security validation.""" 

81 

82 pass 

83 

84 

85class NotificationURLValidator: 

86 """Validates notification service URLs to prevent SSRF and other attacks.""" 

87 

88 # Dangerous protocols that should never be used for notifications 

89 BLOCKED_SCHEMES = ( 

90 "file", # Local file access 

91 "ftp", # FTP can be abused for SSRF 

92 "ftps", # Secure FTP can be abused for SSRF 

93 "data", # Data URIs can leak sensitive data 

94 "javascript", # XSS/code execution 

95 "vbscript", # XSS/code execution 

96 "about", # Browser internal 

97 "blob", # Browser internal 

98 ) 

99 

100 # Allowed protocols for notification services 

101 ALLOWED_SCHEMES = ( 

102 "http", # Webhook services 

103 "https", # Webhook services (preferred) 

104 "mailto", # Email notifications 

105 "discord", # Discord webhooks 

106 "slack", # Slack webhooks 

107 "tgram", # Apprise Telegram scheme 

108 "gotify", # Gotify notifications 

109 "pushover", # Pushover notifications 

110 "ntfy", # ntfy.sh notifications (http) 

111 "ntfys", # ntfy.sh notifications (https) 

112 "signal", # Signal via signal-api-rest container 

113 "matrix", # Matrix protocol 

114 "mattermost", # Mattermost webhooks 

115 "rocketchat", # Rocket.Chat webhooks 

116 "teams", # Microsoft Teams 

117 "json", # Generic JSON webhooks 

118 "xml", # Generic XML webhooks 

119 "form", # Form-encoded webhooks 

120 ) 

121 

122 # Apprise query parameters that can select a second outbound destination 

123 # or resource independently of the URL authority screened below. 

124 # 

125 # ``template`` is consumed by Discord/Slack/Workflows (including native 

126 # HTTPS webhook URLs that Apprise converts to those plugins) as an 

127 # unrestricted AppriseAttachment, which can read local files or fetch 

128 # remote URLs. ``redirect`` overrides the hardened AppriseAsset setting 

129 # used by the notification service and would re-enable redirect-based SSRF. 

130 BLOCKED_APPRISE_QUERY_KEYS = frozenset(("redirect", "template")) 

131 

132 # mailto-specific secondary egress/resource selectors. ``smtp`` replaces 

133 # the actual SMTP host; PGP key paths are unrestricted attachments; and 

134 # WKD derives HTTPS fetches from recipient-controlled domains. 

135 BLOCKED_MAILTO_QUERY_KEYS = frozenset( 

136 ("smtp", "pgppub", "pgpkey", "pgpprv", "wkd") 

137 ) 

138 

139 # Schemes whose authority identifies a network destination. Token-style 

140 # schemes such as discord and slack deliberately do not belong here. 

141 HOST_BEARING_PLUGIN_SCHEMES: ClassVar[frozenset[str]] = frozenset( 

142 { 

143 "signal", 

144 "gotify", 

145 "ntfy", 

146 "ntfys", 

147 "mattermost", 

148 "rocketchat", 

149 "matrix", 

150 "json", 

151 "xml", 

152 "form", 

153 "mailto", 

154 } 

155 ) 

156 ADDRESS_BEARING_SCHEMES: ClassVar[frozenset[str]] = ( 

157 frozenset({"http", "https"}) | HOST_BEARING_PLUGIN_SCHEMES 

158 ) 

159 

160 # Reuse shared private IP range definitions 

161 PRIVATE_IP_RANGES = _PRIVATE_IP_RANGES 

162 

163 # Prefix emitted by validate_service_url when an http(s) URL targets a 

164 # private/internal IP. Pinned as a class-level constant so the 

165 # validator, the hint logic, and the call site (service.py) share a 

166 # single source of truth. 

167 PRIVATE_IP_REJECTION_PREFIX = "Blocked private/internal IP address:" 

168 

169 @staticmethod 

170 def _ip_matches_blocked_range( 

171 ip, 

172 allow_private_ips: bool = False, 

173 allow_nat64: Optional[bool] = None, 

174 block_link_local: bool = False, 

175 ) -> bool: 

176 """Block-decision for a parsed IP, delegating to 

177 ``ssrf_validator.is_ip_blocked`` so the two validators share a 

178 single source of truth. 

179 

180 Honors: 

181 - ALWAYS_BLOCKED_METADATA_IPS (cloud metadata, absolute) 

182 - is_nat64_wrapped_metadata_ip (NAT64-wrapped IMDS, absolute) 

183 - security.allow_nat64 env carve-out for the two NAT64 prefixes 

184 (overridable via ``allow_nat64``: None reads env, an explicit 

185 bool answers the "would NAT64 unblock this?" hint probe) 

186 - allow_private_ips: when True, RFC1918 / CGNAT / loopback / 

187 link-local / IPv6 ULA are allowed BUT the two absolute checks 

188 above still fire. This closes the historical bypass where 

189 ``allow_private_ips=True`` skipped the host check entirely 

190 and let metadata IPs through the notification path. 

191 - block_link_local: when True, the whole link-local range 

192 (169.254.0.0/16, fe80::/10) stays blocked even under 

193 allow_private_ips=True. Used by the plugin-scheme IMDS guard so 

194 metadata reachable in link-local beyond the always-blocked 

195 literals cannot slip through the lenient partition. 

196 """ 

197 from .ssrf_validator import is_ip_blocked 

198 

199 return is_ip_blocked( 

200 str(ip), 

201 allow_private_ips=allow_private_ips, 

202 allow_nat64=allow_nat64, 

203 block_link_local=block_link_local, 

204 ) 

205 

206 @staticmethod 

207 def _resolve_hostname_ips( 

208 hostname: str, 

209 ) -> Optional[List[ResolvedIP]]: 

210 """Resolve hostname to a list of ipaddress objects, with a 5s 

211 timeout. Returns None on resolution failure or timeout. 

212 

213 Extracted from ``_is_private_ip`` so multiple policy decisions 

214 can share a single resolution pass — see 

215 ``validate_service_url_with_hint`` for the single-pass call 

216 site that closes a DNS-rebinding TOCTOU window between the 

217 default-level and elevated-level decisions. 

218 """ 

219 # NOTE: best-effort, validation-time check. Apprise re-resolves 

220 # the hostname when it actually sends the request (via 

221 # requests/urllib3), so a DNS-rebinding attacker could serve a 

222 # public IP here and a private/metadata IP at send time — the 

223 # classic resolve-vs-connect TOCTOU window. 

224 # 

225 # SEND-TIME GUARD (this window is now closed in code for the 

226 # delivery path): ``notifications.service`` forces Apprise into 

227 # synchronous, in-thread delivery (``AppriseAsset(async_mode=False)``), 

228 # disables HTTP redirect-following for the send (asset-level 

229 # ``http_redirects=False`` re-forced per plugin, so a user-supplied 

230 # ``?redirect=yes`` cannot re-enable it), and wraps every ``notify()`` 

231 # in ``dns_pinning.pinned_notification_send`` (see 

232 # ``security.dns_pinning``). Together, for the duration of the send, 

233 # that (a) closes the redirect attack class outright — a webhook can 

234 # no longer 30x-redirect the send to a private/loopback host or to an 

235 # arbitrary public host for data exfiltration, because the send is 

236 # simply not redirected; (b) pins every raw-webhook host in the batch 

237 # to the address validated here, so the ORIGINAL host's own DNS 

238 # rebind cannot steer the connection to a different address; and 

239 # (c) activates a thread-local block-private mode so any UNPINNED 

240 # lookup (a plugin scheme's endpoint) that resolves to a blocked 

241 # address is refused at the ``getaddrinfo`` layer — on a 

242 # timeout-bounded resolution — before a socket is opened. The block 

243 # runs on the very resolution the client connects to, so there is no 

244 # residual race. Per-scheme policy mirrors this validator: http/https 

245 # block private+metadata (unless the operator opts in), plugin / 

246 # raw-webhook schemes block only cloud-metadata (self-hosted LAN 

247 # targets keep working); cloud-metadata is ALWAYS blocked regardless 

248 # of scheme or operator flag. Because the pin/block are thread-local, 

249 # they never disturb a legitimate private request on another thread. 

250 # 

251 # DEFENSE IN DEPTH: the whole outbound-notification path is still 

252 # gated behind an env-only master switch 

253 # (LDR_NOTIFICATIONS_ALLOW_OUTBOUND, default off); enabling it is the 

254 # operator's explicit decision. See SECURITY.md "Notification Webhook 

255 # SSRF". Operators can further avoid raw webhooks by preferring 

256 # plugin schemes (discord://, slack://, ntfy://, ntfys://, gotify://, 

257 # tgram://, mattermost://, etc.) that hardcode their endpoints. 

258 # 

259 # Plugin schemes are not categorically exempt from authority 

260 # validation: some have hardcoded destinations (Discord, Slack, 

261 # Teams Workflows), some send against the URL authority (Signal, 

262 # Gotify, generic webhooks), and ntfy/Matrix are mode-dependent. 

263 # 

264 # concurrent.futures for thread-safe timeout instead of 

265 # socket.setdefaulttimeout() which is process-global. 

266 try: 

267 executor = ThreadPoolExecutor(max_workers=1) 

268 try: 

269 future = executor.submit( 

270 socket.getaddrinfo, 

271 hostname, 

272 None, 

273 socket.AF_UNSPEC, 

274 socket.SOCK_STREAM, 

275 ) 

276 resolved_ips = future.result(timeout=5) 

277 finally: 

278 executor.shutdown(wait=False, cancel_futures=True) 

279 return [ 

280 ipaddress.ip_address(sockaddr[0]) 

281 for _family, _, _, _, sockaddr in resolved_ips 

282 ] 

283 except (socket.gaierror, OSError, TimeoutError, ValueError): 

284 logger.warning( 

285 "DNS resolution failed for hostname {} — " 

286 "allowing request (unable to determine if private)", 

287 hostname, 

288 ) 

289 return None 

290 

291 @staticmethod 

292 def _host_blocks_at_level( 

293 hostname: str, 

294 allow_private_ips: bool = False, 

295 allow_nat64: Optional[bool] = None, 

296 resolved_ips: Optional[List[ResolvedIP]] = None, 

297 block_link_local: bool = False, 

298 ) -> bool: 

299 """Decide whether ``hostname`` is blocked at the given 

300 ``allow_private_ips`` / ``allow_nat64`` level, optionally reusing 

301 a prior DNS resolution. 

302 

303 Pre-resolved IPs can be passed via ``resolved_ips`` to share a 

304 single resolution across multiple policy decisions, closing 

305 DNS-rebinding TOCTOU windows between calls. If ``resolved_ips`` 

306 is None and ``hostname`` is not an IP literal, DNS is resolved 

307 via ``_resolve_hostname_ips``. 

308 

309 ``block_link_local`` forces the whole link-local range blocked even 

310 under ``allow_private_ips=True`` (plugin-scheme IMDS guard). 

311 """ 

312 # Localhost-string shortcuts only apply when the operator hasn't 

313 # opted into private-IP reachability. With allow_private_ips=True 

314 # we let the IP path (DNS-resolved or literal) make the decision 

315 # so metadata-IP literals like "169.254.169.254" still block. 

316 if not allow_private_ips and hostname.lower() in ( 

317 "localhost", 

318 "127.0.0.1", 

319 "::1", 

320 "0.0.0.0", 

321 "::", 

322 ): 

323 return True 

324 

325 # Try to parse as IP address first. 

326 try: 

327 ip = ipaddress.ip_address(hostname) 

328 return NotificationURLValidator._ip_matches_blocked_range( 

329 ip, 

330 allow_private_ips=allow_private_ips, 

331 allow_nat64=allow_nat64, 

332 block_link_local=block_link_local, 

333 ) 

334 except ValueError: 

335 pass 

336 

337 # Hostname — use pre-resolved IPs if provided, otherwise resolve. 

338 if resolved_ips is None: 

339 resolved_ips = NotificationURLValidator._resolve_hostname_ips( 

340 hostname 

341 ) 

342 if resolved_ips is None: 

343 # DNS failed — allow (matches the historical behaviour). 

344 return False 

345 for ip in resolved_ips: 

346 if NotificationURLValidator._ip_matches_blocked_range( 

347 ip, 

348 allow_private_ips=allow_private_ips, 

349 allow_nat64=allow_nat64, 

350 block_link_local=block_link_local, 

351 ): 

352 return True 

353 return False 

354 

355 @staticmethod 

356 def _extract_host(url: str) -> Optional[str]: 

357 """Extract the raw hostname from ``url`` using urllib3's 

358 ``parse_url`` (the same parser requests uses internally). 

359 Returns None if urllib3 rejects the URL. 

360 """ 

361 try: 

362 u3 = parse_url(url) 

363 except LocationParseError: 

364 return None 

365 return u3.host 

366 

367 @staticmethod 

368 def _normalize_host(hostname: Optional[str]) -> str: 

369 """Normalize a raw hostname from ``parse_url``: strip IPv6 

370 bracket notation and trailing dots. Shared by 

371 ``validate_service_url`` and ``validate_service_url_with_hint`` 

372 so the two call sites cannot drift. 

373 """ 

374 if hostname and hostname.startswith("[") and hostname.endswith("]"): 

375 hostname = hostname[1:-1] 

376 if hostname: 

377 hostname = hostname.rstrip(".") 

378 return hostname or "" 

379 

380 @staticmethod 

381 def _apprise_query_keys(url: str) -> Tuple[List[str], bool]: 

382 """Return canonical query keys using Apprise 1.12 semantics. 

383 

384 Apprise splits pairs on both ``&`` and ``;``, preserves a leading 

385 ``+`` key namespace marker, converts later ``+`` characters to spaces, 

386 percent-decodes once, then lowercases and strips the key. Its URL 

387 parser also treats everything after the first ``?`` as query data, 

388 including text that ``urllib.parse`` would classify as a fragment. 

389 

390 The boolean reports malformed percent escapes in a key. Values are 

391 intentionally neither decoded nor returned: they can contain secrets, 

392 file paths, and alternate destinations and are irrelevant to policy. 

393 """ 

394 query_at = url.find("?") 

395 if query_at < 0: 

396 return [], False 

397 

398 raw_query = url[query_at + 1 :] 

399 keys = [] 

400 hex_digits = "0123456789abcdefABCDEF" 

401 

402 for amp_pair in raw_query.split("&"): 

403 for pair in amp_pair.split(";"): 

404 raw_key = pair.split("=", 1)[0] 

405 

406 index = 0 

407 while index < len(raw_key): 

408 if raw_key[index] != "%": 

409 index += 1 

410 continue 

411 if ( 

412 index + 2 >= len(raw_key) 

413 or raw_key[index + 1] not in hex_digits 

414 or raw_key[index + 2] not in hex_digits 

415 ): 

416 return keys, True 

417 index += 3 

418 

419 apprise_key = ( 

420 raw_key[:1] + raw_key[1:].replace("+", " ") 

421 if raw_key 

422 else "" 

423 ) 

424 keys.append(unquote(apprise_key).lower().strip()) 

425 

426 return keys, False 

427 

428 @staticmethod 

429 def _validate_apprise_query_policy(url: str, scheme: str) -> Optional[str]: 

430 """Reject Apprise query options that bypass destination validation.""" 

431 keys, malformed = NotificationURLValidator._apprise_query_keys(url) 

432 if malformed: 

433 logger.warning( 

434 "Blocked notification URL with malformed percent-encoding " 

435 "in a query parameter name" 

436 ) 

437 return "Malformed percent-encoding in notification parameter name" 

438 

439 blocked_keys = NotificationURLValidator.BLOCKED_APPRISE_QUERY_KEYS 

440 if scheme == "mailto": 

441 blocked_keys = ( 

442 blocked_keys 

443 | NotificationURLValidator.BLOCKED_MAILTO_QUERY_KEYS 

444 ) 

445 

446 for key in keys: 

447 if key in blocked_keys: 

448 # Never log the value: it may contain a token, local path, or 

449 # alternate destination selected by an attacker. 

450 logger.warning( 

451 "Blocked unsafe notification parameter {} for {} URL", 

452 key, 

453 scheme, 

454 ) 

455 return f"Blocked unsafe notification parameter: {key}" 

456 

457 return None 

458 

459 @staticmethod 

460 def _is_private_ip( 

461 hostname: str, 

462 allow_private_ips: bool = False, 

463 allow_nat64: Optional[bool] = None, 

464 _resolved_ips: Optional[List[ResolvedIP]] = None, 

465 block_link_local: bool = False, 

466 ) -> bool: 

467 """ 

468 Check if hostname resolves to a private IP address. 

469 

470 Args: 

471 hostname: Hostname to check 

472 allow_private_ips: When True, RFC1918 / CGNAT / loopback / 

473 link-local / IPv6 ULA are NOT considered private. Cloud 

474 metadata IPs and NAT64-wrapped metadata IPs are blocked 

475 regardless — the operator opt-in cannot license IMDS 

476 exposure. 

477 allow_nat64: Override for the ``security.allow_nat64`` carve-out. 

478 None (default) reads the env setting; an explicit bool 

479 answers the hint probe. 

480 block_link_local: When True, the whole link-local range stays 

481 blocked even under allow_private_ips=True. Used by the 

482 plugin-scheme IMDS guard (metadata lives in link-local 

483 beyond the always-blocked literals). 

484 

485 Returns: 

486 True if hostname is a private IP or localhost (subject to 

487 allow_private_ips), or wraps a metadata IP unconditionally 

488 """ 

489 return NotificationURLValidator._host_blocks_at_level( 

490 hostname, 

491 allow_private_ips=allow_private_ips, 

492 allow_nat64=allow_nat64, 

493 resolved_ips=_resolved_ips, 

494 block_link_local=block_link_local, 

495 ) 

496 

497 @staticmethod 

498 def validate_service_url( 

499 url: str, 

500 allow_private_ips: bool = False, 

501 allow_nat64: Optional[bool] = None, 

502 ) -> Tuple[bool, Optional[str]]: 

503 """Validate a notification service URL for security issues. 

504 

505 Thin public wrapper around ``_validate_service_url_impl``. 

506 Always performs its own DNS resolution (no caller-supplied 

507 resolution state). 

508 

509 Args: 

510 url: Service URL to validate (e.g., "discord://webhook_id/token") 

511 allow_private_ips: Whether to allow private IPs (default: False) 

512 Set to True for development/testing environments 

513 allow_nat64: Override for the ``security.allow_nat64`` 

514 carve-out. None (default) reads the env setting; 

515 an explicit bool overrides it. 

516 

517 Returns: 

518 Tuple of (is_valid, error_message) 

519 """ 

520 return NotificationURLValidator._validate_service_url_impl( 

521 url, 

522 allow_private_ips=allow_private_ips, 

523 allow_nat64=allow_nat64, 

524 ) 

525 

526 @staticmethod 

527 def _validate_service_url_impl( 

528 url: str, 

529 allow_private_ips: bool = False, 

530 allow_nat64: Optional[bool] = None, 

531 resolved_ips: Optional[List[ResolvedIP]] = None, 

532 ) -> Tuple[bool, Optional[str]]: 

533 """Internal validation with optional pre-resolved DNS state. 

534 

535 ``resolved_ips`` is trusted internal state used by 

536 ``validate_service_url_with_hint`` to share a single DNS 

537 resolution across multiple policy decisions. Must NEVER be 

538 exposed on the public API — callers could pass ``[]`` to 

539 bypass hostname resolution and private-IP checks. 

540 

541 Returns: 

542 Tuple of (is_valid, error_message) 

543 - is_valid: True if URL passes security checks 

544 - error_message: None if valid, error description if invalid 

545 

546 Examples: 

547 >>> validate_service_url("discord://webhook_id/token") 

548 (True, None) 

549 

550 >>> validate_service_url("file:///etc/passwd") 

551 (False, "Blocked unsafe protocol: file") 

552 

553 >>> validate_service_url("http://localhost:5000/webhook") 

554 (False, "Blocked private/internal IP address: localhost") 

555 

556 Caller contract: 

557 ``notifications.service.NotificationService.test_service`` 

558 calls ``validate_service_url_with_hint`` (single-pass sibling 

559 below) to obtain the validation result AND, when the URL is 

560 rejected, a ``hint_would_help`` flag indicating whether the 

561 rejection targets a recoverable destination — one that 

562 ``LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true`` (for RFC1918 / 

563 CGNAT / loopback / link-local / IPv6 ULA, plus NAT64-wrapped 

564 non-metadata destinations via the NAT64 carve-out) would 

565 actually unblock. If ``hint_would_help`` is False the URL 

566 targets an always-blocked category (cloud-metadata IPs, 6to4, 

567 Teredo, discard prefix, IPv4-mapped IPv6 of metadata, 

568 NAT64-wrapped metadata) and the hint is suppressed because 

569 naming the env var would mislead. The parametrized 

570 integration test ``test_test_service_ip_rejection_matrix`` 

571 in tests/web/services/test_notification_coverage.py locks 

572 this contract end-to-end across every IP category — if the 

573 wording here changes, that test fails and the call site 

574 needs updating. 

575 """ 

576 if not url or not isinstance(url, str): 

577 return False, "Service URL must be a non-empty string" 

578 

579 # Strip whitespace (must run before the RFC-illegal char check 

580 # so legitimate URLs with surrounding whitespace are not rejected). 

581 url = url.strip() 

582 

583 # Reject URLs containing characters that drive parser-differential 

584 # SSRF bypasses (backslash, whitespace, control bytes) — see 

585 # GHSA-g23j-2vwm-5c25. The URL is omitted from the log line because 

586 # userinfo (RFC 3986 §3.2.1) may contain credentials and rejected 

587 # URLs are by definition adversarial-shaped. 

588 if RFC_FORBIDDEN_URL_CHARS_RE.search(url): 

589 logger.warning( 

590 "Blocked notification URL containing RFC-illegal characters" 

591 ) 

592 return ( 

593 False, 

594 "URL contains characters that are not allowed (whitespace, backslash, or control bytes)", 

595 ) 

596 

597 # Parse URL 

598 try: 

599 parsed = urlparse(url) 

600 except Exception: 

601 # Never echo the parser exception back to the caller: this error 

602 # string is surfaced to the user by the test-URL endpoint, and the 

603 # exception text can carry parser internals / stack-trace fragments 

604 # (CWE-209, py/stack-trace-exposure). Log at WARNING without a 

605 # traceback to match the sibling LocationParseError handler below 

606 # — a malformed URL is benign user input, not a server fault, so an 

607 # ERROR-level stack trace would only add noise. 

608 logger.warning("Failed to parse service URL") 

609 return False, "Invalid URL format" 

610 

611 # Check for scheme 

612 if not parsed.scheme: 

613 return False, "Service URL must have a protocol (e.g., https://)" 

614 

615 scheme = parsed.scheme.lower() 

616 

617 # Check for blocked schemes 

618 if scheme in NotificationURLValidator.BLOCKED_SCHEMES: 

619 logger.warning( 

620 f"Blocked unsafe notification protocol: {scheme} in URL: {redact_url_for_log(url)}" 

621 ) 

622 return False, f"Blocked unsafe protocol: {scheme}" 

623 

624 # Check for allowed schemes 

625 if scheme not in NotificationURLValidator.ALLOWED_SCHEMES: 

626 if scheme == "telegram": 

627 # Dedicated migration error: 'telegram://' was an Apprise 

628 # scheme removed upstream; point users at the canonical 

629 # 'tgram://' form instead of the generic allowed-list. 

630 return ( 

631 False, 

632 "The 'telegram://' scheme is no longer supported by " 

633 "Apprise (removed upstream); replace it with Apprise's " 

634 "canonical 'tgram://<bot_token>/<chat_id>' form " 

635 "(e.g. tgram://123456789:AAexample_token/123456789)", 

636 ) 

637 logger.warning( 

638 f"Unknown notification protocol: {scheme} in URL: {redact_url_for_log(url)}" 

639 ) 

640 return ( 

641 False, 

642 f"Unsupported protocol: {scheme}. " 

643 f"Allowed: {', '.join(NotificationURLValidator.ALLOWED_SCHEMES[:5])}...", 

644 ) 

645 

646 # Apprise supports query parameters that can select resources or 

647 # outbound destinations independently of the authority we validate 

648 # below. Enforce this deterministic policy before parsing the host or 

649 # performing any attacker-controlled DNS lookup. 

650 query_error = NotificationURLValidator._validate_apprise_query_policy( 

651 url, scheme 

652 ) 

653 if query_error: 

654 return False, query_error 

655 

656 # Reject authorities with more than one "@" (fail closed). urllib3's 

657 # parse_url (which the SSRF host-check + the DNS pin validate against) 

658 # and Apprise's own parse_url disagree on which host a multi-"@" 

659 # authority denotes: for 

660 # ``json://token@169.254.169.254@decoy-public.example.com/path`` 

661 # urllib3 sees ``decoy-public.example.com`` (validated/pinned) while 

662 # Apprise connects to ``169.254.169.254``. This is the same 

663 # parser-differential class as GHSA-g23j-2vwm-5c25; there is no 

664 # legitimate reason for a second "@" in a notification authority, so 

665 # reject it before the host is ever extracted. ``netloc`` is the 

666 # authority only (path/query/fragment excluded), so an "@" in a query 

667 # string is not counted. A single "@" (RFC 3986 userinfo, e.g. 

668 # ``mailto://user:pass@host``) is still allowed. 

669 if parsed.netloc.count("@") > 1: 

670 logger.warning( 

671 "Blocked notification URL: authority contains multiple '@' " 

672 "(parser-differential SSRF risk)" 

673 ) 

674 return ( 

675 False, 

676 "URL authority contains multiple '@' characters, " 

677 "which is not allowed", 

678 ) 

679 

680 # Reject a scheme://-form URL whose authority is EMPTY while content 

681 # (a host smuggled into the path) follows — another parser-differential 

682 # SSRF (defense-in-depth; the send-time block window backstops it). 

683 # urllib3/requests parse an empty ``//`` authority as "no host", so the 

684 # per-scheme IP checks below are skipped and the URL is accepted, but 

685 # Apprise falls back to treating the first path segment as the host and 

686 # dials it: ``json:///169.254.169.254/path`` reaches 

687 # ``169.254.169.254``. No legitimate notification URL has an empty 

688 # ``//`` authority, so reject it before host extraction. This fires 

689 # ONLY when the ``//`` authority marker is present — a scheme that 

690 # legitimately has no ``//`` authority (an RFC ``mailto:user@host`` 

691 # form) never reaches here, so it is not broken. 

692 after_scheme = url[len(parsed.scheme) + 1 :] 

693 if after_scheme.startswith("//") and not parsed.netloc: 

694 remainder = after_scheme[2:] # everything past the '//' 

695 if remainder: 

696 logger.warning( 

697 "Blocked notification URL with empty authority but " 

698 "in-path host (parser-differential SSRF risk)" 

699 ) 

700 return ( 

701 False, 

702 "URL host (authority) is empty; a host in the path after " 

703 "'scheme:///' is not allowed", 

704 ) 

705 

706 # ``tgram://`` is checked here, BEFORE host extraction, because its 

707 # authority is a CREDENTIAL (``<bot_id>:<token>``), not a network 

708 # destination: Apprise's Telegram plugin always dials the hardcoded 

709 # api.telegram.org endpoint. urllib3 cannot parse that authority at 

710 # all (it reads ``:<token>`` as a port and raises LocationParseError), 

711 # so the host-based checks below can never run for a valid tgram URL. 

712 # 

713 # ORDERING NOTE for future merges: this early return deliberately 

714 # precedes the numeric-IPv4 screening below, and that is safe ONLY 

715 # because ``tgram`` is not in ADDRESS_BEARING_SCHEMES. The regex below 

716 # is a strict allowlist that already rejects every legacy IPv4 host 

717 # form with a non-decimal component (``0x7f000001``, ``0177.0.0.1``, 

718 # ``127.1`` all fail ``[0-9]+``). The one remaining overlap — a bare 

719 # decimal integer such as ``2130706433`` — is exactly the shape of a 

720 # legitimate Telegram bot id, so it must NOT be rejected as an 

721 # ambiguous numeric IPv4 host. Any check that IS address-bearing must 

722 # be placed above this block, not below it. 

723 if scheme == "tgram": 

724 if not _APPRISE_TELEGRAM_URL_RE.match(url): 

725 return ( 

726 False, 

727 "Invalid Telegram service URL; expected " 

728 "tgram://<bot_id>:<token>/<chat_id>", 

729 ) 

730 return True, None 

731 

732 # Extract the host for address-bearing schemes. We use urllib3 (the 

733 # parser ``requests`` uses internally) instead of urlparse — 

734 # urlparse is vulnerable to parser-differential bypasses like 

735 # ``http://127.0.0.1\@1.1.1.1`` (GHSA-g23j-2vwm-5c25). 

736 # 

737 # Per-scheme policy applied below: 

738 # - http/https: full ``_is_private_ip`` check, honoring the 

739 # operator ``allow_private_ips`` opt-in. RFC1918 / loopback 

740 # are allowed through with the flag, but cloud-metadata and 

741 # NAT64-wrapped metadata always block. 

742 # - Apprise plugin schemes: private-IP authority values are 

743 # intentionally allowed for self-hosted modes, but the absolute 

744 # cloud-metadata block still applies. Some modes send against the 

745 # authority (for example Signal and ntfy private); fixed-destination 

746 # modes such as Discord, Slack, ntfy cloud, and Matrix t2bot treat it 

747 # as a token/topic instead. Mail can use its authority or a fixed 

748 # provider mapping. Screening every authority uniformly prevents 

749 # host-bearing modes from bypassing the IMDS protection without 

750 # incorrectly claiming every authority is the network destination. 

751 # Secondary resource/destination parameters were rejected above. 

752 try: 

753 u3 = parse_url(url) 

754 except LocationParseError: 

755 logger.warning( 

756 "Blocked notification URL: urllib3 parser rejected it" 

757 ) 

758 return False, "Invalid URL format (parser rejected)" 

759 hostname = u3.host 

760 # Authority must be ASCII printable (forward-defence vs urllib3 

761 # ever loosening its IDN handling). 

762 if hostname and any(ord(c) < 0x20 or ord(c) > 0x7E for c in hostname): 762 ↛ 763line 762 didn't jump to line 763 because the condition on line 762 was never true

763 logger.warning( 

764 "Blocked notification URL with non-ASCII / control bytes in host" 

765 ) 

766 return False, "URL host contains disallowed characters" 

767 hostname = NotificationURLValidator._normalize_host(hostname) 

768 

769 # A populated ``//`` authority that urllib3 parses to an EMPTY host 

770 # (e.g. ``json://169.254.169.254:80@/`` — an IP smuggled into the 

771 # userinfo with nothing after the ``@``) is the sibling of the 

772 # multi-``@`` / empty-``//``-authority cases above: urllib3 returns no 

773 # host, so the per-scheme ``_is_private_ip`` checks below (guarded on 

774 # ``if hostname``) are skipped and the URL is accepted, yet the 

775 # authority still carries a target that another parser could dial. No 

776 # legitimate notification URL has a non-empty ``//`` authority with an 

777 # empty host, so reject it fail-closed before the scheme checks. 

778 if after_scheme.startswith("//") and parsed.netloc and not hostname: 

779 logger.warning( 

780 "Blocked notification URL: non-empty authority with empty host " 

781 "(parser-differential SSRF risk)" 

782 ) 

783 return ( 

784 False, 

785 "URL authority has no host; a host smuggled into the userinfo " 

786 "is not allowed", 

787 ) 

788 

789 if ( 

790 hostname 

791 and scheme in NotificationURLValidator.ADDRESS_BEARING_SCHEMES 

792 ): 

793 if is_percent_encoded_numeric_ipv4_host(hostname): 

794 logger.warning( 

795 "Blocked notification URL with encoded numeric IPv4 host" 

796 ) 

797 return False, ENCODED_NUMERIC_IPV4_HOST_ERROR 

798 if is_ambiguous_numeric_ipv4_host(hostname): 

799 logger.warning( 

800 "Blocked notification URL with ambiguous numeric IPv4 host" 

801 ) 

802 return False, AMBIGUOUS_NUMERIC_IPV4_HOST_ERROR 

803 

804 if scheme in ("http", "https"): 

805 if hostname and NotificationURLValidator._is_private_ip( 

806 hostname, 

807 allow_private_ips=allow_private_ips, 

808 _resolved_ips=resolved_ips, 

809 allow_nat64=allow_nat64, 

810 ): 

811 logger.warning( 

812 f"Blocked private/internal IP in notification URL: " 

813 f"{hostname}" 

814 ) 

815 return ( 

816 False, 

817 f"{NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX} {hostname}", 

818 ) 

819 elif scheme in NotificationURLValidator.HOST_BEARING_PLUGIN_SCHEMES: 

820 # Plugin-scheme IMDS guard. ``allow_private_ips=True`` leaves 

821 # ALWAYS_BLOCKED_METADATA_IPS and NAT64-wrapped metadata as 

822 # active blocks in ``_is_private_ip``; ``block_link_local=True`` 

823 # additionally keeps the whole link-local range blocked — cloud 

824 # metadata lives in link-local beyond the always-blocked literals 

825 # (e.g. Scaleway 169.254.42.42) and no legitimate self-hosted 

826 # notifier does. Exactly the set we want to enforce regardless of 

827 # operator flags (RFC1918 / loopback / non-link-local ULA still 

828 # allowed). 

829 # 

830 # resolved_ips is intentionally NOT forwarded here: the 

831 # plugin-scheme guard must always perform its own resolution 

832 # (it cannot rely on http(s)-centric pre-resolution). 

833 if hostname and NotificationURLValidator._is_private_ip( 

834 hostname, 

835 allow_private_ips=True, 

836 allow_nat64=allow_nat64, 

837 block_link_local=True, 

838 ): 

839 logger.warning( 

840 "Blocked cloud-metadata / link-local IP in notification " 

841 f"URL: {hostname}" 

842 ) 

843 return ( 

844 False, 

845 f"Blocked cloud-metadata / link-local IP address: {hostname}", 

846 ) 

847 

848 # Passed all security checks 

849 return True, None 

850 

851 @staticmethod 

852 def _compute_hint( 

853 hostname: str, 

854 resolved_ips: Optional[List[ResolvedIP]], 

855 ) -> bool: 

856 """Return True iff the combined elevated policy 

857 (LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true AND 

858 LDR_SECURITY_ALLOW_NAT64=true) would unblock every resolved 

859 IP. Probing both flags jointly — rather than each in 

860 isolation — is required for mixed address lists where 

861 neither flag alone unblocks every IP (e.g. 10.0.0.1 needs 

862 the private-IPs flag, 64:ff9b::5db8:d822 needs the NAT64 

863 flag). 

864 """ 

865 hint = not NotificationURLValidator._host_blocks_at_level( 

866 hostname, 

867 allow_private_ips=True, 

868 allow_nat64=True, 

869 resolved_ips=resolved_ips, 

870 ) 

871 if not hint: 

872 logger.debug( 

873 "hint suppressed: {} targets an always-blocked " 

874 "category (metadata / 6to4 / Teredo / discard / " 

875 "NAT64-wrapped metadata)", 

876 hostname, 

877 ) 

878 return hint 

879 

880 @staticmethod 

881 def validate_service_url_with_hint( 

882 url: str, allow_private_ips: bool = False 

883 ) -> Tuple[bool, Optional[str], bool]: 

884 """Validate URL and compute the admin-hint decision, using a 

885 two-phase approach to ensure DNS is only queried for URLs that 

886 pass ALL deterministic (non-network) checks. 

887 

888 Phase 1 — structural validation (no DNS): 

889 Calls ``_validate_service_url_impl`` with ``resolved_ips=[]``, which 

890 suppresses DNS for the http(s) private-IP check. IP literals 

891 are still evaluated directly. Plugin-scheme IMDS guard resolves 

892 internally. This means bad schemes, illegal characters, parse 

893 errors, and malformed URLs are all rejected WITHOUT network 

894 activity. 

895 

896 Phase 2 — DNS resolution + IP policy (http(s) hostnames only): 

897 Only reached for http(s) URLs with non-literal hostnames that 

898 passed Phase 1. Resolves DNS ONCE and shares the result across 

899 the default-level rejection and the combined elevated-policy hint check, 

900 closing the DNS-rebinding TOCTOU window. When DNS fails, the 

901 failure is authoritative (``[]`` sentinel prevents retries). 

902 

903 For IP literals there is no DNS to rebind. 

904 

905 Args: 

906 url: Service URL to validate 

907 allow_private_ips: Operator-level flag. When True, the 

908 private address categories (RFC1918 / CGNAT / loopback 

909 / link-local / IPv6 ULA) are already permitted at this 

910 level. ``hint_would_help`` may still be True for a 

911 rejected NAT64-wrapped non-metadata destination, 

912 because the NAT64 carve-out is governed by the separate 

913 ``allow_nat64`` path rather than this flag. 

914 

915 Returns: 

916 ``(is_valid, error_msg, hint_would_help)``: 

917 

918 * ``is_valid`` — True iff URL passes at the requested 

919 ``allow_private_ips`` level. 

920 * ``error_msg`` — None on success, the validator's reason 

921 on rejection (identical to ``validate_service_url``). 

922 * ``hint_would_help`` — True iff enabling the documented 

923 escape hatches together 

924 (``LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS`` for 

925 RFC1918/CGNAT/loopback/link-local/IPv6 ULA, and 

926 ``LDR_SECURITY_ALLOW_NAT64`` for NAT64-wrapped 

927 non-metadata destinations) would unblock all resolved 

928 addresses — covering cases needing one or both flags. 

929 Always False when ``is_valid`` is True. May still be 

930 True under ``allow_private_ips=True`` for a rejected 

931 NAT64-wrapped non-metadata destination, because the 

932 NAT64 carve-out is governed by the separate 

933 ``allow_nat64`` path. Only meaningful when 

934 ``error_msg`` starts with 

935 ``PRIVATE_IP_REJECTION_PREFIX`` (i.e., the URL was 

936 rejected for an http(s) private-IP target). 

937 

938 Caller contract: 

939 ``notifications.service.NotificationService.test_service`` 

940 calls this to surface the validator's reason AND decide 

941 whether to append the admin env-var hint in a single pass. 

942 The previous design called ``validate_service_url`` two to 

943 three times (once at the default level, then one or two 

944 more via the former ``_admin_hint_would_help`` helper for 

945 the private-IPs and NAT64 probes), opening a DNS-rebinding 

946 window between the resolutions. The parametrized test 

947 ``test_test_service_ip_rejection_matrix`` and the direct 

948 contract test 

949 ``test_validate_service_url_with_hint_contract`` in 

950 tests/web/services/test_notification_coverage.py pin this 

951 end-to-end. 

952 """ 

953 if not url or not isinstance(url, str): 

954 return False, "Service URL must be a non-empty string", False 

955 

956 url = url.strip() 

957 

958 if allow_private_ips and not url: 

959 return ( 

960 False, 

961 "Service URL must have a protocol (e.g., https://)", 

962 False, 

963 ) 

964 

965 # ------------------------------------------------------------------ 

966 # Phase 1: structural validation (no DNS for http(s) hostnames). 

967 # 

968 # resolved_ips=[] suppresses DNS in the http(s) private-IP check 

969 # (_host_blocks_at_level treats [] as "no blocked IPs"). IP 

970 # literals are still checked directly. Plugin-scheme IMDS guard 

971 # resolves internally (does not use resolved_ips). This means 

972 # ALL deterministic rejections (bad scheme, illegal chars, parse 

973 # errors, non-ASCII host, invalid structure) complete before any 

974 # attacker-controlled DNS query. 

975 # ------------------------------------------------------------------ 

976 is_valid, error_msg = ( 

977 NotificationURLValidator._validate_service_url_impl( 

978 url, 

979 allow_private_ips=allow_private_ips, 

980 resolved_ips=[], 

981 ) 

982 ) 

983 

984 # If the URL was rejected in Phase 1, no DNS was needed. 

985 if not is_valid: 

986 if error_msg in ( 

987 AMBIGUOUS_NUMERIC_IPV4_HOST_ERROR, 

988 ENCODED_NUMERIC_IPV4_HOST_ERROR, 

989 ): 

990 return False, error_msg, False 

991 if not error_msg or not error_msg.startswith( 

992 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX 

993 ): 

994 logger.debug( 

995 "hint suppressed: non-private-IP rejection ({})", 

996 error_msg[:60] if error_msg else "<none>", 

997 ) 

998 return False, error_msg, False 

999 # Private-IP rejection — must be an IP literal (hostnames 

1000 # weren't resolved in Phase 1). Compute hint without DNS. 

1001 hostname = NotificationURLValidator._normalize_host( 

1002 NotificationURLValidator._extract_host(url) 

1003 ) 

1004 if not hostname: 1004 ↛ 1005line 1004 didn't jump to line 1005 because the condition on line 1004 was never true

1005 return False, error_msg, False 

1006 return ( 

1007 False, 

1008 error_msg, 

1009 NotificationURLValidator._compute_hint(hostname, None), 

1010 ) 

1011 

1012 # URL passed Phase 1. For non-http(s) URLs or IP literals, 

1013 # validation is complete. 

1014 try: 

1015 parsed_scheme = urlparse(url).scheme.lower() 

1016 except Exception: 

1017 parsed_scheme = "" 

1018 

1019 hostname = NotificationURLValidator._normalize_host( 

1020 NotificationURLValidator._extract_host(url) 

1021 ) 

1022 

1023 is_http_hostname = False 

1024 if hostname and parsed_scheme in ("http", "https"): 

1025 try: 

1026 ipaddress.ip_address(hostname) 

1027 except ValueError: 

1028 is_http_hostname = True 

1029 

1030 if not is_http_hostname: 

1031 # Non-http(s) or IP literal — fully validated in Phase 1. 

1032 return True, None, False 

1033 

1034 # ------------------------------------------------------------------ 

1035 # Phase 2: resolve DNS ONCE for the http(s) hostname and run the 

1036 # real IP check. resolved_ips is authoritative for the rejection 

1037 # and hint check — no additional DNS lookups. 

1038 # 

1039 # [] = DNS attempted and failed (fail-open, no retry). 

1040 # [...] = resolved IPs shared across all policy decisions. 

1041 # ------------------------------------------------------------------ 

1042 dns_result = NotificationURLValidator._resolve_hostname_ips(hostname) 

1043 resolved_ips: Optional[List[ResolvedIP]] = ( 

1044 dns_result if dns_result is not None else [] 

1045 ) 

1046 

1047 is_valid, error_msg = ( 

1048 NotificationURLValidator._validate_service_url_impl( 

1049 url, 

1050 allow_private_ips=allow_private_ips, 

1051 resolved_ips=resolved_ips, 

1052 ) 

1053 ) 

1054 if is_valid: 

1055 return True, None, False