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 .notification_destination import ( 

20 NotificationDestinationError, 

21 NotificationDestinationKind, 

22 normalize_host_for_transport, 

23 parse_notification_destination, 

24) 

25from .ip_ranges import PRIVATE_IP_RANGES as _PRIVATE_IP_RANGES 

26from .legacy_ipv4 import ( 

27 AMBIGUOUS_NUMERIC_IPV4_HOST_ERROR, 

28 ENCODED_NUMERIC_IPV4_HOST_ERROR, 

29 is_ambiguous_numeric_ipv4_host, 

30 is_percent_encoded_numeric_ipv4_host, 

31) 

32from .log_sanitizer import redact_and_bound_for_log 

33from .ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE, redact_url_for_log 

34 

35# (#6938) Cap for a request-supplied hostname or scheme in a log line, and 

36# in the rejection messages this module returns, which callers log 

37# (``NotificationService.test_service``) — a scoped IPv6 literal such as 

38# ``[fe80::1%25<100k chars>]`` is accepted by the parser and rejected here 

39# with its full host. 256 keeps every name within the 253-char DNS limit 

40# intact. Each value is redacted before it is cut (``redact_and_bound_for_log``). 

41_LOG_HOST_MAX_CHARS = 256 

42 

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

44# ipaddress._BaseAddress API). 

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

46 

47# ``localhost`` carries an inline case-insensitive group rather than a 

48# pattern-wide re.IGNORECASE: hostnames are case-insensitive, but the other two 

49# alternatives are already case-agnostic (``\d`` and explicit ``A-Za-z`` 

50# classes), and a blanket flag would additionally fold Unicode look-alikes 

51# (U+017F, U+212A) into those ``A-Za-z`` classes. Scoping the flag keeps the 

52# widening to the one alternative that needed it. 

53# ``(?://)?`` mirrors the IPv6 arm below: a scheme-relative fragment names the 

54# same host as its bare spelling, so ``//localhost/x`` has to fail closed 

55# exactly like ``localhost/x``. 

56_SCHEMELESS_URL_FRAGMENT_RE = re.compile( 

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

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

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

60) 

61# A userinfo prefix hides the host from the anchored pattern above, so 

62# ``user@127.0.0.1/x`` would otherwise read as opaque data. Two deliberate 

63# narrowings keep this from swallowing ordinary URL data: 

64# * only the loopback-name and IPv4-literal alternatives are carried over. 

65# This alternation, not the trailer below, is what keeps 

66# ``recipient@example.com`` in a ``?to=`` list an e-mail address: 

67# ``more:secret:parts@example.com/webhook`` is credentials in front of an 

68# ordinary domain — exactly the shape an Apprise service URL uses — and a 

69# public domain is not a blocked shortcut in the first place. 

70# * a ``[/?#]`` trailer is required rather than also accepting 

71# end-of-string. That is what keeps a bare ``admin@localhost`` or 

72# ``recipient@127.0.0.1`` — a recipient whose domain part happens to be 

73# spelled like a loopback host — an address rather than an authority. 

74_USERINFO_URL_FRAGMENT_RE = re.compile( 

75 r"^(?://)?[^/?#@\s]*@(?:(?i:localhost)|(?:\d{1,3}\.){3}\d{1,3})" 

76 r"(?::\d+)?[/?#]" 

77) 

78# Legacy inet_aton spellings (decimal, hexadecimal, octal, shortened) name the 

79# same host as their dotted-quad form; ``legacy_ipv4`` already owns that 

80# grammar for whole URLs. The ``[/?#]`` trailer is NOT the safeguard here — 

81# it is the common case. Apprise's comma-separated target lists put a ``/`` 

82# or a ``?`` directly after the last id (``tgram://…/111111111,222222222? 

83# format=markdown``, ``twilio://…/2125551234,4085551234?batch=yes``), and the 

84# one-component inet_aton form accepts every decimal below 2**32, so a bare 

85# id would read as a host. ``_names_legacy_numeric_host`` therefore demands 

86# an authority marker as well: two or more dot-separated components, a 

87# ``//`` prefix, or a ``userinfo@`` prefix. ``%`` is in the host class so the 

88# percent-encoded spellings reach the same decision. 

