Coverage for src/local_deep_research/notifications/service.py: 93%

1""" 

2Core notification service using Apprise. 

3""" 

4 

5import contextlib 

6import functools 

7import re 

8from typing import Dict, List, Optional, Any, Tuple 

9from urllib.parse import urlparse 

10 

11import apprise 

12from loguru import logger 

13from tenacity import ( 

14 retry, 

15 stop_after_attempt, 

16 wait_exponential, 

17 retry_if_not_exception_type, 

18) 

19 

20from .exceptions import ServiceError, SendError, SecurityBlockError 

21from .apprise_log_utils import install_apprise_log_record_factory 

22from .templates import EventType, NotificationTemplate 

23from ..security import dns_pinning 

24from ..security.notification_destination import ( 

25 NotificationDestinationError, 

26 NotificationDestinationKind, 

27 parse_notification_destination, 

28) 

29from ..security.notification_validator import ( 

30 NotificationURLValidator, 

31 parse_notification_url_list, 

32) 

33 

34PRIVATE_IP_REJECTION_PREFIX = ( 

35 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX 

36) 

37 

38 

39# Backward compatibility constants - now handled by Tenacity internally 

40MAX_RETRY_ATTEMPTS = 3 

41INITIAL_RETRY_DELAY = 0.5 

42RETRY_BACKOFF_MULTIPLIER = 2 

43 

44# Maximum number of notification targets Apprise may parse out of one 

45# test/configured URL string. Prevents an authenticated caller from 

46# triggering unbounded DNS resolution work (and an unbounded fan-out of 

47# outbound requests) via a comma/space-separated hostname list. 

48MAX_NOTIFICATION_TARGETS = 20 

49 

50 

51class NotificationService: 

52 """ 

53 Low-level notification service that wraps Apprise. 

54 """ 

55 

56 # Regex patterns for common service types (for validation) 

57 SERVICE_PATTERNS = { 

58 "email": r"^mailto://", 

59 "discord": r"^discord://", 

60 "slack": r"^slack://", 

61 "telegram": r"^tgram://", 

62 "smtp": r"^(smtp|smtps)://", 

63 } 

64 

65 def __init__( 

66 self, 

67 allow_private_ips: bool = False, 

68 outbound_allowed: bool = False, 

69 ): 

70 """ 

71 Initialize the notification service. 

72 

73 Args: 

74 allow_private_ips: Whether to allow notifications to private/local IPs 

75 (default: False for security). Set to True for 

76 development/testing environments only. 

77 outbound_allowed: Server-level master switch 

78 (env-only via LDR_NOTIFICATIONS_ALLOW_OUTBOUND). 

79 Default False — outbound notifications are off until 

80 the operator opts in. See SECURITY.md "Notification 

81 Webhook SSRF" for the rationale. 

82 """ 

83 install_apprise_log_record_factory() 

84 self.apprise = self._new_apprise() 

85 self.allow_private_ips = allow_private_ips 

86 self.outbound_allowed = outbound_allowed 

87 

88 @staticmethod 

89 def _new_apprise() -> apprise.Apprise: 

90 """Create an Apprise instance that delivers synchronously and does 

91 not follow HTTP redirects. 

92 

93 ``async_mode=False`` forces Apprise to run every plugin's 

94 ``notify()`` in the calling thread instead of fanning out to a 

95 ``ThreadPoolExecutor``. This is required for the SSRF hardening in 

96 :meth:`_send_with_retry`: the DNS pin and the block-private window 

97 (``security.dns_pinning``) are thread-local, so they only apply to 

98 the resolution the sending thread performs. With the default async 

99 fan-out the plugins would resolve in worker threads that carry no 

100 pin/block and the guard would silently not apply (verified: the 

101 default asset resolves target hosts on worker threads). 

102 

103 ``http_redirects=False`` disables 30x redirect-following for the 

104 raw-webhook plugins (``json``/``xml``/``form``, which pass 

105 ``allow_redirects=self.redirects`` to ``requests``). This closes the 

106 redirect attack class outright: a user-configured webhook can no 

107 longer 30x-redirect the send to a private/loopback host or to an 

108 arbitrary public host for data exfiltration. The pin + 

109 block-private window is then only responsible for the original 

110 validated host's own rebind, not for chasing redirect hops. It is a 

111 DEFAULT: a per-URL ``?redirect=yes`` can still override it, so 

112 :meth:`_disable_redirects` re-forces it off per-plugin after 

113 ``add()`` (see there). 

114 """ 

115 return apprise.Apprise( 

116 asset=apprise.AppriseAsset(async_mode=False, http_redirects=False) 

117 ) 

118 

119 @staticmethod 

120 def _disable_redirects(apprise_instance: apprise.Apprise) -> None: 

121 """Force redirect-following OFF on every added plugin. 

122 

123 ``_new_apprise`` sets the asset-level ``http_redirects=False`` 

124 default, but Apprise lets a per-URL ``?redirect=yes`` query 

125 parameter override that default (``self.redirects = 

126 parse_bool(kwargs.get("redirect", asset.http_redirects))``). A 

127 low-privilege user could therefore re-enable redirect-following on 

128 their own webhook and reopen redirect-to-private / redirect-to-exfil. 

129 Setting ``redirects = False`` directly on each plugin AFTER it is 

130 parsed makes the closure airtight: the value is read at send time 

131 (``allow_redirects=self.redirects``), so this wins over any URL 

132 parameter. Redirect-following is never needed for legitimate 

133 webhook/plugin delivery (real endpoints answer 2xx directly), so 

134 this cannot break a genuine send. 

135 

136 Not every plugin threads ``self.redirects`` into its HTTP call: the 

137 raw-webhook ``json``/``xml``/``form`` schemes and e.g. ``ntfy`` / 

138 ``gotify`` honour it, but a few (e.g. ``matrix://``) issue requests 

139 without it, so re-forcing the flag has no effect there. Those are 

140 backstopped by the block-private send window 

141 (``dns_pinning.pinned_notification_send``), which refuses a 

142 redirect/rebind to a blocked address at the ``getaddrinfo`` layer 

143 regardless of whether the plugin followed the redirect. 

144 """ 

