Coverage for src/local_deep_research/notifications/service.py: 98%
97 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Core notification service using Apprise.
3"""
5import re
6from typing import Dict, List, Optional, Any
7from urllib.parse import urlparse
9import apprise
10from loguru import logger
11from tenacity import (
12 retry,
13 stop_after_attempt,
14 wait_exponential,
15 retry_if_exception_type,
16)
18from .exceptions import ServiceError, SendError
19from .templates import EventType, NotificationTemplate
20from ..security.url_builder import mask_sensitive_url
21from ..security.notification_validator import (
22 NotificationURLValidator,
23)
25PRIVATE_IP_REJECTION_PREFIX = (
26 NotificationURLValidator.PRIVATE_IP_REJECTION_PREFIX
27)
30# Backward compatibility constants - now handled by Tenacity internally
31MAX_RETRY_ATTEMPTS = 3
32INITIAL_RETRY_DELAY = 0.5
33RETRY_BACKOFF_MULTIPLIER = 2
36class NotificationService:
37 """
38 Low-level notification service that wraps Apprise.
39 """
41 # Regex patterns for common service types (for validation)
42 SERVICE_PATTERNS = {
43 "email": r"^mailto://",
44 "discord": r"^discord://",
45 "slack": r"^slack://",
46 "telegram": r"^tgram://",
47 "smtp": r"^(smtp|smtps)://",
48 }
50 def __init__(
51 self,
52 allow_private_ips: bool = False,
53 outbound_allowed: bool = False,
54 ):
55 """
56 Initialize the notification service.
58 Args:
59 allow_private_ips: Whether to allow notifications to private/local IPs
60 (default: False for security). Set to True for
61 development/testing environments only.
62 outbound_allowed: Server-level master switch
63 (env-only via LDR_NOTIFICATIONS_ALLOW_OUTBOUND).
64 Default False — outbound notifications are off until
65 the operator opts in. See SECURITY.md "Notification
66 Webhook SSRF" for the rationale.
67 """
68 self.apprise = apprise.Apprise()
69 self.allow_private_ips = allow_private_ips
70 self.outbound_allowed = outbound_allowed
72 @retry(
73 stop=stop_after_attempt(3),
74 wait=wait_exponential(multiplier=1, min=0.5, max=10),
75 retry=retry_if_exception_type((Exception,)),
76 reraise=True,
77 )
78 def _send_with_retry(
79 self,
80 title: str,
81 body: str,
82 apprise_instance: apprise.Apprise,
83 tag: Optional[str] = None,
84 attach: Optional[List[str]] = None,
85 ) -> bool:
86 """
87 Send a notification using the provided Apprise instance with retry logic.
89 This method is decorated with Tenacity to handle retries automatically.
91 Args:
92 title: Notification title
93 body: Notification body text
94 apprise_instance: Apprise instance to use for sending
95 tag: Optional tag to target specific services
96 attach: Optional list of file paths to attach
98 Returns:
99 True if notification was sent successfully
101 Raises:
102 SendError: If sending fails after all retry attempts
103 """
104 logger.debug(
105 f"Sending notification: title='{title[:50]}...', tag={tag}"
106 )
107 logger.debug(f"Body preview: {body[:200]}...")
109 # Send notification
110 notify_result = apprise_instance.notify(
111 title=title,
112 body=body,
113 tag=tag,
114 attach=attach,
115 )
117 if notify_result:
118 logger.debug(f"Notification sent successfully: '{title[:50]}...'")
119 return True
120 error_msg = "Failed to send notification to any service"
121 logger.warning(error_msg)
122 raise SendError(error_msg)
124 def send(
125 self,
126 title: str,
127 body: str,
128 service_urls: Optional[str] = None,
129 tag: Optional[str] = None,
130 attach: Optional[List[str]] = None,
131 ) -> bool:
132 """
133 Send a notification to service URLs with automatic retry.
135 Args:
136 title: Notification title
137 body: Notification body text
138 service_urls: Comma-separated list of service URLs to override configured ones
139 tag: Optional tag to target specific services
140 attach: Optional list of file paths to attach
142 Returns:
143 True if notification was sent successfully to at least one service
145 Raises:
146 SendError: If sending fails after all retry attempts
148 Note:
149 Temporary Apprise instances are created for each send operation
150 and are automatically garbage collected by Python when they go
151 out of scope. This simple approach is ideal for small deployments
152 (~5 users) and avoids memory management complexity.
153 """
154 # Defense-in-depth: enforce the operator-level master switch at
155 # the service layer too, not just at NotificationManager.
156 # Today the manager always wraps this method, but keeping the
157 # gate here means a future direct caller cannot accidentally
158 # bypass it. See SECURITY.md "Notification Webhook SSRF".
159 if not self.outbound_allowed:
160 logger.warning(
161 "Notification not sent: outbound notifications are disabled "
162 "at the server level. Set "
163 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See "
164 "SECURITY.md 'Notification Webhook SSRF'."
165 )
166 return False
168 # If service_urls are provided, validate before trying to send
169 if service_urls:
170 logger.debug("Creating Apprise instance for provided service URLs")
172 # Validate service URLs for security (SSRF prevention)
173 is_valid, error_msg = (
174 NotificationURLValidator.validate_multiple_urls(
175 service_urls, allow_private_ips=self.allow_private_ips
176 )
177 )
179 if not is_valid:
180 logger.error(
181 f"Service URL validation failed: {error_msg}. "
182 f"URL: {mask_sensitive_url(service_urls)}"
183 )
184 raise ServiceError(f"Invalid service URL: {error_msg}")
186 try:
187 # If service_urls are provided, create a new Apprise instance
188 if service_urls:
189 temp_apprise = apprise.Apprise()
190 result = temp_apprise.add(service_urls, tag=tag)
192 if not result:
193 logger.error(
194 f"Failed to add service URLs to Apprise: "
195 f"{mask_sensitive_url(service_urls)}"
196 )
197 return False
199 # Send notification with the temp instance (with retry)
200 return self._send_with_retry(
201 title, body, temp_apprise, tag, attach
202 )
203 # Use the configured apprise instance
204 if len(self.apprise) == 0: 204 ↛ 209line 204 didn't jump to line 209 because the condition on line 204 was always true
205 logger.debug("No notification services configured in Apprise")
206 return False
208 # Send notification (with retry)
209 return self._send_with_retry(title, body, self.apprise, tag, attach)
211 except Exception as e:
212 # Tenacity will retry, but if all retries fail, raise SendError
213 logger.exception(
214 f"Failed to send notification after retries: '{title[:50]}...'"
215 )
216 raise SendError(f"Failed to send notification: {str(e)}")
218 def send_event(
219 self,
220 event_type: EventType,
221 context: Dict[str, Any],
222 service_urls: Optional[str] = None,
223 tag: Optional[str] = None,
224 custom_template: Optional[Dict[str, str]] = None,
225 ) -> bool:
226 """
227 Send a notification for a specific event type.
229 Args:
230 event_type: Type of event
231 context: Context data for template formatting
232 service_urls: Comma-separated list of service URLs
233 tag: Optional tag to target specific services
234 custom_template: Optional custom template override
236 Returns:
237 True if notification was sent successfully
238 """
239 logger.debug(f"send_event: event_type={event_type.value}, tag={tag}")
240 logger.debug(f"Context: {context}")
242 # Format notification using template
243 message = NotificationTemplate.format(
244 event_type, context, custom_template
245 )
246 logger.debug(
247 f"Template formatted - title: '{message['title'][:50]}...'"
248 )
250 # Send notification
251 return self.send(
252 title=message["title"],
253 body=message["body"],
254 service_urls=service_urls,
255 tag=tag,
256 )
258 def test_service(self, url: str) -> Dict[str, Any]:
259 """
260 Test a notification service.
262 Args:
263 url: Apprise-compatible service URL
265 Returns:
266 Dict with 'success' boolean and optional 'error' message
267 """
268 # Server-level master switch (env-only). Mirrors
269 # NotificationManager's gate so the "Send Test Notification"
270 # button cannot bypass it. WARNING level so the operator sees
271 # the actionable signal naming the env var.
272 if not self.outbound_allowed:
273 logger.warning(
274 "Notification test refused: outbound notifications are "
275 "disabled at the server level. Set "
276 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See "
277 "SECURITY.md 'Notification Webhook SSRF'."
278 )
279 return {
280 "success": False,
281 "error": (
282 "Outbound notifications are disabled. The server "
283 "administrator must set "
284 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable "
285 "notification webhooks. See SECURITY.md "
286 "'Notification Webhook SSRF' for details."
287 ),
288 }
290 try:
291 # Validate service URL for security (SSRF prevention) and,
292 # in the same pass, compute whether the admin env-var hint
293 # would actually unblock a recoverable private-IP rejection.
294 # Single-pass avoids a DNS-rebinding TOCTOU window between
295 # the default-level validation and the elevated-level hint
296 # decision — see NotificationURLValidator.validate_service_url_with_hint.
297 is_valid, error_msg, hint_would_help = (
298 NotificationURLValidator.validate_service_url_with_hint(
299 url, allow_private_ips=self.allow_private_ips
300 )
301 )
303 if not is_valid:
304 logger.warning(
305 f"Test service URL validation failed: {error_msg}. "
306 f"URL: {mask_sensitive_url(url)}"
307 )
308 # Surface the validator's reason so users know what to fix.
309 # The hostname/scheme echoed here was supplied by the user
310 # in the same request, so this is not a server-side leak.
311 # When the rejection is a recoverable private/internal IP —
312 # one that LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true OR
313 # LDR_SECURITY_ALLOW_NAT64=true would actually unblock —
314 # append the admin escape hatches; the user cannot fix this
315 # themselves. Suppress the hint for always-blocked categories
316 # (metadata, 6to4, Teredo, discard, IPv4-mapped IPv6 of
317 # metadata, NAT64-wrapped metadata): no env var can help, so
318 # naming one would mislead.
319 user_error = error_msg or "Invalid notification service URL."
320 if hint_would_help:
321 user_error += (
322 ". To unblock this destination, ask the server "
323 "administrator to set "
324 "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS=true "
325 "(IPv6-only NAT64 deployments also need "
326 "LDR_SECURITY_ALLOW_NAT64=true). "
327 "See SECURITY.md 'Notification Webhook SSRF'."
328 )
329 return {
330 "success": False,
331 "error": user_error,
332 }
334 # Create temporary Apprise instance
335 temp_apprise = apprise.Apprise()
336 add_result = temp_apprise.add(url)
338 if not add_result:
339 return {
340 "success": False,
341 "error": "Failed to add service URL",
342 }
344 # Send test notification
345 result = temp_apprise.notify(
346 title="Test Notification",
347 body=(
348 "This is a test notification from Local Deep Research. "
349 "If you see this, your service is configured correctly!"
350 ),
351 )
353 if result:
354 return {
355 "success": True,
356 "message": "Test notification sent successfully",
357 }
358 return {
359 "success": False,
360 "error": "Failed to send test notification",
361 }
363 except Exception:
364 logger.exception("Error testing notification service")
365 return {
366 "success": False,
367 "error": "Failed to test notification service.",
368 }
370 @staticmethod
371 def _validate_url(url: str) -> None:
372 """
373 Validate a notification service URL.
375 Args:
376 url: URL to validate
378 Raises:
379 ServiceError: If URL is invalid
381 Note:
382 URL scheme validation is handled by Apprise itself, which maintains
383 a comprehensive whitelist of supported notification services.
384 Apprise will reject unsupported schemes like 'file://' or 'javascript://'.
385 See: https://github.com/caronc/apprise/wiki
386 """
387 if not url or not isinstance(url, str):
388 raise ServiceError("URL must be a non-empty string")
390 # Check if it looks like a URL
391 parsed = urlparse(url)
392 if not parsed.scheme:
393 raise ServiceError(
394 "Invalid URL format. Must be an Apprise-compatible "
395 "service URL (e.g., discord://webhook_id/token)"
396 )
398 def get_service_type(self, url: str) -> Optional[str]:
399 """
400 Detect service type from URL.
402 Args:
403 url: Service URL
405 Returns:
406 Service type name or None if unknown
407 """
408 for service_name, pattern in self.SERVICE_PATTERNS.items():
409 if re.match(pattern, url, re.IGNORECASE):
410 return service_name
411 return "unknown"