89_NUMERIC_HOST_FRAGMENT_RE = re.compile( 

90 r"^(?P<prefix>(?://)?(?:[^/?#@\s]*@)?)" 

91 r"(?P<host>[0-9A-Fa-fXx.%]+)(?::\d+)?[/?#]" 

92) 

93# Keep candidate detection deliberately broad. IPv6Address below is the sole 

94# authority for validity, including dotted-quad and scoped IPv6 forms. 

95# Brackets delimit the authority instead of belonging to the address, so the 

96# bracketed arm takes a repeated or unclosed pair and any trailing run that 

97# is not a separator: ``[::1``, ``[[::1]]`` and ``[::1]x`` each name ``::1``. 

98_IPV6_FRAGMENT_CANDIDATE_RE = re.compile( 

99 r"^(?://)?(?:\[+(?P<bracketed>[^\s,\[\]/?#]+)" 

100 r"[^\s,/?#]*|" 

101 r"(?P<bare>(?=[^\s,/?#]*:[^\s,/?#]*:)[^\s,/?#]+))" 

102 r"(?:[/?#]|$)" 

103) 

104_PERCENT_ENCODED_BRACKETS = {"%5b": "[", "%5d": "]"} 

105_PERCENT_ENCODED_BRACKET_RE = re.compile( 

106 "|".join(_PERCENT_ENCODED_BRACKETS), re.IGNORECASE 

107) 

108# ``%2e`` spells the label separator, so it decides both where a host's 

109# trailing root dot ends and how many components the host has. 

110_PERCENT_ENCODED_DOTS = ("%2e", "%2E") 

111# One scan finds the end of a fragment's authority for every helper below. 

112_AUTHORITY_END_RE = re.compile(r"[/?#]") 

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

114# Keep this boundary zero-width to avoid quadratic backtracking on 

115# separator runs. 

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

117_APPRISE_TELEGRAM_URL_RE = re.compile( 

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

119 re.IGNORECASE, 

120) 

121 

122 

123def _strip_notification_url_delimiters(entry: str) -> str: 

124 start = 0 

125 end = len(entry) 

126 while start < end and (entry[start] == "," or entry[start].isspace()): 

127 start += 1 

128 while end > start and (entry[end - 1] == "," or entry[end - 1].isspace()): 

129 end -= 1 

130 return entry[start:end] 

131 

132 

133def _is_ipv6_literal(host: str) -> bool: 

134 try: 

135 ipaddress.IPv6Address(host) 

136 except ipaddress.AddressValueError: 

137 return False 

138 return True 

139 

140 

141def _decode_bracket_escapes(fragment: str) -> str: 

142 """Decode ``%5b``/``%5d`` so an encoded IPv6 literal stays recognizable.""" 

143 return _PERCENT_ENCODED_BRACKET_RE.sub( 

144 lambda match: _PERCENT_ENCODED_BRACKETS[match.group(0).lower()], 

145 fragment, 

146 ) 

147 

148 

149def _has_label_separator(host: str) -> bool: 

150 """Return whether a host carries a dot in either spelling.""" 

151 return "." in host or any(dot in host for dot in _PERCENT_ENCODED_DOTS) 

152 

153 

154def _strip_trailing_host_dots(host: str) -> str: 

155 r"""Drop a host's trailing root dots, ``%2e`` included. 

156 

157 A fragment's authority is attacker-supplied and unbounded, so the 

158 literal dots come off with one C-level ``rstrip`` and the encoded ones 

159 walk back by index. Repeated slicing and a ``(?:\.|%2e)+$`` regex are 

160 both quadratic on a long run of dots. 

161 """ 

162 stripped = host.rstrip(".") 

163 if not stripped.endswith(_PERCENT_ENCODED_DOTS): 

164 return stripped 

165 end = len(stripped) 

166 while True: 

167 if end >= 3 and stripped[end - 3 : end].lower() == "%2e": 

168 end -= 3 

169 elif end > 0 and stripped[end - 1] == ".": 

170 end -= 1 

171 else: 

172 return stripped[:end] 

173 

174 

175def _authority_end(rest: str) -> int: 

176 """Return the offset of a fragment's first ``/``, ``?`` or ``#``.""" 

177 boundary = _AUTHORITY_END_RE.search(rest) 