145 try: 

146 servers = list(apprise_instance) 

147 except TypeError as exc: 

148 # A mocked Apprise in unit tests is not iterable, so there is 

149 # nothing to harden (those tests never exercise a real redirect). 

150 # But if a REAL Apprise ever stops being iterable (upstream API 

151 # drift), silently returning here would leave the per-plugin 

152 # redirect-disable un-applied — fail-open. The asset-level 

153 # ``http_redirects=False`` default and the block-private window 

154 # still apply, but a per-URL ``?redirect=yes`` could re-enable 

155 # redirect-following. Log a warning (narrowed to TypeError) so 

156 # that drift is visible instead of silent. 

157 logger.warning( 

158 "Could not enumerate Apprise servers to force " 

159 "redirect-following off per plugin ({}); relying on the " 

160 "asset-level default and the block-private window", 

161 type(exc).__name__, 

162 ) 

163 return 

164 for server in servers: 

165 server.redirects = False 

166 

167 def _enforce_guarded_send_invariants( 

168 self, apprise_instance: apprise.Apprise, tag: Optional[str] 

169 ) -> None: 

170 """Enforce the fail-closed invariants for a DNS-guarded send. 

171 

172 The pin + block-private window (``security.dns_pinning``) are 

173 thread-local and only cover IN-THREAD delivery. Refuse the send 

174 (fail closed) rather than silently sending unguarded unless: 

175 

176 * ``async_mode`` is off — otherwise Apprise fans the plugins out to 

177 worker threads that carry neither the pin nor the block; and 

178 * ``tag`` is None — a multi-token tag likewise makes Apprise 

179 dispatch matching plugins across worker threads. No production 

180 caller passes a tag; this neutralizes a future one. 

181 

182 Then force redirect-following off per plugin so a per-URL 

183 ``?redirect=yes`` cannot override the asset-level default. 

184 

185 Raises: 

186 SecurityBlockError: if either invariant is violated — a 

187 deliberate fail-closed security refusal, not an incidental 

188 runtime error, so it is raised directly as this type 

189 (rather than a generic ``RuntimeError`` a caller would have 

190 to re-classify) and is excluded from Tenacity retries so it 

191 fails fast. See ``send()``'s docstring for why this must 

192 not be mistaken for a transient delivery failure. 

193 """ 

194 if getattr(apprise_instance.asset, "async_mode", True) is True: 

195 raise SecurityBlockError( 

196 "Guarded notification send requires async_mode=False; " 

197 "refusing to send (the thread-local DNS pin/block would not " 

198 "apply to Apprise's worker-thread fan-out)." 

199 ) 

200 if tag is not None: 

201 raise SecurityBlockError( 

202 "Guarded notification send does not support tag targeting; " 

203 "refusing to send (a tag can fan delivery out to worker " 

204 "threads that bypass the thread-local DNS pin/block)." 

205 ) 

206 self._disable_redirects(apprise_instance) 

207 

208 @staticmethod 

209 def _partition_urls( 

210 service_urls: str, separator: str = "," 

211 ) -> Tuple[List[str], List[str]]: 

212 """Split an Apprise URL string into (http(s), everything-else). 

213 

214 Uses the same separator-aware parser as validation and egress filtering 

215 (``parse_notification_url_list``). The default comma parser recognizes 

216 scheme-aware boundaries — a comma/whitespace sequence only separates 

217 entries when a URL scheme follows it, so commas INSIDE one Apprise URL 

218 (e.g. a multi-target Telegram ``?to=id1,id2``) are preserved and 

219 dispatched as a single entry; an explicit custom separator preserves 

220 commas and whitespace inside each URL outright. ``_dispatch`` 

221 further separates host-bearing plugins from fixed vendor plugins 

222 and applies the matching send-time DNS policy. 

223 

224 NOTE: the ``strict`` (``http``/``https``) partition is only 

225 PARTLY deliverable. Apprise has no generic http(s) notifier, so an 

226 arbitrary webhook is rejected at ``add()`` 

227 (``apprise.Apprise().add("https://example.com/hook")`` is False on 

228 apprise 1.13.0) and a strict-only ``add()`` in :meth:`_dispatch` 

229 returns False without sending. But Apprise DOES claim the native 

230 ``https://`` forms of several vendors, so those deliver from this 

231 partition: 

232 ``apprise.Apprise().add("https://discord.com/api/webhooks/<id>/<token>")`` 

233 and the ``https://hooks.slack.com/services/...`` form both return 

234 True. Generic webhooks should still use the raw-webhook schemes 

235 ``json``/``xml``/``form`` (in the ``lenient`` partition). 

236 """ 

237 strict: List[str] = [] 

238 lenient: List[str] = [] 

239 entries, invalid_fragment = parse_notification_url_list( 

240 service_urls, separator 

241 ) 

242 if invalid_fragment is not None: 

243 # The only correct response to a non-``None`` fragment is to 

244 # refuse the whole input (see ``parse_notification_url_list``'s 

245 # docstring) — never dispatch the malformed entry verbatim. 

246 # Never log the fragment itself — it may contain credentials; 

247 # length/count only, as the other fragment-log sites do. 

248 # ``send()`` runs ``validate_multiple_urls`` (same 

249 # separator-aware parse) before reaching here, so a malformed 

250 # fragment is already rejected there; this branch only fires 

251 # for direct callers of this helper. Fail closed rather than 

252 # dispatching only the valid subset — the caller must validate 

253 # the list themselves before dispatching it. 

254 # Rendered positionally: loguru kwargs with no template 

255 # placeholder reach no production sink (#5619). 

