Coverage for src/local_deep_research/notifications/service.py: 91%
191 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Core notification service using Apprise.
3"""
5import contextlib
6import functools
7import re
8from typing import Dict, List, Optional, Any, Tuple
9from urllib.parse import urlparse
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)
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_validator import (
25 NotificationURLValidator,
26 parse_notification_url_list,
27)
29PRIVATE_IP_REJECTION_PREFIX = (
30 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX
31)
34# Backward compatibility constants - now handled by Tenacity internally
35MAX_RETRY_ATTEMPTS = 3
36INITIAL_RETRY_DELAY = 0.5
37RETRY_BACKOFF_MULTIPLIER = 2
39# Maximum number of notification targets Apprise may parse out of one
40# test/configured URL string. Prevents an authenticated caller from
41# triggering unbounded DNS resolution work (and an unbounded fan-out of
42# outbound requests) via a comma/space-separated hostname list.
43MAX_NOTIFICATION_TARGETS = 20
46class NotificationService:
47 """
48 Low-level notification service that wraps Apprise.
49 """
51 # Regex patterns for common service types (for validation)
52 SERVICE_PATTERNS = {
53 "email": r"^mailto://",
54 "discord": r"^discord://",
55 "slack": r"^slack://",
56 "telegram": r"^tgram://",
57 "smtp": r"^(smtp|smtps)://",
58 }
60 def __init__(
61 self,
62 allow_private_ips: bool = False,
63 outbound_allowed: bool = False,
64 ):
65 """
66 Initialize the notification service.
68 Args:
69 allow_private_ips: Whether to allow notifications to private/local IPs
70 (default: False for security). Set to True for
71 development/testing environments only.
72 outbound_allowed: Server-level master switch
73 (env-only via LDR_NOTIFICATIONS_ALLOW_OUTBOUND).
74 Default False — outbound notifications are off until
75 the operator opts in. See SECURITY.md "Notification
76 Webhook SSRF" for the rationale.
77 """
78 install_apprise_log_record_factory()
79 self.apprise = self._new_apprise()
80 self.allow_private_ips = allow_private_ips
81 self.outbound_allowed = outbound_allowed
83 @staticmethod
84 def _new_apprise() -> apprise.Apprise:
85 """Create an Apprise instance that delivers synchronously and does
86 not follow HTTP redirects.
88 ``async_mode=False`` forces Apprise to run every plugin's
89 ``notify()`` in the calling thread instead of fanning out to a
90 ``ThreadPoolExecutor``. This is required for the SSRF hardening in
91 :meth:`_send_with_retry`: the DNS pin and the block-private window
92 (``security.dns_pinning``) are thread-local, so they only apply to
93 the resolution the sending thread performs. With the default async
94 fan-out the plugins would resolve in worker threads that carry no
95 pin/block and the guard would silently not apply (verified: the
96 default asset resolves target hosts on worker threads).
98 ``http_redirects=False`` disables 30x redirect-following for the
99 raw-webhook plugins (``json``/``xml``/``form``, which pass
100 ``allow_redirects=self.redirects`` to ``requests``). This closes the
101 redirect attack class outright: a user-configured webhook can no
102 longer 30x-redirect the send to a private/loopback host or to an
103 arbitrary public host for data exfiltration. The pin +
104 block-private window is then only responsible for the original
105 validated host's own rebind, not for chasing redirect hops. It is a
106 DEFAULT: a per-URL ``?redirect=yes`` can still override it, so
107 :meth:`_disable_redirects` re-forces it off per-plugin after
108 ``add()`` (see there).
109 """
110 return apprise.Apprise(
111 asset=apprise.AppriseAsset(async_mode=False, http_redirects=False)
112 )
114 @staticmethod
115 def _disable_redirects(apprise_instance: apprise.Apprise) -> None:
116 """Force redirect-following OFF on every added plugin.
118 ``_new_apprise`` sets the asset-level ``http_redirects=False``
119 default, but Apprise lets a per-URL ``?redirect=yes`` query
120 parameter override that default (``self.redirects =
121 parse_bool(kwargs.get("redirect", asset.http_redirects))``). A
122 low-privilege user could therefore re-enable redirect-following on
123 their own webhook and reopen redirect-to-private / redirect-to-exfil.
124 Setting ``redirects = False`` directly on each plugin AFTER it is
125 parsed makes the closure airtight: the value is read at send time
126 (``allow_redirects=self.redirects``), so this wins over any URL
127 parameter. Redirect-following is never needed for legitimate
128 webhook/plugin delivery (real endpoints answer 2xx directly), so
129 this cannot break a genuine send.
131 Not every plugin threads ``self.redirects`` into its HTTP call: the
132 raw-webhook ``json``/``xml``/``form`` schemes and e.g. ``ntfy`` /
133 ``gotify`` honour it, but a few (e.g. ``matrix://``) issue requests
134 without it, so re-forcing the flag has no effect there. Those are
135 backstopped by the block-private send window
136 (``dns_pinning.pinned_notification_send``), which refuses a
137 redirect/rebind to a blocked address at the ``getaddrinfo`` layer
138 regardless of whether the plugin followed the redirect.
139 """
140 try:
141 servers = list(apprise_instance)
142 except TypeError as exc:
143 # A mocked Apprise in unit tests is not iterable, so there is
144 # nothing to harden (those tests never exercise a real redirect).
145 # But if a REAL Apprise ever stops being iterable (upstream API
146 # drift), silently returning here would leave the per-plugin
147 # redirect-disable un-applied — fail-open. The asset-level
148 # ``http_redirects=False`` default and the block-private window
149 # still apply, but a per-URL ``?redirect=yes`` could re-enable
150 # redirect-following. Log a warning (narrowed to TypeError) so
151 # that drift is visible instead of silent.
152 logger.warning(
153 "Could not enumerate Apprise servers to force "
154 "redirect-following off per plugin ({}); relying on the "
155 "asset-level default and the block-private window",
156 type(exc).__name__,
157 )
158 return
159 for server in servers:
160 server.redirects = False
162 def _enforce_guarded_send_invariants(
163 self, apprise_instance: apprise.Apprise, tag: Optional[str]
164 ) -> None:
165 """Enforce the fail-closed invariants for a DNS-guarded send.
167 The pin + block-private window (``security.dns_pinning``) are
168 thread-local and only cover IN-THREAD delivery. Refuse the send
169 (fail closed) rather than silently sending unguarded unless:
171 * ``async_mode`` is off — otherwise Apprise fans the plugins out to
172 worker threads that carry neither the pin nor the block; and
173 * ``tag`` is None — a multi-token tag likewise makes Apprise
174 dispatch matching plugins across worker threads. No production
175 caller passes a tag; this neutralizes a future one.
177 Then force redirect-following off per plugin so a per-URL
178 ``?redirect=yes`` cannot override the asset-level default.
180 Raises:
181 SecurityBlockError: if either invariant is violated — a
182 deliberate fail-closed security refusal, not an incidental
183 runtime error, so it is raised directly as this type
184 (rather than a generic ``RuntimeError`` a caller would have
185 to re-classify) and is excluded from Tenacity retries so it
186 fails fast. See ``send()``'s docstring for why this must
187 not be mistaken for a transient delivery failure.
188 """
189 if getattr(apprise_instance.asset, "async_mode", True) is True:
190 raise SecurityBlockError(
191 "Guarded notification send requires async_mode=False; "
192 "refusing to send (the thread-local DNS pin/block would not "
193 "apply to Apprise's worker-thread fan-out)."
194 )
195 if tag is not None:
196 raise SecurityBlockError(
197 "Guarded notification send does not support tag targeting; "
198 "refusing to send (a tag can fan delivery out to worker "
199 "threads that bypass the thread-local DNS pin/block)."
200 )
201 self._disable_redirects(apprise_instance)
203 @staticmethod
204 def _partition_urls(service_urls: str) -> Tuple[List[str], List[str]]:
205 """Split an Apprise URL string into (http(s), everything-else).
207 Partitioning uses the notification validator's scheme-boundary
208 split (``parse_notification_url_list``): a comma/whitespace
209 sequence only separates entries when a URL scheme follows it, so
210 commas INSIDE one Apprise URL (e.g. a multi-target Telegram
211 ``?to=id1,id2``) are preserved and dispatched as a single entry —
212 the same partition the validator applies before the send. The two
213 groups get different send-time SSRF policies (see
214 :meth:`_dispatch`), mirroring the notification validator's
215 per-scheme rules: ``http``/``https`` block private IPs
216 unless the operator opted in, while plugin / raw-webhook schemes
217 allow private (self-hosted LAN) but always block cloud-metadata.
219 NOTE: the ``strict`` (``http``/``https``) partition is UNREACHABLE
220 for delivery today — Apprise rejects bare ``http(s)://`` notification
221 URLs at ``add()`` (verified: ``apprise.Apprise().add("http://x/")``
222 is False), so a strict-partition ``add()`` in :meth:`_dispatch`
223 returns False and nothing is sent. The stricter policy is retained
224 as defense-in-depth for a hypothetical future where a raw ``http(s)``
225 scheme becomes deliverable; deliverable webhooks use the raw-webhook
226 schemes ``json``/``xml``/``form`` (in the ``lenient`` partition).
227 """
228 strict: List[str] = []
229 lenient: List[str] = []
230 entries, invalid_fragment = parse_notification_url_list(
231 service_urls, ","
232 )
233 if invalid_fragment is not None: 233 ↛ 240line 233 didn't jump to line 240 because the condition on line 233 was never true
234 # Never log the fragment itself — it may contain credentials.
235 # ``send()`` runs ``validate_multiple_urls`` (same
236 # scheme-boundary parse) before reaching here, so a malformed
237 # fragment is already rejected there; this branch only fires
238 # for direct callers of this helper, which must validate the
239 # list themselves before dispatching it.
240 logger.warning(
241 "Notification service_url list contains a malformed "
242 "fragment; callers must validate the list before dispatch"
243 )
244 for entry in entries:
245 entry = entry.strip()
246 if not entry: 246 ↛ 247line 246 didn't jump to line 247 because the condition on line 246 was never true
247 continue
248 scheme = entry.split(":", 1)[0].lower() if ":" in entry else ""
249 if scheme in ("http", "https"):
250 strict.append(entry)
251 else:
252 lenient.append(entry)
253 return strict, lenient
255 def _guarded_notify(
256 self,
257 title: str,
258 body: str,
259 apprise_instance: apprise.Apprise,
260 guard_factory=None,
261 tag: Optional[str] = None,
262 attach: Optional[List[str]] = None,
263 ) -> bool:
264 """Run a single Apprise ``notify()`` under the DNS-guard and the
265 fail-closed invariants. Returns Apprise's raw boolean result.
267 Single source of truth for the guarded send: both the retrying
268 delivery path (:meth:`_send_with_retry`) and the admin
269 "Send Test Notification" path (:meth:`test_service`) call this, so
270 neither re-implements the invariant-enforcement / redirect-disable /
271 guarded-``notify`` sequence that the SSRF hardening depends on.
273 When ``guard_factory`` is provided, the fail-closed invariants are
274 enforced (``async_mode`` off, no ``tag``, redirects forced off) and
275 the send runs inside the pin + block-private window it returns. The
276 per-thread SSRF-block marker is reset first so a block during THIS
277 ``notify`` is attributable to it (see
278 ``dns_pinning.consume_ssrf_block``). ``guard_factory=None`` sends
279 WITHOUT a guard and is for tests ONLY — it exercises the unguarded
280 baseline the SSRF teeth-tests assert against; every production caller
281 MUST pass a ``guard_factory``.
283 Args:
284 title: Notification title
285 body: Notification body text
286 apprise_instance: Apprise instance to use for sending
287 guard_factory: Optional zero-arg callable returning a context
288 manager wrapped around the ``notify()`` call (the DNS pin +
289 block-private window; see ``security.dns_pinning``).
290 tag: Optional tag to target specific services
291 attach: Optional list of file paths to attach
293 Returns:
294 True iff Apprise reported the notification delivered.
295 """
296 if guard_factory is not None:
297 # Guarded (user-supplied URL) send: enforce the fail-closed
298 # invariants the thread-local pin/block depend on and disable
299 # redirects before dispatching, then clear any stale SSRF-block
300 # marker so consume_ssrf_block reflects only this attempt.
301 self._enforce_guarded_send_invariants(apprise_instance, tag)
302 dns_pinning.reset_ssrf_block()
304 guard = (
305 guard_factory()
306 if guard_factory is not None
307 else contextlib.nullcontext()
308 )
310 # Send notification inside the guard so the pin + block-private
311 # window (if any) governs Apprise's send-time DNS resolution.
312 # ``notify`` is typed ``bool | None``; coerce to a definite bool
313 # (None == nothing delivered == False).
314 with guard:
315 return bool(
316 apprise_instance.notify(
317 title=title,
318 body=body,
319 tag=tag,
320 attach=attach,
321 )
322 )
324 @retry(
325 stop=stop_after_attempt(3),
326 wait=wait_exponential(multiplier=1, min=0.5, max=10),
327 # Retry transient delivery failures (SendError, connection errors,
328 # …) but NOT deliberate security refusals: a ValueError is a
329 # connect-time SSRF block (the pin's rebind-to-private/metadata catch,
330 # or the block-private window's send-time refusal surfaced below); a
331 # RuntimeError (specifically
332 # ``dns_pinning.NotificationGuardUnavailableError``) is the DNS-pin
333 # shim not being installed; and a SecurityBlockError is the
334 # async_mode/tag fail-closed invariant guard
335 # (``_enforce_guarded_send_invariants``), raised directly as that
336 # type rather than a generic RuntimeError. A confirmed-malicious or
337 # misconfigured destination must fail fast, not be retried 3x.
338 retry=retry_if_not_exception_type(
339 (ValueError, RuntimeError, SecurityBlockError)
340 ),
341 reraise=True,
342 )
343 def _send_with_retry(
344 self,
345 title: str,
346 body: str,
347 apprise_instance: apprise.Apprise,
348 guard_factory=None,
349 tag: Optional[str] = None,
350 attach: Optional[List[str]] = None,
351 ) -> bool:
352 """
353 Send a notification using the provided Apprise instance with retry logic.
355 This method is decorated with Tenacity to handle retries automatically.
356 A fresh guard context is created per attempt (via ``guard_factory``)
357 so retries re-resolve and re-pin.
359 Args:
360 title: Notification title
361 body: Notification body text
362 apprise_instance: Apprise instance to use for sending
363 guard_factory: Optional zero-arg callable returning the per-send
364 DNS guard context (see :meth:`_guarded_notify`).
365 tag: Optional tag to target specific services
366 attach: Optional list of file paths to attach
368 Returns:
369 True if notification was sent successfully
371 Raises:
372 SendError: If sending fails after all retry attempts (transient).
373 ValueError: If a guarded send failed because the block-private
374 window refused a send-time resolution as an SSRF target —
375 a confirmed security block, raised non-retryably so it fails
376 fast instead of being retried.
377 SecurityBlockError: Propagated non-retryably from
378 ``_guarded_notify`` -> ``_enforce_guarded_send_invariants``
379 if the async_mode/tag fail-closed invariant is violated.
380 dns_pinning.NotificationGuardUnavailableError: Propagated
381 non-retryably (a ``RuntimeError`` subclass) from
382 ``_guarded_notify`` -> ``guard_factory`` if the DNS-pin
383 shim is not installed.
384 """
385 logger.debug(
386 f"Sending notification: title='{title[:50]}...', tag={tag}"
387 )
388 logger.debug(f"Body preview: {body[:200]}...")
390 notify_result = self._guarded_notify(
391 title, body, apprise_instance, guard_factory, tag, attach
392 )
394 if notify_result:
395 logger.debug(f"Notification sent successfully: '{title[:50]}...'")
396 return True
398 # A guarded send that failed AND tripped the block-private window is a
399 # confirmed SSRF block: the window raised socket.gaierror on a
400 # send-time resolution to an internal/private/metadata IP, which
401 # Apprise swallowed into a generic delivery failure. Fail fast — raise
402 # a non-retryable ValueError (excluded from the retry predicate above)
403 # instead of a retryable SendError, so a confirmed-malicious /
404 # misconfigured destination is not retried 3x (~10s and 3x DNS to a
405 # hostile authority). The pin path's own rebind catch already raises
406 # ValueError before this point; this covers the UNPINNED block-window
407 # path (plugin-scheme endpoints, unresolvable-at-pin-time hosts).
408 if guard_factory is not None and dns_pinning.consume_ssrf_block():
409 logger.warning(
410 "Notification send blocked by the SSRF block-private window; "
411 "failing fast (not retrying)"
412 )
413 raise ValueError(
414 "Notification send refused: a send-time DNS resolution was "
415 "blocked as internal/private/metadata (possible SSRF)"
416 )
418 error_msg = "Failed to send notification to any service"
419 logger.warning(error_msg)
420 raise SendError(error_msg)
422 def send(
423 self,
424 title: str,
425 body: str,
426 service_urls: Optional[str] = None,
427 tag: Optional[str] = None,
428 attach: Optional[List[str]] = None,
429 ) -> bool:
430 """
431 Send a notification to service URLs with automatic retry.
433 Args:
434 title: Notification title
435 body: Notification body text
436 service_urls: Comma-separated list of service URLs to override configured ones
437 tag: Optional tag to target specific services
438 attach: Optional list of file paths to attach
440 Returns:
441 True if notification was sent successfully to at least one service
443 Raises:
444 SendError: If sending fails after all retry attempts (transient
445 — the endpoint refused/failed the dispatch).
446 SecurityBlockError: If a send-time SSRF/DNS-rebind block
447 confirmed the destination is internal/private/metadata
448 (permanent — not worth retrying). A ``ServiceError``
449 subclass — it still satisfies ``except ServiceError`` /
450 ``pytest.raises(ServiceError)`` and shares the same
451 non-retryable signal as pre-dispatch validation rejects,
452 but is tagged distinctly so callers can give it a more
453 accurate message than "invalid URL".
455 Note:
456 Temporary Apprise instances are created for each send operation
457 and are automatically garbage collected by Python when they go
458 out of scope. This simple approach is ideal for small deployments
459 (~5 users) and avoids memory management complexity.
460 """
461 # Defense-in-depth: enforce the operator-level master switch at
462 # the service layer too, not just at NotificationManager.
463 # Today the manager always wraps this method, but keeping the
464 # gate here means a future direct caller cannot accidentally
465 # bypass it. See SECURITY.md "Notification Webhook SSRF".
466 if not self.outbound_allowed:
467 logger.warning(
468 "Notification not sent: outbound notifications are disabled "
469 "at the server level. Set "
470 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See "
471 "SECURITY.md 'Notification Webhook SSRF'."
472 )
473 return False
475 # If service_urls are provided, validate before trying to send
476 if service_urls:
477 logger.debug("Creating Apprise instance for provided service URLs")
479 # Validate service URLs for security (SSRF prevention)
480 is_valid, error_msg = (
481 NotificationURLValidator.validate_multiple_urls(
482 service_urls, allow_private_ips=self.allow_private_ips
483 )
484 )
486 if not is_valid:
487 logger.error("Service URL validation failed")
488 raise ServiceError(f"Invalid service URL: {error_msg}")
490 try:
491 # If service_urls are provided, partition them by scheme and
492 # dispatch each group with its own SSRF send-time guard.
493 if service_urls:
494 strict_urls, lenient_urls = self._partition_urls(service_urls)
495 if not strict_urls and not lenient_urls: 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 logger.error("No service URLs after parsing")
497 return False
498 return self._dispatch(
499 title, body, strict_urls, lenient_urls, tag, attach
500 )
501 # The pre-configured instance (self.apprise) is created empty in
502 # __init__ and is never populated by this class — all delivery
503 # goes through the guarded per-call service_urls path above.
504 if len(self.apprise) == 0: 504 ↛ 513line 504 didn't jump to line 513 because the condition on line 504 was always true
505 logger.debug("No notification services configured in Apprise")
506 return False
508 # Defense-in-depth: if a future caller ever adds URLs directly to
509 # self.apprise, refuse rather than deliver them here — this branch
510 # has no DNS pin / block-private guard, so sending would silently
511 # bypass the SSRF hardening the service_urls path enforces. Fail
512 # closed. See SECURITY.md "Notification Webhook SSRF".
513 logger.error(
514 "Refusing to send via the pre-configured Apprise instance: "
515 "it is populated but this path is unguarded. Pass "
516 "service_urls to send() so the DNS pin / block-private guard "
517 "applies."
518 )
519 return False
521 except SecurityBlockError:
522 # _enforce_guarded_send_invariants (called from _guarded_notify,
523 # which _send_with_retry / _dispatch call) raises this DIRECTLY
524 # — not a generic RuntimeError — for the async_mode/tag
525 # fail-closed invariant guard: a deliberate security refusal,
526 # not an incidental runtime error. It is already the precise,
527 # final exception type (see its docstring), so propagate it
528 # unchanged rather than re-wrapping: NotificationManager's
529 # ``except SecurityBlockError`` maps it to the same
530 # non-transient INVALID_URL reason as the confirmed-block paths
531 # below.
532 raise
534 except ValueError as e:
535 # _send_with_retry raises ValueError (excluded from Tenacity's
536 # retry predicate) ONLY for a confirmed, non-retryable send-time
537 # SSRF/DNS-rebind block — see its docstring and
538 # dns_pinning.pinned_notification_send's own rebind catch. That
539 # is a permanent, invalid-destination outcome, not a transient
540 # delivery failure, so it must NOT be re-wrapped as SendError:
541 # NotificationManager's ``except SendError`` maps to the
542 # retryable WEBHOOK_FAILED reason, which would mislabel a
543 # confirmed security block as "webhook delivery failed after
544 # retries" and could invite a caller/monitor to retry a
545 # malicious/rebinding destination. Route it through
546 # SecurityBlockError — a ServiceError subclass, so it still
547 # maps to INVALID_URL via NotificationManager's existing
548 # ``except ServiceError`` handling — but tagged distinctly so
549 # the manager can give this confirmed security block a more
550 # accurate user-facing detail than the pre-dispatch
551 # "check your URL/settings" text (see manager.py).
552 logger.exception(
553 f"Notification send blocked (SSRF/DNS-rebind): "
554 f"'{title[:50]}...'"
555 )
556 raise SecurityBlockError(str(e)) from e
558 except dns_pinning.NotificationGuardUnavailableError as e:
559 # pinned_notification_send (via the guard_factory _dispatch
560 # builds) raises this — a RuntimeError subclass defined in
561 # security.dns_pinning, deliberately NOT imported/raised as a
562 # notifications exception there so that general-purpose
563 # security module stays free of a dependency on this
564 # higher-level package — when the getaddrinfo shim it depends
565 # on is not the active resolver. Also a deliberate fail-closed
566 # security refusal, not a transient send failure (excluded from
567 # Tenacity's retry predicate as a RuntimeError — see
568 # _send_with_retry's decorator), so route it through
569 # SecurityBlockError for the same reason as the ValueError
570 # branch above: it must not be mislabeled as a retryable
571 # WEBHOOK_FAILED.
572 logger.exception(
573 f"Notification send blocked (DNS-pin guard unavailable): "
574 f"'{title[:50]}...'"
575 )
576 raise SecurityBlockError(str(e)) from e
578 except SendError:
579 # _send_with_retry's own terminal raise (line ~420, once all
580 # retries are exhausted) is already the precise, final
581 # exception for a confirmed webhook delivery failure.
582 # Propagate unchanged — re-wrapping it here would just nest a
583 # second SendError around the first for no benefit, and
584 # routing it through the catch-all below would make it
585 # indistinguishable from an incidental dispatch-time bug.
586 raise
588 except Exception:
589 # Anything else reaching here is an incidental/unexpected
590 # error from the dispatch machinery itself (e.g. a bug in
591 # this method's call wiring, an Apprise internals exception
592 # that isn't a genuine delivery failure) — NOT a confirmed
593 # webhook delivery failure. Do NOT launder it into SendError:
594 # NotificationManager maps SendError to the retryable
595 # WEBHOOK_FAILED reason, which would mislabel an incidental
596 # bug as "webhook delivery failed after retries" and would
597 # make the manager's own ``except Exception`` -> EXCEPTION
598 # branch unreachable for dispatch-time errors. Let it
599 # propagate as-is so the manager's catch-all handles it.
600 logger.exception(
601 f"Unexpected error sending notification: '{title[:50]}...'"
602 )
603 raise
605 def _dispatch(
606 self,
607 title: str,
608 body: str,
609 strict_urls: List[str],
610 lenient_urls: List[str],
611 tag: Optional[str],
612 attach: Optional[List[str]],
613 ) -> bool:
614 """Send each scheme-partition under its SSRF send-time guard.
616 ``strict_urls`` (http/https): pin each host and block private +
617 metadata resolution for the send (honoring the operator
618 ``allow_private_ips`` opt-in) — closes rebinding and
619 redirect-to-internal.
621 ``lenient_urls`` (plugin / raw-webhook schemes): pin the raw-webhook
622 hosts (json/xml/form) and block cloud-metadata AND the whole
623 link-local range (``allow_private_ips=True`` + ``block_link_local=True``)
624 — self-hosted LAN targets (RFC1918 / loopback / non-link-local ULA)
625 keep working while a send-time rebind/redirect to a metadata or
626 link-local IP is refused. This mirrors the notification validator's
627 per-scheme policy exactly.
629 Each non-empty partition is dispatched (and retried by Tenacity)
630 under its own guard. A partition that adds no URLs returns False
631 (matching the historical add-failure behavior); a partition that
632 fails to deliver after all retries raises ``SendError`` (which the
633 caller re-wraps), so overall success requires every partition to
634 deliver.
635 """
636 partitions = []
637 if strict_urls: 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 partitions.append(
639 (
640 strict_urls,
641 lambda: dns_pinning.pinned_notification_send(
642 strict_urls,
643 allow_localhost=False,
644 allow_private_ips=self.allow_private_ips,
645 ),
646 )
647 )
648 if lenient_urls: 648 ↛ 667line 648 didn't jump to line 667 because the condition on line 648 was always true
649 partitions.append(
650 (
651 lenient_urls,
652 lambda: dns_pinning.pinned_notification_send(
653 lenient_urls,
654 allow_localhost=False,
655 allow_private_ips=True,
656 # Plugin/raw-webhook schemes allow private LAN targets
657 # but NOT link-local: cloud-provider metadata lives
658 # there beyond the always-blocked literals (e.g.
659 # some providers' IMDS) and no legitimate self-hosted
660 # notifier does. Mirrors the validator's plugin-scheme
661 # IMDS guard.
662 block_link_local=True,
663 ),
664 )
665 )
667 any_ok = False
668 for urls, guard_factory in partitions:
669 temp_apprise = self._new_apprise()
670 # apprise types ``add(servers=...)`` as an invariant ``list`` whose
671 # element union includes ``str``; a ``list[str]`` is valid at
672 # runtime but mypy rejects it on list invariance (upstream should
673 # use ``Sequence``).
674 if not temp_apprise.add(urls, tag=tag): # type: ignore[arg-type]
675 logger.error("Failed to add service URLs to Apprise")
676 return False
677 # _send_with_retry returns True or raises SendError after
678 # Tenacity exhausts its retries; a raise propagates to send().
679 self._send_with_retry(
680 title, body, temp_apprise, guard_factory, tag, attach
681 )
682 any_ok = True
683 return any_ok
685 def send_event(
686 self,
687 event_type: EventType,
688 context: Dict[str, Any],
689 service_urls: Optional[str] = None,
690 tag: Optional[str] = None,
691 custom_template: Optional[Dict[str, str]] = None,
692 ) -> bool:
693 """
694 Send a notification for a specific event type.
696 Args:
697 event_type: Type of event
698 context: Context data for template formatting
699 service_urls: Comma-separated list of service URLs
700 tag: Optional tag to target specific services
701 custom_template: Optional custom template override
703 Returns:
704 True if notification was sent successfully
705 """
706 logger.debug(f"send_event: event_type={event_type.value}, tag={tag}")
707 logger.debug(f"Context: {context}")
709 # Format notification using template
710 message = NotificationTemplate.format(
711 event_type, context, custom_template
712 )
713 logger.debug(
714 f"Template formatted - title: '{message['title'][:50]}...'"
715 )
717 # Send notification
718 return self.send(
719 title=message["title"],
720 body=message["body"],
721 service_urls=service_urls,
722 tag=tag,
723 )
725 def test_service(self, url: str) -> Dict[str, Any]:
726 """
727 Test a notification service.
729 Args:
730 url: Apprise-compatible service URL
732 Returns:
733 Dict with 'success' boolean and optional 'error' message
734 """
735 # Server-level master switch (env-only). Mirrors
736 # NotificationManager's gate so the "Send Test Notification"
737 # button cannot bypass it. WARNING level so the operator sees
738 # the actionable signal naming the env var.
739 if not self.outbound_allowed:
740 logger.warning(
741 "Notification test refused: outbound notifications are "
742 "disabled at the server level. Set "
743 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See "
744 "SECURITY.md 'Notification Webhook SSRF'."
745 )
746 return {
747 "success": False,
748 "error": (
749 "Outbound notifications are disabled. The server "
750 "administrator must set "
751 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable "
752 "notification webhooks. See SECURITY.md "
753 "'Notification Webhook SSRF' for details."
754 ),
755 }
757 try:
758 # ``url`` is a single form field, but Apprise's own
759 # scheme-aware splitter treats it as a LIST — and commas are
760 # legal URL characters. Validating ``url`` as ONE url
761 # therefore checked only the leading entry while Apprise
762 # still registered, and notified, every other entry:
763 # ``discord://valid/tok,json://<cloud-metadata-host>/x``
764 # passed validation on the discord entry alone and the
765 # cloud-metadata target received the test notification
766 # unvetted. Partition the input with the shared scheme-aware
767 # parser — the same one the configured-send path uses via
768 # ``validate_multiple_urls`` — and vet EVERY entry.
769 url_entries, invalid_fragment = parse_notification_url_list(url)
770 if invalid_fragment is not None:
771 # A URL-like fragment with no scheme: Apprise may repair
772 # it or absorb it into a neighbouring entry, so refuse the
773 # whole input rather than dispatch the parseable subset.
774 # Mirrors validate_multiple_urls. The fragment fails the
775 # validation loop below, which sources the user message.
776 url_entries = [invalid_fragment]
777 if not url_entries: 777 ↛ 778line 777 didn't jump to line 778 because the condition on line 777 was never true
778 return {
779 "success": False,
780 "error": "Invalid notification service URL.",
781 }
782 if len(url_entries) > MAX_NOTIFICATION_TARGETS:
783 logger.warning(
784 f"Test notification refused: {len(url_entries)} targets "
785 f"exceeds the cap of {MAX_NOTIFICATION_TARGETS}."
786 )
787 return {
788 "success": False,
789 "error": (
790 f"Too many notification targets "
791 f"({len(url_entries)}). Maximum is "
792 f"{MAX_NOTIFICATION_TARGETS}."
793 ),
794 }
796 for entry in url_entries:
797 # Validate each entry for security (SSRF prevention) and,
798 # in the same pass, compute whether the admin env-var hint
799 # would actually unblock a recoverable private-IP
800 # rejection. Single-pass avoids a DNS-rebinding TOCTOU
801 # window between the default-level validation and the
802 # elevated-level hint decision — see
803 # NotificationURLValidator.validate_service_url_with_hint.
804 is_valid, error_msg, hint_would_help = (
805 NotificationURLValidator.validate_service_url_with_hint(
806 entry, allow_private_ips=self.allow_private_ips
807 )
808 )
810 if not is_valid:
811 # Never log the entry itself — an Apprise service URL
812 # carries credentials (bot tokens, webhook secrets).
813 # #5576 removed the masked-URL echo from this line for
814 # exactly that reason; keep it out of the per-entry
815 # loop too.
816 logger.warning(
817 f"Test service URL validation failed: {error_msg}"
818 )
819 # Surface the validator's reason so users know what to
820 # fix. The hostname/scheme echoed here was supplied by
821 # the user in the same request, so this is not a
822 # server-side leak. When the rejection is a recoverable
823 # private/internal IP — one that
824 # LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true OR
825 # LDR_SECURITY_ALLOW_NAT64=true would actually unblock —
826 # append the admin escape hatches; the user cannot fix
827 # this themselves. Suppress the hint for always-blocked
828 # categories (metadata, 6to4, Teredo, discard,
829 # IPv4-mapped IPv6 of metadata, NAT64-wrapped metadata):
830 # no env var can help, so naming one would mislead.
831 user_error = (
832 error_msg or "Invalid notification service URL."
833 )
834 if hint_would_help:
835 user_error += (
836 ". To unblock this destination, ask the server "
837 "administrator to set "
838 "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true "
839 "(IPv6-only NAT64 deployments also need "
840 "LDR_SECURITY_ALLOW_NAT64=true). "
841 "See SECURITY.md 'Notification Webhook SSRF'."
842 )
843 return {
844 "success": False,
845 "error": user_error,
846 }
848 # Create temporary Apprise instance (synchronous so the pin +
849 # block-private window below applies in this thread).
850 temp_apprise = self._new_apprise()
851 add_result = temp_apprise.add(url)
853 if not add_result:
854 return {
855 "success": False,
856 "error": "Failed to add service URL",
857 }
859 # Fail closed on a parser differential: if Apprise registered
860 # MORE targets than we validated, at least one destination
861 # would be notified without ever passing the SSRF validator.
862 # Fewer is harmless (Apprise dropped an entry), so only the
863 # smuggling direction is refused.
864 if len(temp_apprise) > len(url_entries):
865 logger.warning(
866 f"Test notification refused: Apprise registered "
867 f"{len(temp_apprise)} targets but only "
868 f"{len(url_entries)} were validated."
869 )
870 return {
871 "success": False,
872 "error": (
873 "Notification service URL could not be parsed "
874 "unambiguously; refusing to send. Configure one "
875 "service URL per test."
876 ),
877 }
879 # Send the test through the SAME guarded dispatch core as the real
880 # send path (:meth:`_guarded_notify`) — fail-closed invariants,
881 # per-plugin redirect-disable, and the pin + block-private window
882 # — instead of re-implementing it here. Single-shot (no Tenacity
883 # retry) for fast test feedback. allow_private_ips mirrors the
884 # validator's per-scheme policy: the operator flag for http/https,
885 # True (metadata-only) for plugin / raw-webhook schemes. tag is
886 # always None here.
887 #
888 # With multi-entry input the batch takes the STRICTEST policy of
889 # the schemes present: deriving it from the leading entry alone
890 # would let ``json://…,http://…`` run the http target under the
891 # plugin batch's allow_private_ips=True.
892 schemes = {
893 entry.split(":", 1)[0].lower()
894 for entry in url_entries
895 if ":" in entry
896 }
897 has_http = bool(schemes & {"http", "https"})
898 is_plugin_scheme = not schemes or bool(schemes - {"http", "https"})
899 allow_private = self.allow_private_ips if has_http else True
900 # Plugin/raw-webhook schemes run with allow_private_ips=True but
901 # must still block the whole link-local range (metadata territory
902 # beyond the always-blocked literals). Mirrors _dispatch's lenient
903 # partition and the validator's plugin-scheme IMDS guard.
904 guard_factory = functools.partial(
905 dns_pinning.pinned_notification_send,
906 list(url_entries),
907 allow_localhost=False,
908 allow_private_ips=allow_private,
909 block_link_local=is_plugin_scheme,
910 )
912 result = self._guarded_notify(
913 "Test Notification",
914 (
915 "This is a test notification from Local Deep Research. "
916 "If you see this, your service is configured correctly!"
917 ),
918 temp_apprise,
919 guard_factory,
920 tag=None,
921 )
923 if result:
924 return {
925 "success": True,
926 "message": "Test notification sent successfully",
927 }
929 # A guarded test send that failed AND tripped the block-private
930 # window is a confirmed SSRF block: the window raised
931 # socket.gaierror on a send-time resolution to an
932 # internal/private/metadata IP, which Apprise swallowed into a
933 # generic delivery failure. Surface the specific reason to the
934 # admin instead of the generic message — mirrors the
935 # consume_ssrf_block short-circuit in _send_with_retry. (The pin
936 # path's own connect-time rebind catch raises ValueError before
937 # this point and is reported via the except below; this covers the
938 # UNPINNED block-window path — plugin-scheme endpoints and
939 # unresolvable-at-pin-time hosts.)
940 if dns_pinning.consume_ssrf_block():
941 logger.warning(
942 "Test notification blocked by the SSRF block-private "
943 "window (send-time resolution to internal/private/"
944 "metadata)"
945 )
946 return {
947 "success": False,
948 "error": (
949 "Test notification refused: the destination resolved "
950 "to an internal/private/metadata address at send time "
951 "(possible SSRF). See SECURITY.md 'Notification "
952 "Webhook SSRF'."
953 ),
954 }
955 return {
956 "success": False,
957 "error": "Failed to send test notification",
958 }
960 except Exception:
961 logger.exception("Error testing notification service")
962 return {
963 "success": False,
964 "error": "Failed to test notification service.",
965 }
967 @staticmethod
968 def _validate_url(url: str) -> None:
969 """
970 Validate a notification service URL.
972 Args:
973 url: URL to validate
975 Raises:
976 ServiceError: If URL is invalid
978 Note:
979 URL scheme validation is handled by Apprise itself, which maintains
980 a comprehensive whitelist of supported notification services.
981 Apprise will reject unsupported schemes like 'file://' or 'javascript://'.
982 See: https://github.com/caronc/apprise/wiki
983 """
984 if not url or not isinstance(url, str):
985 raise ServiceError("URL must be a non-empty string")
987 # Check if it looks like a URL
988 parsed = urlparse(url)
989 if not parsed.scheme:
990 raise ServiceError(
991 "Invalid URL format. Must be an Apprise-compatible "
992 "service URL (e.g., discord://webhook_id/token)"
993 )
995 def get_service_type(self, url: str) -> Optional[str]:
996 """
997 Detect service type from URL.
999 Args:
1000 url: Service URL
1002 Returns:
1003 Service type name or None if unknown
1004 """
1005 for service_name, pattern in self.SERVICE_PATTERNS.items():
1006 if re.match(pattern, url, re.IGNORECASE):
1007 return service_name
1008 return "unknown"