178 return len(rest) if boundary is None else boundary.start() 

179 

180 

181def _strip_fragment_host_dots(fragment: str) -> str: 

182 """Drop a host's trailing dots the way ``_normalize_host`` does. 

183 

184 ``localhost.`` and ``localhost`` are one host to every resolver, so the 

185 fully-qualified spelling must not read as opaque data here when the bare 

186 one fails closed. ``%2e`` spells the same dot, so ``localhost%2e`` 

187 normalizes onto the same host as ``localhost.`` does. 

188 """ 

189 prefix, rest = "", fragment 

190 if rest.startswith("//"): 

191 prefix, rest = "//", rest[2:] 

192 boundary = _authority_end(rest) 

193 host, tail = rest[:boundary], rest[boundary:] 

194 port = "" 

195 colon = host.rfind(":") 

196 # Split a port off a colon-free host, or off a bracketed one: a BARE 

197 # IPv6 literal's own colons are part of the address rather than a port 

198 # delimiter, but a bracketed literal has already closed, so ``[::1].:80`` 

199 # has to split the way ``[::1]:80`` does. 

200 if ( 

201 colon > 0 

202 and (":" not in host[:colon] or "]" in host[:colon]) 

203 and host[colon + 1 :].isdigit() 

204 ): 

205 host, port = host[:colon], host[colon:] 

206 stripped = _strip_trailing_host_dots(host) 

207 if stripped == host: 

208 return fragment 

209 return prefix + stripped + port + tail 

210 

211 

212def _strip_fragment_userinfo(fragment: str) -> str: 

213 """Drop a ``userinfo@`` prefix from a fragment's authority component.""" 

214 prefix, rest = "", fragment 

215 if rest.startswith("//"): 

216 prefix, rest = "//", rest[2:] 

217 authority_end = _authority_end(rest) 

218 # The LAST ``@`` inside the authority delimits userinfo; any ``@`` in the 

219 # path or query belongs to the data, not to the host. 

220 at_sign = rest.rfind("@", 0, authority_end) 

221 return fragment if at_sign < 0 else prefix + rest[at_sign + 1 :] 

222 

223 

224def _is_ipv6_url_fragment(fragment: str) -> bool: 

225 candidate = _IPV6_FRAGMENT_CANDIDATE_RE.match(fragment) 

226 if candidate is None: 

227 return False 

228 

229 bracketed = candidate.group("bracketed") 

230 host = bracketed or candidate.group("bare") 

231 if _is_ipv6_literal(host): 

232 return True 

233 if bracketed is not None: 

234 return False 

235 # The bare alternative has no port group, so its greedy run swallows a 

236 # trailing ``:<port>``. Without this retry ``::1:99999`` -- a valid host 

237 # with an out-of-range port -- fails ``IPv6Address`` as one literal and is 

238 # silently accepted as URL data, while the bracketed ``[::1]:99999`` is 

239 # refused. Retry once with the final colon-segment removed so both forms 

240 # fail closed on the same input. 

241 head, sep, _tail = host.rpartition(":") 

242 return bool(sep) and _is_ipv6_literal(head) 

243 

244 

245def _names_legacy_numeric_host(prefix: str, host: str) -> bool: 

246 """Return whether a fragment's leading token is an inet_aton authority. 

247 

248 A one-component inet_aton host is any integer below 2**32, which is also 

249 the shape of every Telegram chat id, Twilio number and Unix timestamp 

250 that an Apprise comma-separated target list leaves in front of a ``/`` 

251 or ``?``. Such a token only names a host once the fragment carries an 

252 authority marker of its own: a second dot-separated component (a 

253 trailing root dot does not count), a ``//`` prefix, or a ``userinfo@`` 

254 prefix. 

255 """ 

256 # A trailing root dot (``.`` or ``%2e``) is not a second component: 

257 # ``222222222%2e`` is still a one-component numeric token. 

258 bare_host = re.sub(r"(?:\.|%2[eE])+$", "", host) 

259 if not prefix and not _has_label_separator(bare_host): 

260 return False 

261 return is_ambiguous_numeric_ipv4_host( 

262 host 

263 ) or is_percent_encoded_numeric_ipv4_host(host) 

264 

265 

266def _matches_scheme_less_url_shape(fragment: str) -> bool: 