256 logger.warning( 

257 "Notification service_url list contains a malformed " 

258 "fragment; refusing the whole list " 

259 "(fragment_length={}, entries_parsed={})", 

260 len(invalid_fragment), 

261 len(entries), 

262 ) 

263 return [], [] 

264 for entry in entries: 

265 scheme = entry.split(":", 1)[0].lower() if ":" in entry else "" 

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

267 strict.append(entry) 

268 else: 

269 lenient.append(entry) 

270 return strict, lenient 

271 

272 def _guarded_notify( 

273 self, 

274 title: str, 

275 body: str, 

276 apprise_instance: apprise.Apprise, 

277 guard_factory=None, 

278 tag: Optional[str] = None, 

279 attach: Optional[List[str]] = None, 

280 ) -> bool: 

281 """Run a single Apprise ``notify()`` under the DNS-guard and the 

282 fail-closed invariants. Returns Apprise's raw boolean result. 

283 

284 Single source of truth for the guarded send: both the retrying 

285 delivery path (:meth:`_send_with_retry`) and the admin 

286 "Send Test Notification" path (:meth:`test_service`) call this, so 

287 neither re-implements the invariant-enforcement / redirect-disable / 

288 guarded-``notify`` sequence that the SSRF hardening depends on. 

289 

290 When ``guard_factory`` is provided, the fail-closed invariants are 

291 enforced (``async_mode`` off, no ``tag``, redirects forced off) and 

292 the send runs inside the pin + block-private window it returns. The 

293 per-thread SSRF-block marker is reset first so a block during THIS 

294 ``notify`` is attributable to it (see 

295 ``dns_pinning.consume_ssrf_block``). ``guard_factory=None`` sends 

296 WITHOUT a guard and is for tests ONLY — it exercises the unguarded 

297 baseline the SSRF teeth-tests assert against; every production caller 

298 MUST pass a ``guard_factory``. 

299 

300 Args: 

301 title: Notification title 

302 body: Notification body text 

303 apprise_instance: Apprise instance to use for sending 

304 guard_factory: Optional zero-arg callable returning a context 

305 manager wrapped around the ``notify()`` call (the DNS pin + 

306 block-private window; see ``security.dns_pinning``). 

307 tag: Optional tag to target specific services 

308 attach: Optional list of file paths to attach 

309 

310 Returns: 

311 True iff Apprise reported the notification delivered. 

312 """ 

313 if guard_factory is not None: 

314 # Guarded (user-supplied URL) send: enforce the fail-closed 

315 # invariants the thread-local pin/block depend on and disable 

316 # redirects before dispatching, then clear any stale SSRF-block 

317 # marker so consume_ssrf_block reflects only this attempt. 

318 self._enforce_guarded_send_invariants(apprise_instance, tag) 

319 dns_pinning.reset_ssrf_block() 

320 

321 guard = ( 

322 guard_factory() 

323 if guard_factory is not None 

324 else contextlib.nullcontext() 

325 ) 

326 

327 # Send notification inside the guard so the pin + block-private 

328 # window (if any) governs Apprise's send-time DNS resolution. 

329 # ``notify`` is typed ``bool | None``; coerce to a definite bool 

330 # (None == nothing delivered == False). 

331 with guard: 

332 return bool( 

333 apprise_instance.notify( 

334 title=title, 

335 body=body, 

336 tag=tag, 

337 attach=attach, 

338 ) 

339 ) 

340 

341 @retry( 

342 stop=stop_after_attempt(3), 

343 wait=wait_exponential(multiplier=1, min=0.5, max=10), 

344 # Retry transient delivery failures (SendError, connection errors, 

345 # …) but NOT deliberate security refusals: a ValueError is a 

346 # connect-time SSRF block (the pin's rebind-to-private/metadata catch, 

347 # or the block-private window's send-time refusal surfaced below); a 

348 # RuntimeError (specifically 

349 # ``dns_pinning.NotificationGuardUnavailableError``) is the DNS-pin 

350 # shim not being installed; and a SecurityBlockError is the 

351 # async_mode/tag fail-closed invariant guard 

352 # (``_enforce_guarded_send_invariants``), raised directly as that 

353 # type rather than a generic RuntimeError. A confirmed-malicious or 

354 # misconfigured destination must fail fast, not be retried 3x. 

355 retry=retry_if_not_exception_type( 

356 (ValueError, RuntimeError, SecurityBlockError) 

357 ), 

358 reraise=True, 

359 ) 

360 def _send_with_retry( 

361 self, 

362 title: str, 

363 body: str, 

364 apprise_instance: apprise.Apprise, 

365 guard_factory=None, 

366 tag: Optional[str] = None, 

367 attach: Optional[List[str]] = None, 

368 ) -> bool: 

369 """ 

370 Send a notification using the provided Apprise instance with retry logic. 

371 

372 This method is decorated with Tenacity to handle retries automatically. 

373 A fresh guard context is created per attempt (via ``guard_factory``) 

374 so retries re-resolve and re-pin. 

375 

376 Args: 

377 title: Notification title 

378 body: Notification body text 

379 apprise_instance: Apprise instance to use for sending 

380 guard_factory: Optional zero-arg callable returning the per-send 

381 DNS guard context (see :meth:`_guarded_notify`). 

382 tag: Optional tag to target specific services 

383 attach: Optional list of file paths to attach 

384 

385 Returns: 

386 True if notification was sent successfully 

387 

388 Raises: 

389 SendError: If sending fails after all retry attempts (transient). 

390 ValueError: If a guarded send failed because the block-private 

391 window refused a send-time resolution as an SSRF target — 

392 a confirmed security block, raised non-retryably so it fails 

393 fast instead of being retried. 

394 SecurityBlockError: Propagated non-retryably from 

395 ``_guarded_notify`` -> ``_enforce_guarded_send_invariants`` 

396 if the async_mode/tag fail-closed invariant is violated. 

397 dns_pinning.NotificationGuardUnavailableError: Propagated 

398 non-retryably (a ``RuntimeError`` subclass) from 

399 ``_guarded_notify`` -> ``guard_factory`` if the DNS-pin 

400 shim is not installed. 

401 """ 