267 if _SCHEMELESS_URL_FRAGMENT_RE.match(fragment): 

268 return True 

269 if _USERINFO_URL_FRAGMENT_RE.match(fragment): 

270 return True 

271 

272 numeric = _NUMERIC_HOST_FRAGMENT_RE.match(fragment) 

273 if numeric is not None and _names_legacy_numeric_host( 

274 numeric.group("prefix"), numeric.group("host") 

275 ): 

276 return True 

277 

278 return _is_ipv6_url_fragment(fragment) 

279 

280 

281def _is_scheme_less_url_fragment(fragment: str) -> bool: 

282 """Recognize a comma-separated fragment that names a host without a scheme. 

283 

284 Every check is additive: an alternate spelling is only ever consulted 

285 after the literal fragment has been cleared, so normalization can widen 

286 the refusal set but can never readmit a fragment that already fails 

287 closed. 

288 """ 

289 if _matches_scheme_less_url_shape(fragment): 

290 return True 

291 # Spellings that normalize onto a host the checks above already refuse: 

292 # percent-encoded brackets and a trailing root dot, encoded or literal. 

293 normalized = _strip_fragment_host_dots(_decode_bracket_escapes(fragment)) 

294 if normalized != fragment and _matches_scheme_less_url_shape(normalized): 

295 return True 

296 # A userinfo prefix hides an IPv6 authority from the anchored patterns 

297 # above. Only the IPv6 arm is retried on the stripped form: stripping 

298 # userinfo from ``recipient@example.com`` would read an e-mail address as 

299 # a hostname, which is why the hostname arm demands the ``[/?#]`` trailer 

300 # that proves an authority instead. 

301 without_userinfo = _strip_fragment_userinfo(normalized) 

302 return without_userinfo != normalized and _is_ipv6_url_fragment( 

303 without_userinfo 

304 ) 

305 

306 

307def parse_notification_url_list( 

308 urls: str, separator: str = "," 

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

310 r"""Partition notification URLs without dropping malformed fragments. 

311 

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

313 This preserves commas inside one Apprise service URL while keeping 

314 whitespace a hard separator. Commas and surrounding whitespace — any 

315 Unicode whitespace, matching the bare ``str.strip()`` every other 

316 consumer of an entry applies — are trimmed from each entry's ENDS 

317 before it is inspected, so a value pasted with a trailing NBSP is not 

318 mistaken for a URL with an illegal character in it. An entry whose 

319 head carries a scheme but which contains characters that are illegal 

320 unencoded in a URI (whitespace, control bytes, backslash — 

321 ``RFC_FORBIDDEN_URL_CHARS_RE``, the same class ``validate_service_url`` 

322 rejects) is malformed no matter what trails it: the token after the first illegal character is returned 

323 as the invalid fragment, so ``discord://x garbage`` fails closed exactly 

324 like ``discord://x example.com/hook`` instead of being dispatched as a 

325 single entry. When the illegal character only TRAILS the entry 

326 (``slack://t/x/y\``) there is no token after it, so the whole entry is 

327 returned as the fragment. URL-like fragments without a scheme are 

328 returned separately so callers can fail closed instead of dispatching 

329 only the valid subset. 

330 

331 ``invalid_fragment`` is a REFUSAL SIGNAL, not a substitute URL, and it 

332 carries three warnings for callers: 

333 

334 * It may CARRY A SCHEME. ``RFC_FORBIDDEN_URL_CHARS_RE`` covers 

335 backslash and the non-``\s`` control bytes, which the 

336 scheme-boundary regex does not split on, so 

337 ``a://h1/p\https://h2/q`` stays ONE entry whose fragment is 

338 ``https://h2/q``. A caller that validates the fragment *in place of* 

339 the entry validates a decoy while the raw original string — pointing 

340 at ``h1`` — is what gets dispatched. 

341 * It may be VALID on its own. Never assume validating it yields an 

342 error message. 

343 * It may CONTAIN CREDENTIALS. When the illegal character trails the 

344 entry the fragment IS the entry, and for token-in-authority Apprise 

345 schemes even a ``scheme://host`` redaction of it is the secret. Log 

346 its length, never its content. 

347 

348 The only correct response to a non-``None`` fragment is to refuse the 

349 whole input. 

350 

351 Custom separators are explicit boundaries: commas and whitespace inside 

352 each resulting URL remain data, subject to service URL validation. Every 

353 parsed entry still requires a top-level URL scheme. The scheme-less 

354 fragment check below is the one guard a custom separator does NOT get: 

355 its entries take ``validate_service_url`` but are never scanned for 

356 comma-separated scheme-less fragments, because with an explicit separator 

357 a comma is ordinary data inside one URL. 

358 

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

360 historical explicit-split behavior. 

361 """ 

362 if separator != ",": 

363 parsed_urls = [ 

364 entry.strip() for entry in urls.split(separator) if entry.strip() 

365 ] 

366 else: 

367 parsed_urls = [] 

368 for entry in _URL_BOUNDARY_RE.split(urls): 

369 normalized_entry = _strip_notification_url_delimiters(entry) 

370 if normalized_entry: 

371 parsed_urls.append(normalized_entry) 

372 for entry in parsed_urls: 

373 if not _URL_SCHEME_RE.match(entry): 

374 return parsed_urls, entry 

375 if separator != ",": 

376 continue 

377 # A URL entry containing characters illegal unencoded in a URI is 

378 # malformed regardless of what trails it: ``discord://x garbage`` 

379 # is not one URL with "garbage" in it, it is a URL followed by 

380 # junk. Any token after the first illegal character surfaces as 

381 # the invalid fragment so every caller fails closed on the shape. 

382 if RFC_FORBIDDEN_URL_CHARS_RE.search(entry): 

383 fragments = [ 

384 fragment 

385 for fragment in RFC_FORBIDDEN_URL_CHARS_RE.split(entry) 

386 if fragment 

387 ] 

388 # fragments[0] is the prefix before the illegal character; the 

389 # next non-empty token is the smuggled fragment. When the 

390 # illegal character only trails the entry, the entry itself 

391 # is the malformed fragment. 

392 return parsed_urls, fragments[1] if len(fragments) > 1 else entry 

393 # Commas are legal inside one Apprise URL (e.g. a multi-target 

394 # ``?to=id1,id2``), so a comma-separated fragment only fails closed 

395 # when it is itself a scheme-less URL that Apprise could repair 

396 # away or absorb into a neighbouring entry. 

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

398 if _is_scheme_less_url_fragment(fragment): 

399 return parsed_urls, fragment 

400 return parsed_urls, None 

401 

402 

403class NotificationURLValidationError(ValueError): 

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

405 

406 pass 

407 

408 

409class NotificationURLValidator: 

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

411 

412 # Dangerous protocols that should never be used for notifications 

413 BLOCKED_SCHEMES = ( 

414 "file", # Local file access 

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

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

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

418 "javascript", # XSS/code execution 

419 "vbscript", # XSS/code execution 

420 "about", # Browser internal 

421 "blob", # Browser internal 

422 ) 

423 

424 # Allowed protocols for notification services 

425 ALLOWED_SCHEMES = ( 

426 "http", # Webhook services 

427 "https", # Webhook services (preferred) 

428 "mailto", # Email notifications 

429 "discord", # Discord webhooks 

430 "slack", # Slack webhooks 

431 "tgram", # Apprise Telegram scheme 

432 "gotify", # Gotify notifications 

433 "pushover", # Pushover notifications 

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

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

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

437 "matrix", # Matrix protocol 

438 "mattermost", # Mattermost webhooks 

439 "rocketchat", # Rocket.Chat webhooks 

440 "teams", # Microsoft Teams 

441 "json", # Generic JSON webhooks 

442 "xml", # Generic XML webhooks 

443 "form", # Form-encoded webhooks 

444 ) 

445 

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

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

448 # 

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

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

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

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

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

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

455 

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

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

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

459 BLOCKED_MAILTO_QUERY_KEYS = frozenset( 

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

461 ) 

462 

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

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

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

466 { 

467 "signal", 

468 "gotify", 

469 "ntfy", 

470 "ntfys", 

471 "mattermost", 

472 "rocketchat", 

473 "matrix", 

474 "json", 

475 "xml", 

476 "form", 

477 "mailto", 

478 } 

479 ) 

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

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

482 ) 

483 

484 # Reuse shared private IP range definitions 