402 logger.debug( 

403 f"Sending notification: title='{title[:50]}...', tag={tag}" 

404 ) 

405 logger.debug(f"Body preview: {body[:200]}...") 

406 

407 notify_result = self._guarded_notify( 

408 title, body, apprise_instance, guard_factory, tag, attach 

409 ) 

410 

411 if notify_result: 

412 logger.debug(f"Notification sent successfully: '{title[:50]}...'") 

413 return True 

414 

415 # A guarded send that failed AND tripped the block-private window is a 

416 # confirmed SSRF block: the window raised socket.gaierror on a 

417 # send-time resolution to an internal/private/metadata IP, which 

418 # Apprise swallowed into a generic delivery failure. Fail fast — raise 

419 # a non-retryable ValueError (excluded from the retry predicate above) 

420 # instead of a retryable SendError, so a confirmed-malicious / 

421 # misconfigured destination is not retried 3x (~10s and 3x DNS to a 

422 # hostile authority). The pin path's own rebind catch already raises 

423 # ValueError before this point; this covers the UNPINNED block-window 

424 # path (plugin-scheme endpoints, unresolvable-at-pin-time hosts). 

425 if guard_factory is not None and dns_pinning.consume_ssrf_block(): 

426 logger.warning( 

427 "Notification send blocked by the SSRF block-private window; " 

428 "failing fast (not retrying)" 

429 ) 

430 raise ValueError( 

431 "Notification send refused: a send-time DNS resolution was " 

432 "blocked as internal/private/metadata (possible SSRF)" 

433 ) 

434 

435 error_msg = "Failed to send notification to any service" 

436 logger.warning(error_msg) 

437 raise SendError(error_msg) 

438 

439 def send( 

440 self, 

441 title: str, 

442 body: str, 

443 service_urls: Optional[str] = None, 

444 tag: Optional[str] = None, 

445 attach: Optional[List[str]] = None, 

446 separator: str = ",", 

447 ) -> bool: 

448 """ 

449 Send a notification to separator-delimited service URLs with retry. 

450 

451 Args: 

452 title: Notification title 

453 body: Notification body text 

454 service_urls: Service URLs delimited by ``separator`` 

455 tag: Optional tag to target specific services 

456 attach: Optional list of file paths to attach 

457 separator: Delimiter between service URLs 

458 

459 Returns: 

460 True if notification was sent successfully to at least one service 

461 

462 Raises: 

463 SendError: If sending fails after all retry attempts (transient 

464 — the endpoint refused/failed the dispatch). 

465 SecurityBlockError: If a send-time SSRF/DNS-rebind block 

466 confirmed the destination is internal/private/metadata 

467 (permanent — not worth retrying). A ``ServiceError`` 

468 subclass — it still satisfies ``except ServiceError`` / 

469 ``pytest.raises(ServiceError)`` and shares the same 

470 non-retryable signal as pre-dispatch validation rejects, 

471 but is tagged distinctly so callers can give it a more 

472 accurate message than "invalid URL". 

473 

474 Note: 

475 Temporary Apprise instances are created for each send operation 

476 and are automatically garbage collected by Python when they go 

477 out of scope. This simple approach is ideal for small deployments 

478 (~5 users) and avoids memory management complexity. 

479 """ 

480 # Defense-in-depth: enforce the operator-level master switch at 

481 # the service layer too, not just at NotificationManager. 

482 # Today the manager always wraps this method, but keeping the 

483 # gate here means a future direct caller cannot accidentally 

484 # bypass it. See SECURITY.md "Notification Webhook SSRF". 

485 if not self.outbound_allowed: 

486 logger.warning( 

487 "Notification not sent: outbound notifications are disabled " 

488 "at the server level. Set " 

489 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See " 

490 "SECURITY.md 'Notification Webhook SSRF'." 

491 ) 

492 return False 

493 

494 # If service_urls are provided, validate before trying to send 

495 if service_urls: 

496 logger.debug("Creating Apprise instance for provided service URLs") 

497 

498 # Validate service URLs for security (SSRF prevention) 

499 is_valid, error_msg = ( 

500 NotificationURLValidator.validate_multiple_urls( 

501 service_urls, 

502 allow_private_ips=self.allow_private_ips, 

503 separator=separator, 

504 ) 

505 ) 

506 

507 if not is_valid: 

508 logger.error("Service URL validation failed") 

509 raise ServiceError(f"Invalid service URL: {error_msg}") 

510 

511 try: 

512 # If service_urls are provided, partition them by scheme and 

513 # dispatch each group with its own SSRF send-time guard. 

514 if service_urls: 

515 strict_urls, lenient_urls = self._partition_urls( 

516 service_urls, separator 

517 ) 

518 if not strict_urls and not lenient_urls: 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true

519 logger.error("No service URLs after parsing") 

520 return False 

521 return self._dispatch( 

522 title, body, strict_urls, lenient_urls, tag, attach 

523 ) 

524 # The pre-configured instance (self.apprise) is created empty in 

525 # __init__ and is never populated by this class — all delivery 

526 # goes through the guarded per-call service_urls path above. 

527 if len(self.apprise) == 0: 527 ↛ 536line 527 didn't jump to line 536 because the condition on line 527 was always true

528 logger.debug("No notification services configured in Apprise") 

529 return False 

530 

531 # Defense-in-depth: if a future caller ever adds URLs directly to 

532 # self.apprise, refuse rather than deliver them here — this branch 

533 # has no DNS pin / block-private guard, so sending would silently 

534 # bypass the SSRF hardening the service_urls path enforces. Fail 

535 # closed. See SECURITY.md "Notification Webhook SSRF". 

536 logger.error( 

537 "Refusing to send via the pre-configured Apprise instance: " 

538 "it is populated but this path is unguarded. Pass " 

539 "service_urls to send() so the DNS pin / block-private guard " 

540 "applies." 

541 ) 

542 return False 

543 

544 except SecurityBlockError: 

545 # _enforce_guarded_send_invariants (called from _guarded_notify, 

546 # which _send_with_retry / _dispatch call) raises this DIRECTLY 

547 # — not a generic RuntimeError — for the async_mode/tag 

548 # fail-closed invariant guard: a deliberate security refusal, 

549 # not an incidental runtime error. It is already the precise, 

550 # final exception type (see its docstring), so propagate it 

551 # unchanged rather than re-wrapping: NotificationManager's 

552 # ``except SecurityBlockError`` maps it to the same 

553 # non-transient INVALID_URL reason as the confirmed-block paths 

554 # below. 

555 raise 

556 

557 except ValueError as e: 

558 # _send_with_retry raises ValueError (excluded from Tenacity's 

559 # retry predicate) ONLY for a confirmed, non-retryable send-time 

560 # SSRF/DNS-rebind block — see its docstring and 

561 # dns_pinning.pinned_notification_send's own rebind catch. That 

562 # is a permanent, invalid-destination outcome, not a transient 

563 # delivery failure, so it must NOT be re-wrapped as SendError: 

564 # NotificationManager's ``except SendError`` maps to the 

565 # retryable WEBHOOK_FAILED reason, which would mislabel a 

566 # confirmed security block as "webhook delivery failed after 

567 # retries" and could invite a caller/monitor to retry a 

568 # malicious/rebinding destination. Route it through 

569 # SecurityBlockError — a ServiceError subclass, so it still 

570 # maps to INVALID_URL via NotificationManager's existing 

571 # ``except ServiceError`` handling — but tagged distinctly so 

572 # the manager can give this confirmed security block a more 

573 # accurate user-facing detail than the pre-dispatch 

574 # "check your URL/settings" text (see manager.py). 

575 logger.exception( 

576 f"Notification send blocked (SSRF/DNS-rebind): " 

577 f"'{title[:50]}...'" 

578 ) 

579 raise SecurityBlockError(str(e)) from e 

580 

581 except dns_pinning.NotificationGuardUnavailableError as e: 

582 # pinned_notification_send (via the guard_factory _dispatch 

583 # builds) raises this — a RuntimeError subclass defined in 

584 # security.dns_pinning, deliberately NOT imported/raised as a 

585 # notifications exception there so that general-purpose 

586 # security module stays free of a dependency on this 

587 # higher-level package — when the getaddrinfo shim it depends 

588 # on is not the active resolver. Also a deliberate fail-closed 

589 # security refusal, not a transient send failure (excluded from 

590 # Tenacity's retry predicate as a RuntimeError — see 

591 # _send_with_retry's decorator), so route it through 

592 # SecurityBlockError for the same reason as the ValueError 

593 # branch above: it must not be mislabeled as a retryable 

594 # WEBHOOK_FAILED. 

595 logger.exception( 

596 f"Notification send blocked (DNS-pin guard unavailable): " 

597 f"'{title[:50]}...'" 

598 ) 

599 raise SecurityBlockError(str(e)) from e 

600 

601 except SendError: 

602 # _send_with_retry's own terminal raise (line ~420, once all 

603 # retries are exhausted) is already the precise, final 

604 # exception for a confirmed webhook delivery failure. 

605 # Propagate unchanged — re-wrapping it here would just nest a 

606 # second SendError around the first for no benefit, and 

607 # routing it through the catch-all below would make it 

608 # indistinguishable from an incidental dispatch-time bug. 

609 raise 

610 

611 except Exception: 

612 # Anything else reaching here is an incidental/unexpected 

613 # error from the dispatch machinery itself (e.g. a bug in 

614 # this method's call wiring, an Apprise internals exception 

615 # that isn't a genuine delivery failure) — NOT a confirmed 

616 # webhook delivery failure. Do NOT launder it into SendError: 

617 # NotificationManager maps SendError to the retryable 

618 # WEBHOOK_FAILED reason, which would mislabel an incidental 

619 # bug as "webhook delivery failed after retries" and would 

620 # make the manager's own ``except Exception`` -> EXCEPTION 

621 # branch unreachable for dispatch-time errors. Let it 

622 # propagate as-is so the manager's catch-all handles it. 

623 logger.exception( 

624 f"Unexpected error sending notification: '{title[:50]}...'" 

625 ) 

626 raise 

627 

628 def _dispatch( 

629 self, 

630 title: str, 

631 body: str, 

632 strict_urls: List[str], 

633 lenient_urls: List[str], 

634 tag: Optional[str], 

635 attach: Optional[List[str]], 

636 ) -> bool: 

637 """Send each scheme-partition under its SSRF send-time guard. 

638 

639 ``strict_urls`` (http/https): pin each host and block private + 

640 metadata resolution for the send (honoring the operator 

641 ``allow_private_ips`` opt-in) — closes rebinding and 

642 redirect-to-internal. Link-local stays blocked even with the 

643 opt-in: Apprise turns some http(s) URLs into plugins that post to 

644 the URL host (Apprise API, Workflows). 

645 

646 ``lenient_urls`` are classified by effective destination. Plugins 

647 whose authority selects the connection host use the operator's 

648 private-IP opt-in and a link-local block. Fixed vendor endpoints 

649 use the same operator opt-in (public-only by default): their hosts 

650 are fixed, so a private answer normally means a forward proxy 

651 (``HTTPS_PROXY``) that the operator must approve. Link-local and 

652 cloud-metadata addresses stay blocked for both. 

653 

654 Each non-empty partition is dispatched (and retried by Tenacity) 

655 under its own guard. A partition that adds no URLs returns False 

656 (matching the historical add-failure behavior); a partition that 

657 fails to deliver after all retries raises ``SendError`` (which the 

658 caller re-wraps), so overall success requires every partition to 

659 deliver. 

660 """ 