485 PRIVATE_IP_RANGES = _PRIVATE_IP_RANGES 

486 

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

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

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

490 # single source of truth. 

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

492 

493 @staticmethod 

494 def _ip_matches_blocked_range( 

495 ip, 

496 allow_private_ips: bool = False, 

497 allow_nat64: Optional[bool] = None, 

498 block_link_local: bool = False, 

499 ) -> bool: 

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

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

502 single source of truth. 

503 

504 Honors: 

505 - ALWAYS_BLOCKED_METADATA_IPS (cloud metadata, absolute) 

506 - is_nat64_wrapped_metadata_ip (NAT64-wrapped IMDS, absolute) 

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

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

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

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

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

512 above still fire. This closes the historical bypass where 

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

514 and let metadata IPs through the notification path. 

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

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

517 allow_private_ips=True. The notification validator and send 

518 guards pass it for every host-bearing destination (http(s) 

519 included) so metadata reachable in link-local beyond the 

520 always-blocked literals cannot slip through the opt-in. 

521 """ 

522 from .ssrf_validator import is_ip_blocked 

523 

524 return is_ip_blocked( 

525 str(ip), 

526 allow_private_ips=allow_private_ips, 

527 allow_nat64=allow_nat64, 

528 block_link_local=block_link_local, 

529 ) 

530 

531 @staticmethod 

532 def _resolve_hostname_ips( 

533 hostname: str, 

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

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

536 timeout. Returns None on resolution failure or timeout. 

537 

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

539 can share a single resolution pass — see 

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

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

542 default-level and elevated-level decisions. 

543 """ 

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

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

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

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

548 # classic resolve-vs-connect TOCTOU window. 

549 # 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

569 # residual race. Every partition requires the operator's private-IP 

570 # opt-in for private addresses (LAN targets or a private forward 

571 # proxy); cloud metadata remains blocked. Because the pin/block are thread-local, 

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

573 # 

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

575 # gated behind an env-only master switch 

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

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

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

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

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

581 # 

582 # Plugin schemes are not categorically exempt from authority 

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

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

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

586 # 

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

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

589 try: 

590 executor = ThreadPoolExecutor(max_workers=1) 

591 try: 

592 future = executor.submit( 

593 socket.getaddrinfo, 

594 hostname, 

595 None, 

596 socket.AF_UNSPEC, 

597 socket.SOCK_STREAM, 

598 ) 

599 resolved_ips = future.result(timeout=5) 

600 finally: 

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

602 return [ 

603 ipaddress.ip_address(sockaddr[0]) 

604 for _family, _, _, _, sockaddr in resolved_ips 

605 ] 

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

607 logger.warning( 

608 "DNS resolution failed for hostname {} — " 

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

610 redact_and_bound_for_log(hostname, _LOG_HOST_MAX_CHARS), 

611 ) 

612 return None 

613 

614 @staticmethod 

615 def _host_blocks_at_level( 

616 hostname: str, 

617 allow_private_ips: bool = False, 

618 allow_nat64: Optional[bool] = None, 

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

620 block_link_local: bool = False, 

621 ) -> bool: 

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

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

624 a prior DNS resolution. 

625 

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

627 single resolution across multiple policy decisions, closing 

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

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

630 via ``_resolve_hostname_ips``. 

631 

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

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

634 """ 

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

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

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

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

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

640 "localhost", 

641 "127.0.0.1", 

642 "::1", 

643 # Recognize and validate a local address; no socket is bound here. 

644 "0.0.0.0", # nosec B104 

645 "::", 

646 ): 

647 return True 

648 

649 # Try to parse as IP address first. 

650 try: 

651 ip = ipaddress.ip_address(hostname) 

652 return NotificationURLValidator._ip_matches_blocked_range( 

653 ip, 

654 allow_private_ips=allow_private_ips, 

655 allow_nat64=allow_nat64, 

656 block_link_local=block_link_local, 

657 ) 

658 except ValueError: 

659 pass 

660 

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

662 if resolved_ips is None: 

663 resolved_ips = NotificationURLValidator._resolve_hostname_ips( 

664 hostname 

665 ) 

666 if resolved_ips is None: 

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

668 return False 

669 for ip in resolved_ips: 

670 if NotificationURLValidator._ip_matches_blocked_range( 

671 ip, 

672 allow_private_ips=allow_private_ips, 

673 allow_nat64=allow_nat64, 

674 block_link_local=block_link_local, 

675 ): 

676 return True 

677 return False 

678 

679 @staticmethod 

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

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

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

683 Returns None if urllib3 rejects the URL. 

684 """ 