661 partitions = [] 

662 if strict_urls: 

663 partitions.append( 

664 ( 

665 strict_urls, 

666 lambda: dns_pinning.pinned_notification_send( 

667 strict_urls, 

668 allow_localhost=False, 

669 allow_private_ips=self.allow_private_ips, 

670 block_link_local=True, 

671 ), 

672 ) 

673 ) 

674 authority_urls = [] 

675 vendor_urls = [] 

676 for entry in lenient_urls: 

677 try: 

678 destination = parse_notification_destination(entry) 

679 except NotificationDestinationError as exc: 

680 raise SecurityBlockError( 

681 "Notification destination is unsupported or ambiguous" 

682 ) from exc 

683 if destination.kind is NotificationDestinationKind.AUTHORITY_PLUGIN: 

684 authority_urls.append(entry) 

685 elif destination.kind is NotificationDestinationKind.VENDOR_PLUGIN: 685 ↛ 688line 685 didn't jump to line 688 because the condition on line 685 was always true

686 vendor_urls.append(entry) 

687 else: 

688 raise SecurityBlockError("Notification partition is ambiguous") 

689 if authority_urls: 

690 partitions.append( 

691 ( 

692 authority_urls, 

693 lambda: dns_pinning.pinned_notification_send( 

694 authority_urls, 

695 allow_localhost=False, 

696 allow_private_ips=self.allow_private_ips, 

697 block_link_local=True, 

698 ), 

699 ) 

700 ) 

701 if vendor_urls: 

702 partitions.append( 

703 ( 

704 vendor_urls, 

705 lambda: dns_pinning.pinned_notification_send( 

706 vendor_urls, 

707 allow_localhost=False, 

708 allow_private_ips=self.allow_private_ips, 

709 block_link_local=True, 

710 ), 

711 ) 

712 ) 

713 

714 any_ok = False 

715 for urls, guard_factory in partitions: 

716 temp_apprise = self._new_apprise() 

717 # apprise types ``add(servers=...)`` as an invariant ``list`` whose 

718 # element union includes ``str``; a ``list[str]`` is valid at 

719 # runtime but mypy rejects it on list invariance (upstream should 

720 # use ``Sequence``). 

721 if not temp_apprise.add(urls, tag=tag): # type: ignore[arg-type] 

722 logger.error("Failed to add service URLs to Apprise") 

723 return False 

724 # _send_with_retry returns True or raises SendError after 

725 # Tenacity exhausts its retries; a raise propagates to send(). 

726 self._send_with_retry( 

727 title, body, temp_apprise, guard_factory, tag, attach 

728 ) 

729 any_ok = True 

730 return any_ok 

731 

732 def send_event( 

733 self, 

734 event_type: EventType, 

735 context: Dict[str, Any], 

736 service_urls: Optional[str] = None, 

737 tag: Optional[str] = None, 

738 custom_template: Optional[Dict[str, str]] = None, 

739 separator: str = ",", 

740 ) -> bool: 

741 """ 

742 Send a notification for a specific event type. 

743 

744 Args: 

745 event_type: Type of event 

746 context: Context data for template formatting 

747 service_urls: Service URLs delimited by ``separator`` 

748 tag: Optional tag to target specific services 

749 custom_template: Optional custom template override 

750 separator: Delimiter between service URLs 

751 

752 Returns: 

753 True if notification was sent successfully 

754 """ 

755 logger.debug(f"send_event: event_type={event_type.value}, tag={tag}") 

756 logger.debug(f"Context: {context}") 

757 

758 # Format notification using template 

759 message = NotificationTemplate.format( 

760 event_type, context, custom_template 

761 ) 

762 logger.debug( 

763 f"Template formatted - title: '{message['title'][:50]}...'" 

764 ) 

765 

766 # Send notification 

767 return self.send( 

768 title=message["title"], 

769 body=message["body"], 

770 service_urls=service_urls, 

771 tag=tag, 

772 separator=separator, 

773 ) 

774 

775 def test_service(self, url: str) -> Dict[str, Any]: 

776 """ 

777 Test a notification service. 

778 

779 Args: 

780 url: Apprise-compatible service URL 

781 

782 Returns: 

783 Dict with 'success' boolean and optional 'error' message 

784 """ 

785 # Server-level master switch (env-only). Mirrors 

786 # NotificationManager's gate so the "Send Test Notification" 

787 # button cannot bypass it. WARNING level so the operator sees 

788 # the actionable signal naming the env var. 

789 if not self.outbound_allowed: 

790 logger.warning( 

791 "Notification test refused: outbound notifications are " 

792 "disabled at the server level. Set " 

793 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See " 

794 "SECURITY.md 'Notification Webhook SSRF'." 

795 ) 

796 return { 

797 "success": False, 

798 "error": ( 

799 "Outbound notifications are disabled. The server " 

800 "administrator must set " 

801 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable " 

802 "notification webhooks. See SECURITY.md " 

803 "'Notification Webhook SSRF' for details." 

804 ), 

805 } 

806 

807 try: 

808 # ``url`` is a single form field, but Apprise's own 

809 # scheme-aware splitter treats it as a LIST — and commas are 

810 # legal URL characters. Validating ``url`` as ONE url 

811 # therefore checked only the leading entry while Apprise 

812 # still registered, and notified, every other entry: 

813 # ``discord://valid/tok,json://<cloud-metadata-host>/x`` 

814 # passed validation on the discord entry alone and the 

815 # cloud-metadata target received the test notification 

816 # unvetted. Partition the input with the shared scheme-aware 

817 # parser — the same one the configured-send path uses via 

818 # ``validate_multiple_urls`` — and vet EVERY entry. 

819 url_entries, invalid_fragment = parse_notification_url_list(url) 

820 if invalid_fragment is not None: 

821 # The input does not partition unambiguously: either a 

822 # URL-like fragment carries no scheme (Apprise may repair 

823 # it or absorb it into a neighbouring entry) or an entry 

824 # contains a character that is illegal unencoded in a URI 

825 # (backslash, whitespace, a control byte) and the token 

826 # after it is a second, unvalidated destination. REFUSE 

827 # the whole input — mirrors ``validate_multiple_urls``, 

828 # which returns False for every fragment shape. 

829 # 

830 # Substituting ``url_entries = [invalid_fragment]`` here 

831 # (as this code did before) was exploitable: the 

832 # fragment is NOT necessarily scheme-less, because 

833 # ``_URL_BOUNDARY_RE`` splits only on ``[,\s]+`` while 

834 # the fragment is cut on ``RFC_FORBIDDEN_URL_CHARS_RE``, 

835 # which also covers backslash and non-``\s`` control 

836 # bytes. So ``json://<metadata-host>/x\https://vendor/ok`` 

837 # stays ONE entry whose fragment is the *vendor* URL: 

838 # the loop below would validate only that decoy, the 

839 # smuggle check would compare ``1 > 1``, and the 

840 # then-current ``temp_apprise.add(url)`` would dispatch 

841 # the RAW original string to the metadata host — with the 

842 # pin/block-private policy derived from the decoy's 

843 # scheme. (``add()`` now takes the parsed entry list, so 

844 # the raw string is no longer in the dispatch path at 

845 # all; this refusal is still the primary defence, since 

846 # the decoy would otherwise be the only entry vetted.) 

847 # Never validate a fragment in place of the input it was 

848 # carved out of. 

849 # 

850 # The message is deliberately generic: the fragment's 

851 # own content is attacker-shaped AND may carry the 

852 # operator's real credentials, so nothing derived from 

853 # it is echoed to the user or the log. 

854 logger.warning( 

855 "Test notification refused: the service URL does not " 

856 "partition into unambiguous entries " 

857 "({} entries parsed, {}-character trailing fragment).", 

858 len(url_entries), 

859 len(invalid_fragment), 

860 ) 

861 return { 

862 "success": False, 

863 "error": ( 

864 "Notification service URL could not be parsed " 

865 "unambiguously; refusing to send. Every entry must " 

866 "begin with a protocol such as discord:// — remove " 

867 "any spaces, backslashes or control characters and " 

868 "configure one complete service URL per test." 

869 ), 

870 } 

871 if not url_entries: 871 ↛ 872line 871 didn't jump to line 872 because the condition on line 871 was never true

872 return { 

873 "success": False, 

874 "error": "Invalid notification service URL.", 

875 } 

876 if len(url_entries) > MAX_NOTIFICATION_TARGETS: 

877 logger.warning( 

878 f"Test notification refused: {len(url_entries)} targets " 

879 f"exceeds the cap of {MAX_NOTIFICATION_TARGETS}." 

880 ) 

881 return { 

882 "success": False, 

883 "error": ( 

884 f"Too many notification targets " 

885 f"({len(url_entries)}). Maximum is " 

886 f"{MAX_NOTIFICATION_TARGETS}." 

887 ), 

888 } 

889 

890 for entry in url_entries: 

891 # Validate each entry for security (SSRF prevention) and, 

892 # in the same pass, compute whether the admin env-var hint 

893 # would actually unblock a recoverable private-IP 

894 # rejection. Single-pass avoids a DNS-rebinding TOCTOU 

895 # window between the default-level validation and the 

896 # elevated-level hint decision — see 

897 # NotificationURLValidator.validate_service_url_with_hint. 

898 is_valid, error_msg, hint_would_help = ( 

899 NotificationURLValidator.validate_service_url_with_hint( 

900 entry, allow_private_ips=self.allow_private_ips 

901 ) 

902 ) 

903 

904 if not is_valid: 

905 # Never log the entry itself — an Apprise service URL 

906 # carries credentials (bot tokens, webhook secrets). 

907 # #5576 removed the masked-URL echo from this line for 

908 # exactly that reason; keep it out of the per-entry 

909 # loop too. 

910 logger.warning( 

911 f"Test service URL validation failed: {error_msg}" 

912 ) 

913 # Surface the validator's reason so users know what to 

914 # fix. The hostname/scheme echoed here was supplied by 

915 # the user in the same request, so this is not a 

916 # server-side leak. When the rejection is a recoverable 

917 # private/internal IP — one that 

918 # LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true OR 

919 # LDR_SECURITY_ALLOW_NAT64=true would actually unblock — 

920 # append the admin escape hatches; the user cannot fix 

921 # this themselves. Suppress the hint for always-blocked 

922 # categories (metadata, 6to4, Teredo, discard, 

923 # IPv4-mapped IPv6 of metadata, IPv4-Compatible IPv6, 

924 # IPv4-Translated/SIIT, NAT64-wrapped metadata): no env 

925 # var can help, so naming one would mislead. 

926 user_error = ( 

927 error_msg or "Invalid notification service URL." 

928 ) 

929 if hint_would_help: 

930 user_error += ( 

931 ". To unblock this destination, ask the server " 

932 "administrator to set " 

933 "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true " 

934 "(IPv6-only NAT64 deployments also need " 

935 "LDR_SECURITY_ALLOW_NAT64=true). " 

936 "See SECURITY.md 'Notification Webhook SSRF'." 

937 ) 

938 return { 

939 "success": False, 

940 "error": user_error, 

941 } 

942 

943 # Create temporary Apprise instance (synchronous so the pin + 

944 # block-private window below applies in this thread). 

945 temp_apprise = self._new_apprise() 

946 # Hand Apprise the ALREADY-PARTITIONED entries, not the raw 

947 # string — the same thing :meth:`_dispatch` does. ``add()`` 

948 # runs its own ``parse_urls`` splitter ONLY on a ``str`` 

949 # argument; given a list it instantiates exactly one plugin 