685 try: 

686 u3 = parse_url(url) 

687 except LocationParseError: 

688 return None 

689 return u3.host 

690 

691 @staticmethod 

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

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

694 bracket notation and trailing dots. Shared by 

695 ``validate_service_url`` and ``validate_service_url_with_hint`` 

696 so the two call sites cannot drift. 

697 """ 

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

699 hostname = hostname[1:-1] 

700 if hostname: 

701 hostname = hostname.rstrip(".") 

702 return hostname or "" 

703 

704 @staticmethod 

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

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

707 

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

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

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

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

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

713 

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

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

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

717 """ 

718 query_at = url.find("?") 

719 if query_at < 0: 

720 return [], False 

721 

722 raw_query = url[query_at + 1 :] 

723 keys = [] 

724 hex_digits = "0123456789abcdefABCDEF" 

725 

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

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

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

729 

730 index = 0 

731 while index < len(raw_key): 

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

733 index += 1 

734 continue 

735 if ( 

736 index + 2 >= len(raw_key) 

737 or raw_key[index + 1] not in hex_digits 

738 or raw_key[index + 2] not in hex_digits 

739 ): 

740 return keys, True 

741 index += 3 

742 

743 apprise_key = ( 

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

745 if raw_key 

746 else "" 

747 ) 

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

749 

750 return keys, False 

751 

752 @staticmethod 

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

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

755 keys, malformed = NotificationURLValidator._apprise_query_keys(url) 

756 if malformed: 

757 logger.warning( 

758 "Blocked notification URL with malformed percent-encoding " 

759 "in a query parameter name" 

760 ) 

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

762 

763 blocked_keys = NotificationURLValidator.BLOCKED_APPRISE_QUERY_KEYS 

764 if scheme == "mailto": 

765 blocked_keys = ( 

766 blocked_keys 

767 | NotificationURLValidator.BLOCKED_MAILTO_QUERY_KEYS 

768 ) 

769 

770 for key in keys: 

771 if key in blocked_keys: 

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

773 # alternate destination selected by an attacker. 

774 logger.warning( 

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

776 key, 

777 scheme, 

778 ) 

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

780 

781 return None 

782 

783 @staticmethod 

784 def _is_private_ip( 

785 hostname: str, 

786 allow_private_ips: bool = False, 

787 allow_nat64: Optional[bool] = None, 

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

789 block_link_local: bool = False, 

790 ) -> bool: 

791 """ 

792 Check if hostname resolves to a private IP address. 

793 

794 Args: 

795 hostname: Hostname to check 

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

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

798 metadata IPs and NAT64-wrapped metadata IPs are blocked 

799 regardless — the operator opt-in cannot license IMDS 

800 exposure. 

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

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

803 answers the hint probe. 

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

805 blocked even under allow_private_ips=True. Passed for every 

806 host-bearing notification destination, http(s) included 

807 (metadata lives in link-local beyond the always-blocked 

808 literals). 

809 

810 Returns: 

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

812 allow_private_ips), or wraps a metadata IP unconditionally 

813 """ 

814 return NotificationURLValidator._host_blocks_at_level( 

815 hostname, 

816 allow_private_ips=allow_private_ips, 

817 allow_nat64=allow_nat64, 

818 resolved_ips=_resolved_ips, 

819 block_link_local=block_link_local, 

820 ) 

821 

822 @staticmethod 

823 def validate_service_url( 

824 url: str, 

825 allow_private_ips: bool = False, 

826 allow_nat64: Optional[bool] = None, 

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

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

829 

830 Thin public wrapper around ``_validate_service_url_impl``. 

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

832 resolution state). 

833 

834 Args: 

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

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

837 Set to True for development/testing environments 

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

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

840 an explicit bool overrides it. 

841 

842 Returns: 

843 Tuple of (is_valid, error_message) 