950 # per element. Passing ``url`` here let LDR's boundary regex 

951 # and Apprise's ``URL_DETECTION_RE`` each get a vote on where 

952 # the entries start, and the two do not agree: Apprise's 

953 # scheme class is ``[a-z0-9]`` (a LEADING DIGIT starts a new 

954 # entry, ``+ . -`` do not) while LDR's requires a leading 

955 # letter and accepts ``+ . -``. So ``discord://a/b,7z://x`` 

956 # is ONE entry to LDR (validated as a discord URL) and TWO to 

957 # Apprise — the second never seen by the SSRF validator. Only 

958 # the count guard below caught that, and only because every 

959 # scheme in ``ALLOWED_SCHEMES`` happens to be pure lowercase 

960 # alpha, which is a property of a constant rather than of 

961 # this code. With a list there is ONE parser in the path and 

962 # the set Apprise dispatches is exactly the set validated 

963 # above. 

964 add_result = temp_apprise.add(list(url_entries)) 

965 

966 if not add_result: 

967 return { 

968 "success": False, 

969 "error": "Failed to add service URL", 

970 } 

971 

972 # Fail closed on a parser differential: if Apprise registered 

973 # MORE targets than we validated, at least one destination 

974 # would be notified without ever passing the SSRF validator. 

975 # Fewer is harmless (Apprise dropped an entry), so only the 

976 # smuggling direction is refused. 

977 # 

978 # With the list-form ``add()`` above this guard is STRUCTURAL, 

979 # not a live check: ``Apprise.__len__`` counts one per 

980 # instantiated plugin and the list form appends AT MOST ONE 

981 # plugin per element, so ``len(temp_apprise) > len(url_entries)`` 

982 # cannot be true for any input on the current Apprise. It is 

983 # kept as a cheap invariant assertion — it still catches a 

984 # future Apprise release that expands one URL into several 

985 # servers, and it is the last thing standing if the ``add()`` 

986 # argument is ever changed back to a string. The argument 

987 # shape itself is pinned by 

988 # ``test_test_service_success`` in ``tests/notifications/ 

989 # test_service.py``, which asserts ``add`` receives the parsed 

990 # LIST. 

991 if len(temp_apprise) > len(url_entries): 

992 logger.warning( 

993 f"Test notification refused: Apprise registered " 

994 f"{len(temp_apprise)} targets but only " 

995 f"{len(url_entries)} were validated." 

996 ) 

997 return { 

998 "success": False, 

999 "error": ( 

1000 "Notification service URL could not be parsed " 

1001 "unambiguously; refusing to send. Configure one " 

1002 "service URL per test." 

1003 ), 

1004 } 

1005 

1006 # Send the test through the SAME guarded dispatch core as the real 

1007 # send path (:meth:`_guarded_notify`) — fail-closed invariants, 

1008 # per-plugin redirect-disable, and the pin + block-private window 

1009 # — instead of re-implementing it here. Single-shot (no Tenacity 

1010 # retry) for fast test feedback. Every partition uses the 

1011 # operator's private-IP opt-in (public-only by default) and 

1012 # blocks the whole link-local range, as ``_dispatch`` does. 

1013 # Tag is always None here. 

1014 try: 

1015 for entry in url_entries: 

1016 parse_notification_destination(entry) 

1017 except NotificationDestinationError: 

1018 return { 

1019 "success": False, 

1020 "error": "Notification destination is unsupported or ambiguous", 

1021 } 

1022 guard_factory = functools.partial( 

1023 dns_pinning.pinned_notification_send, 

1024 list(url_entries), 

1025 allow_localhost=False, 

1026 allow_private_ips=self.allow_private_ips, 

1027 block_link_local=True, 

1028 ) 

1029 

1030 result = self._guarded_notify( 

1031 "Test Notification", 

1032 ( 

1033 "This is a test notification from Local Deep Research. " 

1034 "If you see this, your service is configured correctly!" 

1035 ), 

1036 temp_apprise, 

1037 guard_factory, 

1038 tag=None, 

1039 ) 

1040 

1041 if result: 

1042 return { 

1043 "success": True, 

1044 "message": "Test notification sent successfully", 

1045 } 

1046 

1047 # A guarded test send that failed AND tripped the block-private 

1048 # window is a confirmed SSRF block: the window raised 

1049 # socket.gaierror on a send-time resolution to an 

1050 # internal/private/metadata IP, which Apprise swallowed into a 

1051 # generic delivery failure. Surface the specific reason to the 

1052 # admin instead of the generic message — mirrors the 

1053 # consume_ssrf_block short-circuit in _send_with_retry. (The pin 

1054 # path's own connect-time rebind catch raises ValueError before 

1055 # this point and is reported via the except below; this covers the 

1056 # UNPINNED block-window path — plugin-scheme endpoints and 

1057 # unresolvable-at-pin-time hosts.) 

1058 if dns_pinning.consume_ssrf_block(): 

1059 logger.warning( 

1060 "Test notification blocked by the SSRF block-private " 

1061 "window (send-time resolution to internal/private/" 

1062 "metadata)" 

1063 ) 

1064 return { 

1065 "success": False, 

1066 "error": ( 

1067 "Test notification refused: the destination resolved " 

1068 "to an internal/private/metadata address at send time " 

1069 "(possible SSRF). See SECURITY.md 'Notification " 

1070 "Webhook SSRF'." 

1071 ), 

1072 } 

1073 return { 

1074 "success": False, 

1075 "error": "Failed to send test notification", 

1076 } 

1077 

1078 except Exception: 

1079 logger.exception("Error testing notification service") 

1080 return { 

1081 "success": False, 

1082 "error": "Failed to test notification service.", 

1083 } 

1084 

1085 @staticmethod 

1086 def _validate_url(url: str) -> None: 

1087 """ 

1088 Validate a notification service URL. 

1089 

1090 Args: 

1091 url: URL to validate