844 """ 

845 return NotificationURLValidator._validate_service_url_impl( 

846 url, 

847 allow_private_ips=allow_private_ips, 

848 allow_nat64=allow_nat64, 

849 ) 

850 

851 @staticmethod 

852 def _validate_service_url_impl( 

853 url: str, 

854 allow_private_ips: bool = False, 

855 allow_nat64: Optional[bool] = None, 

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

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

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

859 

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

861 ``validate_service_url_with_hint`` to share a single DNS 

862 resolution across multiple policy decisions. Must NEVER be 

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

864 bypass hostname resolution and private-IP checks. 

865 

866 Returns: 

867 Tuple of (is_valid, error_message) 

868 - is_valid: True if URL passes security checks 

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

870 

871 Examples: 

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

873 (True, None) 

874 

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

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

877 

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

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

880 

881 Caller contract: 

882 ``notifications.service.NotificationService.test_service`` 

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

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

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

886 rejection targets a recoverable destination — one that 

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

888 CGNAT / loopback / IPv6 ULA, plus NAT64-wrapped 

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

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

891 targets an always-blocked category (link-local, which every 

892 host-bearing notification destination blocks, cloud-metadata IPs, 6to4, 

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

894 IPv4-Compatible IPv6, IPv4-Translated/SIIT, NAT64-wrapped 

895 metadata) and the hint is suppressed because 

896 naming the env var would mislead. The parametrized 

897 integration test ``test_test_service_ip_rejection_matrix`` 

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

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

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

901 needs updating. 

902 """ 

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

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

905 

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

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

908 url = url.strip() 

909 

910 # Reject URLs containing characters that drive parser-differential 

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

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

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

914 # URLs are by definition adversarial-shaped. 

915 if RFC_FORBIDDEN_URL_CHARS_RE.search(url): 

916 logger.warning( 

917 "Blocked notification URL containing RFC-illegal characters" 

918 ) 

919 return ( 

920 False, 

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

922 ) 

923 

924 # Parse URL 

925 try: 

926 parsed = urlparse(url) 

927 except Exception: 

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

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

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

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

932 # traceback to match the sibling LocationParseError handler below 

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

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

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

936 return False, "Invalid URL format" 

937 

938 # Check for scheme 

939 if not parsed.scheme: 

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

941 

942 scheme = parsed.scheme.lower() 

943 

944 # Check for blocked schemes 

945 if scheme in NotificationURLValidator.BLOCKED_SCHEMES: 

946 logger.warning( 

947 "Blocked unsafe notification protocol: {} in URL: {}", 

948 redact_and_bound_for_log(scheme, _LOG_HOST_MAX_CHARS), 

949 redact_url_for_log(url), 

950 ) 

951 return ( 

952 False, 

953 "Blocked unsafe protocol: " 

954 f"{redact_and_bound_for_log(scheme, _LOG_HOST_MAX_CHARS)}", 

955 ) 

956 

957 # Check for allowed schemes 

958 if scheme not in NotificationURLValidator.ALLOWED_SCHEMES: 

959 if scheme == "telegram": 

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

961 # scheme removed upstream; point users at the canonical 

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

963 return ( 

964 False, 

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

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

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

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

969 ) 

970 logger.warning( 

971 "Unknown notification protocol: {} in URL: {}", 

972 redact_and_bound_for_log(scheme, _LOG_HOST_MAX_CHARS), 

973 redact_url_for_log(url), 

974 ) 

975 return ( 

976 False, 

977 "Unsupported protocol: " 

978 f"{redact_and_bound_for_log(scheme, _LOG_HOST_MAX_CHARS)}. " 

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

980 ) 

981 

982 # Apprise supports query parameters that can select resources or 

983 # outbound destinations independently of the authority we validate 

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

985 # performing any attacker-controlled DNS lookup. 

986 query_error = NotificationURLValidator._validate_apprise_query_policy( 

987 url, scheme 

988 ) 

989 if query_error: 

990 return False, query_error 

991 

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

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

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

995 # authority denotes: for 

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

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

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

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

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

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

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

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

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

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

1006 logger.warning( 

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

1008 "(parser-differential SSRF risk)" 

1009 ) 

1010 return ( 

1011 False, 

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

1013 "which is not allowed", 

1014 ) 

1015 

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

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

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

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

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