Coverage for src/local_deep_research/notifications/manager.py: 97%

1""" 

2High-level notification manager with database integration. 

3""" 

4 

5from dataclasses import dataclass 

6from datetime import datetime, timezone, timedelta 

7from enum import Enum 

8from typing import Dict, Optional, Any 

9from urllib.parse import urlunsplit 

10from collections import deque 

11import threading 

12 

13from loguru import logger 

14 

15from ..security.notification_validator import parse_notification_url_list 

16from ..constants import DEFAULT_SEARCH_TOOL 

17from ..security.ssrf_validator import redact_url_for_log 

18from ..security.notification_destination import ( 

19 NotificationDestinationError, 

20 NotificationDestinationKind, 

21 parse_notification_destination, 

22) 

23from ..settings.env_registry import get_env_setting 

24from .service import NotificationService 

25from .templates import EventType 

26from .exceptions import ( 

27 RateLimitError, 

28 SendError, 

29 ServiceError, 

30 SecurityBlockError, 

31) 

32 

33 

34class NotificationReason(str, Enum): 

35 """Precise reason a notification was (not) sent. 

36 

37 Surfaced in :class:`NotificationResult` so callers and log lines can 

38 distinguish, for example, a per-user toggle being off from the 

39 server-level master switch being unset — see issue #4877. 

40 

41 Note: rate-limit refusal is signalled via :class:`RateLimitError` 

42 (preserved API for callers that catch it directly) rather than as a 

43 ``NotificationResult`` value. 

44 """ 

45 

46 SENT = "sent" 

47 SERVER_DISABLED = "server_disabled" 

48 EVENT_DISABLED = "event_disabled" 

49 UNCONFIGURED = "unconfigured" 

50 EGRESS_DENIED = "egress_denied" 

51 INVALID_URL = "invalid_url" 

52 WEBHOOK_FAILED = "webhook_failed" 

53 EXCEPTION = "exception" 

54 

55 

56@dataclass(frozen=True) 

57class NotificationResult: 

58 """Structured outcome of :meth:`NotificationManager.send_notification`. 

59 

60 ``sent`` keeps the legacy bool contract (``if result:`` keeps working — 

61 see ``__bool__``) while ``reason`` names the precise cause so the queue 

62 helpers and error reporter can log a useful, non-misleading line. 

63 ``detail`` is a short operator-facing explanation complementing reason. 

64 """ 

65 

66 sent: bool 

67 reason: NotificationReason 

68 detail: str = "" 

69 

70 def __bool__(self) -> bool: 

71 return self.sent 

72 

73 

74class NotificationManager: 

75 """ 

76 High-level notification manager that uses settings snapshots for 

77 thread-safe access to user settings. 

78 

79 This manager is designed to be used from background threads (e.g., queue 

80 processors) by passing a settings_snapshot dictionary captured from the 

81 main thread. 

82 

83 **Per-User Rate Limiting:** 

84 The rate limiter is shared across ALL NotificationManager instances as a 

85 singleton, but supports per-user rate limit configuration. Each user has 

86 their own rate limits based on their settings, which are configured when 

87 the NotificationManager is initialized with the required user_id parameter. 

88 

89 **How It Works:** 

90 - The first NotificationManager instance creates the shared RateLimiter 

91 with default limits 

92 - Each instance configures user-specific limits by passing user_id to __init__ 

93 - The rate limiter maintains separate counters and limits for each user 

94 - Users are completely isolated - one user's limit doesn't affect others 

95 

96 Example: 

97 >>> # User A with conservative limits 

98 >>> snapshot_a = {"notifications.rate_limit_per_hour": 5} 

99 >>> manager_a = NotificationManager(snapshot_a, user_id="user_a") 

100 >>> # ✅ user_a gets 5/hour 

101 >>> 

102 >>> # User B with generous limits (doesn't affect User A!) 

103 >>> snapshot_b = {"notifications.rate_limit_per_hour": 20} 

104 >>> manager_b = NotificationManager(snapshot_b, user_id="user_b") 

105 >>> # ✅ user_b gets 20/hour, user_a still has 5/hour 

106 """ 

107 

108 # Shared rate limiter instance across all NotificationManager instances 

109 # This ensures rate limits are enforced correctly even when multiple 

110 # NotificationManager instances are created 

111 _shared_rate_limiter: Optional["RateLimiter"] = None 

112 _rate_limiter_lock = threading.Lock() 

113 

114 def __init__(self, settings_snapshot: Dict[str, Any], user_id: str): 

115 """ 

116 Initialize the notification manager. 

117 

118 Args: 

119 settings_snapshot: Dictionary of settings key-value pairs captured 

120 from SettingsManager.get_settings_snapshot(). 

121 This allows thread-safe access to user settings 

122 from background threads. 

123 user_id: User identifier for per-user rate limiting. The rate limits 

124 from settings_snapshot will be configured for this user. 

125 

126 Example: 

127 >>> # In main thread with database session 

128 >>> settings_manager = SettingsManager(session) 

129 >>> snapshot = settings_manager.get_settings_snapshot() 

130 >>> 

131 >>> # In background thread (thread-safe) 

132 >>> notification_manager = NotificationManager( 

133 ... settings_snapshot=snapshot, 

134 ... user_id="user123" 

135 ... ) 

136 >>> notification_manager.send_notification(...) 

137 """ 

138 # Store settings snapshot for thread-safe access 

139 self._settings_snapshot = settings_snapshot 

140 self._user_id = user_id 

141 

142 # Security: read from server-side environment variable only — never from 

143 # user-writable DB settings. Previously this was read via 

144 # _get_setting("notifications.allow_private_ips"), which allowed any 

145 # user to bypass SSRF protection through the settings API. 

146 # Registered as env-only in settings/env_definitions/security.py 

147 # so SettingsManager will always read LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS 

148 # from the environment, never from the database. 

149 allow_private_ips = get_env_setting( 

150 "notifications.allow_private_ips", False 

151 ) 

152 

153 # Server-level master switch (env-only). Distinct from the per-user 

154 # notifications.enabled toggle in the UI: this gate is set by the 

155 # deployment operator and cannot be flipped via the user-writable 

156 # settings API. Disabled by default; enabling it accepts the 

157 # documented DNS-rebinding TOCTOU residual risk 

158 # (see SECURITY.md "Notification Webhook SSRF"). 

159 self._outbound_allowed = bool( 

160 get_env_setting("notifications.allow_outbound", False) 

161 ) 

162 

163 self.service = NotificationService( 

164 allow_private_ips=allow_private_ips, 

165 outbound_allowed=self._outbound_allowed, 

166 ) 

167 

168 # Initialize shared rate limiter on first use 

169 # The shared rate limiter now supports per-user limits, so each user's 

170 # settings are respected regardless of initialization order. 

171 with NotificationManager._rate_limiter_lock: 

172 if NotificationManager._shared_rate_limiter is None: 

173 # Create shared rate limiter with default limits 

174 # (individual users can have different limits) 

175 default_max_per_hour = self._get_setting( 

176 "notifications.rate_limit_per_hour", default=10 

177 ) 

178 default_max_per_day = self._get_setting( 

179 "notifications.rate_limit_per_day", default=50 

180 ) 

181 

182 logger.info( 

183 f"Initializing shared rate limiter with defaults: " 

184 f"{default_max_per_hour}/hour, {default_max_per_day}/day" 

185 ) 

186 

187 NotificationManager._shared_rate_limiter = RateLimiter( 

188 max_per_hour=default_max_per_hour, 

189 max_per_day=default_max_per_day, 

190 ) 

191 

192 # Use the shared instance 

193 self._rate_limiter = NotificationManager._shared_rate_limiter 

194 

195 # Configure per-user rate limits 

196 max_per_hour = self._get_setting( 

197 "notifications.rate_limit_per_hour", default=10 

198 ) 

199 max_per_day = self._get_setting( 

200 "notifications.rate_limit_per_day", default=50 

201 ) 

202 

203 self._rate_limiter.set_user_limits( 

204 user_id, max_per_hour, max_per_day 

205 ) 

206 logger.debug( 

207 f"Configured rate limits for user {user_id}: " 

208 f"{max_per_hour}/hour, {max_per_day}/day" 

209 ) 

210 

211 def _get_setting(self, key: str, default: Any = None) -> Any: 

212 """ 

213 Get a setting value from snapshot. 

214 

215 Args: 

216 key: Setting key 

217 default: Default value if not found 

218 

219 Returns: 

220 Setting value or default 

221 """ 

222 return self._settings_snapshot.get(key, default) 

223 

224 def send_notification( 

225 self, 

226 event_type: EventType, 

227 context: Dict[str, Any], 

228 force: bool = False, 

229 ) -> NotificationResult: 

230 """ 

231 Send a notification for an event. 

232 

233 Uses the user_id that was provided during initialization for 

234 rate limiting and user preferences. 

235 

236 Args: 

237 event_type: Type of event 

238 context: Context data for the notification 

239 force: If True, bypass rate limiting 

240 

241 Returns: 

242 :class:`NotificationResult` describing the outcome. The 

243 dataclass is truthy iff the notification was actually sent, 

244 so legacy ``if manager.send_notification(...):`` callers 

245 continue to work without source changes. 

246 

247 Raises: 

248 RateLimitError: If rate limit is exceeded and force=False 

249 """ 

250 logger.debug( 

251 f"Sending notification: event_type={event_type.value}, " 

252 f"user_id={self._user_id}, force={force}" 

253 ) 

254 logger.debug(f"Context keys: {list(context.keys())}") 

255 

256 # Server-level master switch (env-only). Disabled by default; 

257 # flipping it on is the operator's acknowledgement of the 

258 # documented SSRF rebinding residual risk. force=True does NOT 

259 # bypass this — it bypasses the per-user event-type and 

260 # rate-limit toggles only. WARNING level so an operator wondering 

261 # "why aren't notifications firing?" sees the actionable signal 

262 # at default log level. 

263 if not self._outbound_allowed: 

264 logger.warning( 

265 "Notification refused: outbound notifications are disabled " 

266 "at the server level. Set " 

267 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See " 

268 "SECURITY.md 'Notification Webhook SSRF' for the rationale " 

269 "and residual risk. (event={}, user={})", 

270 event_type.value, 

271 self._user_id, 

272 ) 

273 return NotificationResult( 

274 sent=False, 

275 reason=NotificationReason.SERVER_DISABLED, 

276 detail=( 

277 "outbound notifications are disabled at the server " 

278 "level (LDR_NOTIFICATIONS_ALLOW_OUTBOUND)" 

279 ), 

280 ) 

281 

282 # Check if notifications are enabled for this event type 

283 should_notify = self._should_notify(event_type) 

284 logger.debug( 

285 f"Notification enabled check for {event_type.value}: " 

286 f"{should_notify}" 

287 ) 

288 

289 if not force and not should_notify: 

290 logger.debug( 

291 f"Notifications disabled for event type: " 

292 f"{event_type.value} (user: {self._user_id})" 

293 ) 

294 return NotificationResult( 

295 sent=False, 

296 reason=NotificationReason.EVENT_DISABLED, 

297 detail=f"notifications.on_{event_type.value} is off", 

298 ) 

299 

300 # Check rate limit using the manager's user_id 

301 rate_limit_ok = self._rate_limiter.is_allowed(self._user_id) 

302 logger.debug(f"Rate limit check for {self._user_id}: {rate_limit_ok}") 

303 

304 if not force and not rate_limit_ok: 

305 logger.warning(f"Rate limit exceeded for user {self._user_id}") 

306 raise RateLimitError( 

307 "Notification rate limit exceeded. " 

308 "Please wait before sending more notifications." 

309 ) 

310 

311 try: 

312 # Get service URLs from settings (snapshot or database) 

313 service_urls = self._get_setting( 

314 "notifications.service_url", default="" 

315 ) 

316 

317 if not service_urls or not service_urls.strip(): 

318 logger.debug( 

319 f"No notification service URLs configured for user " 

320 f"{self._user_id}" 

321 ) 

322 return NotificationResult( 

323 sent=False, 

324 reason=NotificationReason.UNCONFIGURED, 

325 detail="notifications.service_url is not configured", 

326 ) 

327 

328 # Reject an unparseable service URL (missing scheme, or a 

329 # scheme-less fragment trailing a valid one) before consulting 

330 # the egress policy at all. `_filter_urls_by_egress_policy` 

331 # also refuses this case (returns ""), which — if reported 

332 # here as EGRESS_DENIED — would falsely claim the policy was 

333 # consulted and refused it, when it was never evaluated. This 

334 # is a genuinely invalid configuration, not a policy denial — 

335 # see issue #5110. 

336 url_entries, invalid_fragment = parse_notification_url_list( 

337 service_urls 

338 ) 

339 if invalid_fragment is not None: 

340 # Log NOTHING derived from the fragment's content — not 

341 # even a redacted form. ``redact_url_for_log`` keeps 

342 # ``scheme://host``, but for the token-in-authority 

343 # Apprise schemes the first authority segment IS the 

344 # secret (``slack://xoxb-SECRET/T00/B00`` redacts to 

345 # ``slack://xoxb-SECRET``), and a fragment can be the 

346 # whole entry when the illegal character only trails it. 

347 # The fragment's length and the parsed-entry count are 

348 # enough to diagnose the misconfiguration. They are 

349 # rendered into the message with positional placeholders: 

350 # loguru kwargs with no template placeholder reach no 

351 # production sink (stderr, log file, database) — #5619. 

352 # The user id and event type are rendered too; this module 

353 # already renders both in its other warnings. 

354 logger.bind(policy_audit=True).warning( 

355 "notification dropped: unparseable service URL " 

356 "fragment (fragment_length={}, entries_parsed={}, " 

357 "user={}, event={})", 

358 len(invalid_fragment), 

359 len(url_entries), 

360 self._user_id, 

361 event_type.value, 

362 ) 

363 return NotificationResult( 

364 sent=False, 

365 reason=NotificationReason.INVALID_URL, 

366 detail=( 

367 "a configured service URL could not be parsed " 

368 "(missing scheme, or an unencoded space, " 

369 "backslash or control character)" 

370 ), 

371 ) 

372 

373 if not url_entries: 

374 # A nonempty setting that parses to zero URLs — 

375 # separator characters only, e.g. "," — never reaches 

376 # the policy's per-URL loop, so the egress branch below 

377 # would report "all configured URLs refused" for a 

378 # refusal that never happened. Invalid configuration, 

379 # same class as a scheme-less fragment — issue #5110. 

380 logger.bind(policy_audit=True).warning( 

381 "notification dropped: service URL setting contains " 

382 "no parseable URLs", 

383 user=self._user_id, 

384 event=event_type.value, 

385 ) 

386 return NotificationResult( 

387 sent=False, 

388 reason=NotificationReason.INVALID_URL, 

389 detail=( 

390 "notifications.service_url contains no parseable URLs" 

391 ), 

392 ) 

393 

394 # Egress policy: each URL is classified by the endpoint Apprise 

395 # will contact. HTTP URLs and authority-bearing plugins are 

396 # checked with evaluate_url against that host; fixed vendor 

397 # endpoints (Matrix t2bot included) are refused under 

398 # PRIVATE_ONLY only, and mailto plus Matrix homeserver/webhook 

399 # modes under PRIVATE_ONLY and STRICT (see 

400 # _filter_urls_by_egress_policy). 

401 # No separator is passed here, so this parses/rejoins with the 

402 # default ",". If a non-default separator is ever exposed to 

403 # callers, it must be passed here too and to service.send_event 

404 # below — otherwise the two calls disagree on where one URL 

405 # ends and the next begins, and the joined output the service 

406 # re-parses merges two entries into one (losing a destination, 

407 # not gaining one). 

408 allowed_urls = self._filter_urls_by_egress_policy(service_urls) 

409 if not allowed_urls: 

410 # Distinguish "the policy was evaluated and refused every 

411 # URL" (empty string) from "the policy itself could not be 

412 # evaluated and we failed closed" (None) so the operator 

413 # fixes the right thing — see issue #5110. 

414 if allowed_urls is None: 

415 detail = "egress policy could not be evaluated" 

416 else: 

417 detail = "all configured URLs refused by egress policy" 

418 logger.bind(policy_audit=True).warning( 

419 "notification dropped: {}", 

420 detail, 

421 user=self._user_id, 

422 event=event_type.value, 

423 ) 

424 return NotificationResult( 

425 sent=False, 

426 reason=NotificationReason.EGRESS_DENIED, 

427 detail=detail, 

428 ) 

429 

430 # Send notification with the allowed subset. No separator is 

431 # passed here either, so this must keep agreeing with the 

432 # default "," used above to build allowed_urls — see the 

433 # comment on the _filter_urls_by_egress_policy call. 

434 logger.debug(f"Calling service.send_event for {event_type.value}") 

435 sent = self.service.send_event( 

436 event_type, context, service_urls=allowed_urls 

437 ) 

438 

439 # Log to database if enabled 

440 if sent: 

441 self._log_notification(event_type, context) 

442 logger.info( 

443 f"Notification sent: {event_type.value} to user " 

444 f"{self._user_id}" 

445 ) 

446 return NotificationResult( 

447 sent=True, 

448 reason=NotificationReason.SENT, 

449 detail="", 

450 ) 

451 # A falsy return from send_event means Apprise rejected a 

452 # service URL (temp_apprise.add() failed — unparseable or 

453 # unsupported scheme) in one of the scheme partitions 

454 # NotificationService._dispatch builds. NOT "none of the 

455 # URLs": _dispatch returns False on the FIRST partition whose 

456 # add() fails, before attempting any later partition — so a 

457 # mixed configuration (e.g. one malformed http(s) URL plus a 

458 # valid json:// URL) is reported here even though the valid 

459 # partition was never tried. Actual delivery failures raise 

460 # SendError and are handled below — see issue #5110. 

461 logger.warning( 

462 f"Notification failed: {event_type.value} to user " 

463 f"{self._user_id} — Apprise rejected a configured " 

464 f"service URL before dispatch completed" 

465 ) 

466 return NotificationResult( 

467 sent=False, 

468 reason=NotificationReason.INVALID_URL, 

469 detail=( 

470 "Apprise rejected a configured service URL before " 

471 "dispatch completed" 

472 ), 

473 ) 

474 

475 except SendError: 

476 # The delivery layer raises SendError ONLY from 

477 # _send_with_retry's own terminal raise once all retries are 

478 # exhausted (dead webhook, HTTP 4xx/5xx, network down) — a 

479 # transient, worth-retrying failure. NotificationService.send() 

480 # re-raises SendError unchanged (no re-wrapping) and does NOT 

481 # launder incidental/unexpected dispatch-time exceptions into 

482 # SendError — those fall through to the generic 

483 # ``except Exception`` branch below and get the EXCEPTION 

484 # reason instead, so this branch stays reserved for genuine 

485 # delivery failures. It also specifically does NOT mean a 

486 # confirmed SSRF/DNS-rebind block: NotificationService.send() 

487 # routes that non-retryable case through ServiceError instead 

488 # (see the except clause below) precisely so it is never 

489 # mislabeled as a retryable webhook failure here. Keep the 

490 # detail static regardless: SendError's own raise site 

491 # (NotificationService._send_with_retry) uses a fixed message 

492 # with no interpolated URL/token today, but this except clause 

493 # also logs the exception via logger.exception above — a 

494 # static, non-interpolated detail here keeps that guarantee 

495 # independent of what any future SendError raise site puts in 

496 # its message. 

497 logger.exception( 

498 f"Webhook delivery failed for {event_type.value} to user " 

499 f"{self._user_id}" 

500 ) 

501 return NotificationResult( 

502 sent=False, 

503 reason=NotificationReason.WEBHOOK_FAILED, 

504 detail="webhook delivery failed after retries", 

505 ) 

506 

507 except SecurityBlockError: 

508 # A ServiceError subclass NotificationService.send() raises 

509 # specifically for a confirmed send-time SSRF/DNS-rebind block 

510 # (destination resolved to an internal/private/metadata 

511 # address after dispatch started) — as opposed to the plain 

512 # ServiceError below, raised for a pre-dispatch validation 

513 # reject. Both are permanent, invalid-destination outcomes 

514 # that map to the same non-transient INVALID_URL reason, but 

515 # this one gets a distinct detail: the generic "check your 

516 # URL/settings" text below would misleadingly suggest a 

517 # config mistake when the true cause is an active security 

518 # block. Static detail — no URL or resolved IP — the 

519 # exception message can echo parts of the URL. 

520 logger.exception( 

521 f"Notification destination blocked by SSRF/egress " 

522 f"protection for {event_type.value} to user " 

523 f"{self._user_id}" 

524 ) 

525 return NotificationResult( 

526 sent=False, 

527 reason=NotificationReason.INVALID_URL, 

528 detail=( 

529 "notification destination blocked by egress/SSRF protection" 

530 ), 

531 ) 

532 

533 except ServiceError: 

534 # The service raises ServiceError when the URL security 

535 # validator rejects a URL before any dispatch is attempted 

536 # (structurally invalid URLs, or valid URLs whose destination 

537 # is blocked, e.g. a LAN webhook without 

538 # LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS). A confirmed send-time 

539 # SSRF/DNS-rebind block is caught above as SecurityBlockError 

540 # (a ServiceError subclass) before reaching this branch. The 

541 # detail must not claim the URL itself is malformed. Static 

542 # detail: the exception message can echo parts of the URL. 

543 logger.exception( 

544 f"Service URL rejected for {event_type.value} to user " 

545 f"{self._user_id}" 

546 ) 

547 return NotificationResult( 

548 sent=False, 

549 reason=NotificationReason.INVALID_URL, 

550 detail=( 

551 "service URL rejected by URL security validation " 

552 "(for a private/LAN webhook, check " 

553 "LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS)" 

554 ), 

555 ) 

556 

557 except Exception as exc: 

558 # Reached for unexpected failures anywhere in the block above 

559 # (settings read, egress filtering, or dispatch). The exception 

560 # string may contain webhook URLs with embedded tokens, SMTP 

561 # credentials, or payload content. Log the raw exception for 

562 # operators via logger.exception, but only expose the exception 

563 # class name (and not the message) in the structured result so 

564 # downstream callers can't echo it into logs or UI surfaces 

565 # verbatim. 

566 logger.exception( 

567 f"Error sending notification for {event_type.value} to user " 

568 f"{self._user_id}" 

569 ) 

570 return NotificationResult( 

571 sent=False, 

572 reason=NotificationReason.EXCEPTION, 

573 detail=f"unexpected {type(exc).__name__} while sending", 

574 ) 

575 

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

577 """ 

578 Test a notification service. 

579 

580 Args: 

581 url: Service URL to test 

582 

583 Returns: 

584 Dict with test results 

585 

586 Note: 

587 Nothing in ``src/`` currently calls this method — the live 

588 ``POST /api/notifications/test-url`` route constructs a 

589 ``NotificationService`` directly and never goes through 

590 ``NotificationManager``, so the egress-policy precheck below 

591 is not applied to that endpoint (see issue #5110 follow-up 

592 discussion). Only test suites exercise this method today. 

593 """ 

594 # Classify an unusable URL the same way ``send_notification`` 

595 # now does, BEFORE consulting the egress policy. 

596 # ``_filter_urls_by_egress_policy`` returns ``""`` for three 

597 # different things — an unparseable fragment, zero parseable 

598 # entries, and a genuine per-URL policy refusal — so mapping 

599 # every ``""`` onto the "Set Egress Scope to 'Unprotected'" 

600 # remediation tells the operator to widen a policy that was 

601 # never consulted (issue #5110 / #5113 follow-up). Parse here so 

602 # the first two are reported as invalid configuration and the 

603 # egress message is reserved for a real refusal. 

604 url_entries, invalid_fragment = parse_notification_url_list(url) 

605 if invalid_fragment is not None: 

606 # Log NOTHING derived from the fragment's content, not even 

607 # a redacted form — see the identical branch in 

608 # ``send_notification`` for why ``scheme://host`` is still 

609 # the secret for token-in-authority Apprise schemes. The 

610 # content-free diagnostics are rendered into the message, as 

611 # there (#5619). 

612 logger.bind(policy_audit=True).warning( 

613 "test notification refused: unparseable service URL " 

614 "fragment (fragment_length={}, entries_parsed={})", 

615 len(invalid_fragment), 

616 len(url_entries), 

617 ) 

618 return { 

619 "status": "error", 

620 "message": ( 

621 "Service URL could not be parsed (missing scheme, or " 

622 "an unencoded space, backslash or control character)." 

623 ), 

624 } 

625 if not url_entries: 

626 return { 

627 "status": "error", 

628 "message": "Service URL contains no parseable URLs.", 

629 } 

630 

631 # Egress policy precheck before forwarding to Apprise. NOT 

632 # currently reached by any live route — see the Note above. 

633 allowed = self._filter_urls_by_egress_policy(url) 

634 if allowed is None: 

635 # Fail-closed: the policy could not be evaluated at all, so 

636 # don't tell the user their scope refused the URL — the 

637 # "Set Egress Scope to 'Unprotected'" remediation below 

638 # fixes the wrong thing. Same distinction send_notification 

639 # draws between "refused every URL" and "could not be 

640 # evaluated". 

641 return { 

642 "status": "error", 

643 "message": ( 

644 "Egress policy could not be evaluated; check the " 

645 "egress policy settings (policy.egress_scope)." 

646 ), 

647 } 

648 if not allowed: 

649 # Reached only for a real per-URL refusal now: the fragment 

650 # and zero-entry cases returned above. 

651 return { 

652 "status": "error", 

653 "message": ( 

654 "URL refused by egress policy. Set Egress Scope to " 

655 "'Unprotected' (or pick a local webhook) to test this URL." 

656 ), 

657 } 

658 return self.service.test_service(allowed) 

659 

660 def _filter_urls_by_egress_policy( 

661 self, service_urls: str, separator: str = "," 

662 ) -> Optional[str]: 

663 """Filter an Apprise URL string by the user's egress policy. 

664 

665 Default comma input preserves commas inside one service URL and is 

666 canonicalized to whitespace-separated output. Custom-separated input 

667 retains its caller-provided separator for the service to consume. 

668 Returns the joined allowed URLs (possibly empty if every URL was 

669 refused), or ``None`` when the policy could not be evaluated and we 

670 failed closed. Both are falsy; callers that care why nothing was 

671 allowed check for ``None``. When no snapshot / context is available, 

672 return the input unchanged for backwards compatibility. 

673 """ 

674 if not isinstance(separator, str) or not separator: 

675 raise ValueError("separator must be a non-empty string") 

676 snapshot = getattr(self, "_settings_snapshot", None) 

677 if not snapshot: 

678 return service_urls 

679 

680 url_entries, invalid_fragment = parse_notification_url_list( 

681 service_urls, separator 

682 ) 

683 if invalid_fragment is not None: 

684 # Same rule as the identical branches in 

685 # ``send_notification`` and ``test_service`` (both parse 

686 # first and return before calling this method, so this 

687 # branch only fires for a direct caller): never log anything 

688 # derived from the fragment's content. Even the redacted 

689 # ``scheme://host`` form leaks the secret for token-in- 

690 # authority Apprise schemes, and the fragment may be the 

691 # entire entry. The content-free diagnostics are rendered 

692 # into the message with positional placeholders, as at those 

693 # sites — loguru kwargs with no template placeholder never 

694 # appear in any production sink's output (#5619). 

695 logger.bind(policy_audit=True).warning( 

696 "all notification URLs refused: unparseable URL " 

697 "fragment (fragment_length={}, entries_parsed={})", 

698 len(invalid_fragment), 

699 len(url_entries), 

700 ) 

701 return "" 

702 

703 try: 

704 from ..security.egress.policy import ( 

705 EgressScope, 

706 PolicyDeniedError, 

707 context_from_snapshot, 

708 evaluate_url, 

709 resolve_run_primary_engine, 

710 ) 

711 except ImportError: 

712 logger.debug( 

713 "egress_policy unavailable in notifications manager; " 

714 "URLs will not be scope-gated" 

715 ) 

716 return service_urls 

717 

718 try: 

719 ctx = context_from_snapshot( 

720 snapshot, 

721 resolve_run_primary_engine( 

722 snapshot, default=DEFAULT_SEARCH_TOOL 

723 ), 

724 # Queue snapshots need not carry identity metadata. Resolve 

725 # private retrievers in the notification owner's namespace. 

726 username=self._user_id, 

727 ) 

728 except (PolicyDeniedError, ValueError) as exc: 

729 # Snapshot present but policy cannot be evaluated. The 

730 # previous bare-except dropped to "return service_urls" 

731 # here, which dispatched every URL unfiltered — fail-open 

732 # on a misconfigured policy. Now refuse all URLs instead 

733 # so an Apprise dispatch with no scope check is impossible. 

734 logger.bind(policy_audit=True).warning( 

735 "all notification URLs refused: egress policy could " 

736 "not be evaluated", 

737 reason=str(exc), 

738 ) 

739 return None 

740 

741 # Classify the endpoint Apprise will contact. Authority-bearing 

742 # plugins must obey the same user egress scope as an HTTP webhook; 

743 # token-based vendor plugins (Matrix t2bot and ntfy cloud included) 

744 # are refused in PRIVATE_ONLY and pass every other scope. 

745 parts = [] 

746 for url_entry in url_entries: 

747 try: 

748 destination = parse_notification_destination(url_entry) 

749 except NotificationDestinationError: 

750 # Preserve the invalid-URL outcome: the service validator 

751 # will reject this entry before Apprise sees it. Dropping it 

752 # here would misreport a malformed URL as an egress refusal. 

753 parts.append(url_entry) 

754 continue 

755 if destination.kind is NotificationDestinationKind.VENDOR_PLUGIN: 

756 if ctx.scope == EgressScope.PRIVATE_ONLY: 

757 logger.bind(policy_audit=True).warning( 

758 "notification vendor destination refused under " 

759 "PRIVATE_ONLY egress scope" 

760 ) 

761 continue 

762 parts.append(url_entry) 

763 continue 

764 # Mail and Matrix homeserver/webhook URLs are refused under 

765 # PRIVATE_ONLY and STRICT whatever their host. Apprise's mail 

766 # plugin can connect to a built-in provider's fixed SMTP host 

767 # instead of the URL authority. The Matrix refusal is 

768 # conservative: Apprise runs server discovery only for 

769 # ``matrixs://``, which the classifier rejects, so ``matrix://`` 

770 # sends to its URL host. Matrix t2bot is a VENDOR_PLUGIN and was 

771 # handled above (refused only under PRIVATE_ONLY). 

772 scheme = url_entry.split(":", 1)[0].lower() 

773 if scheme in {"matrix", "mailto"} and ctx.scope in { 

774 EgressScope.PRIVATE_ONLY, 

775 EgressScope.STRICT, 

776 }: 

777 logger.bind(policy_audit=True).warning( 

778 "notification destination refused: dynamic endpoint " 

779 "cannot be verified local", 

780 scheme=scheme, 

781 ) 

782 continue 

783 if destination.kind is NotificationDestinationKind.HTTP: 

784 policy_url = url_entry 

785 else: 

786 effective_host = destination.effective_host 

787 if not effective_host: 787 ↛ 788line 787 didn't jump to line 788 because the condition on line 787 was never true

788 continue 

789 policy_host = ( 

790 f"[{effective_host}]" 

791 if ":" in effective_host 

792 else effective_host 

793 ) 

794 policy_url = urlunsplit(("http", policy_host, "", "", "")) 

795 try: 

796 decision = evaluate_url(policy_url, ctx) 

797 except ( 

798 AttributeError, 

799 ImportError, 

800 RuntimeError, 

801 TypeError, 

802 ValueError, 

803 ) as exc: 

804 logger.bind(policy_audit=True).warning( 

805 "notification URL refused: egress policy evaluation failed", 

806 reason=type(exc).__name__, 

807 ) 

808 continue 

809 if decision.allowed: 

810 parts.append(url_entry) 

811 else: 

812 audit = logger.bind(policy_audit=True) 

813 if destination.kind is NotificationDestinationKind.HTTP: 

814 audit = audit.bind(url=redact_url_for_log(url_entry)) 

815 audit.warning( 

816 "notification URL refused by egress policy", 

817 scope=ctx.scope.value, 

818 reason=decision.reason, 

819 ) 

820 # Rejoin with the caller's separator so the filtered output stays 

821 # parseable by the same rule it was parsed with. If a non-default 

822 # separator is ever exposed to callers, it must be threaded through 

823 # every call site that reaches this method (see send_notification's 

824 # _filter_urls_by_egress_policy call and its service.send_event call 

825 # above) — a desync here does not add a destination, it MERGES two 

826 # entries into one unparseable one and silently loses a destination. 

827 joiner = " " if separator == "," else separator 

828 return joiner.join(parts) 

829 

830 def _should_notify(self, event_type: EventType) -> bool: 

831 """ 

832 Check if notifications should be sent for this event type. 

833 

834 Uses the manager's settings snapshot to determine if the event type 

835 is enabled for the user. 

836 

837 Args: 

838 event_type: Event type to check 

839 

840 Returns: 

841 True if notifications should be sent 

842 """ 

843 try: 

844 # Check event-specific setting (from snapshot or database) 

845 setting_key = f"notifications.on_{event_type.value}" 

846 enabled = self._get_setting(setting_key, default=False) 

847 

848 return bool(enabled) 

849 

850 except Exception: 

851 logger.warning("Error checking notification preferences") 

852 # Default to disabled on error to avoid infinite loops during login 

853 return False 

854 

855 def _log_notification( 

856 self, event_type: EventType, context: Dict[str, Any] 

857 ) -> None: 

858 """ 

859 Log a sent notification (simplified logging to application logs only). 

860 

861 Uses the manager's user_id for logging. 

862 

863 Args: 

864 event_type: Event type 

865 context: Notification context 

866 """ 

867 try: 

868 title = ( 

869 context.get("query") 

870 or context.get("subscription_name") 

871 or "Unknown" 

872 ) 

873 logger.info( 

874 f"Notification sent: {event_type.value} - {title} " 

875 f"(user: {self._user_id})" 

876 ) 

877 except Exception as e: 

878 logger.debug(f"Failed to log notification: {e}") 

879 

880 

881class RateLimiter: 

882 """ 

883 Simple in-memory rate limiter for notifications with per-user limit support. 

884 

885 This rate limiter tracks notification counts per user and enforces 

886 configurable rate limits. Each user can have their own rate limits, 

887 which are stored separately from the notification counts. 

888 

889 **Per-User Limits:** 

890 Rate limits can be configured per-user using `set_user_limits()`. 

891 If no user-specific limits are set, the default limits (passed to 

892 __init__) are used. 

893 

894 **Memory Storage:** 

895 This implementation stores rate limits in memory only, which means 

896 limits are reset when the server restarts. This is acceptable for normal 

897 users since they cannot restart the server. If an admin restarts the server, 

898 rate limits reset which is reasonable behavior. 

899 

900 **Thread Safety:** 

901 This implementation is thread-safe using threading.Lock() for concurrent 

902 requests from the same user. 

903 

904 **Multi-Worker Limitation:** 

905 In multi-worker deployments, each worker process maintains its own rate 

906 limit counters. Users could potentially bypass rate limits by distributing 

907 requests across different workers, getting up to N × max_per_hour 

908 notifications (where N = number of workers). For single-worker deployments 

909 (the default for LDR), this is not a concern. For production multi-worker 

910 deployments, consider implementing Redis-based rate limiting. 

911 

912 Example: 

913 >>> limiter = RateLimiter(max_per_hour=10, max_per_day=50) 

914 >>> # Set custom limits for specific user 

915 >>> limiter.set_user_limits("user_a", max_per_hour=5, max_per_day=25) 

916 >>> limiter.set_user_limits("user_b", max_per_hour=20, max_per_day=100) 

917 >>> # Users get their configured limits 

918 >>> limiter.is_allowed("user_a") # Limited to 5/hour 

919 >>> limiter.is_allowed("user_b") # Limited to 20/hour 

920 >>> limiter.is_allowed("user_c") # Uses defaults: 10/hour 

921 """ 

922 

923 def __init__( 

924 self, 

925 max_per_hour: int = 10, 

926 max_per_day: int = 50, 

927 cleanup_interval_hours: int = 24, 

928 ): 

929 """ 

930 Initialize rate limiter with default limits. 

931 

932 Args: 

933 max_per_hour: Default maximum notifications per hour per user 

934 max_per_day: Default maximum notifications per day per user 

935 cleanup_interval_hours: How often to run cleanup of inactive users (hours) 

936 """ 

937 # Default limits used when no user-specific limits are set 

938 self.max_per_hour = max_per_hour 

939 self.max_per_day = max_per_day 

940 self.cleanup_interval_hours = cleanup_interval_hours 

941 

942 # Per-user rate limit configuration (user_id -> (max_per_hour, max_per_day)) 

943 self._user_limits: Dict[str, tuple[int, int]] = {} 

944 

945 # Per-user notification counts 

946 self._hourly_counts: Dict[str, deque] = {} 

947 self._daily_counts: Dict[str, deque] = {} 

948 

949 self._last_cleanup = datetime.now(timezone.utc) 

950 self._lock = threading.Lock() # Thread safety for all operations 

951 

952 def set_user_limits( 

953 self, user_id: str, max_per_hour: int, max_per_day: int 

954 ) -> None: 

955 """ 

956 Set rate limits for a specific user. 

957 

958 This allows each user to have their own rate limit configuration. 

959 If not set, the user will use the default limits passed to __init__. 

960 

961 Args: 

962 user_id: User identifier 

963 max_per_hour: Maximum notifications per hour for this user 

964 max_per_day: Maximum notifications per day for this user 

965 

966 Example: 

967 >>> limiter = RateLimiter(max_per_hour=10, max_per_day=50) 

968 >>> limiter.set_user_limits("power_user", max_per_hour=20, max_per_day=100) 

969 >>> limiter.set_user_limits("limited_user", max_per_hour=5, max_per_day=25) 

970 """ 

971 with self._lock: 

972 self._user_limits[user_id] = (max_per_hour, max_per_day) 

973 logger.debug( 

974 f"Set rate limits for user {user_id}: " 

975 f"{max_per_hour}/hour, {max_per_day}/day" 

976 ) 

977 

978 def get_user_limits(self, user_id: str) -> tuple[int, int]: 

979 """ 

980 Get the effective rate limits for a user. 

981 

982 Returns the user-specific limits if set, otherwise returns defaults. 

983 

984 Args: 

985 user_id: User identifier 

986 

987 Returns: 

988 Tuple of (max_per_hour, max_per_day) 

989 """ 

990 with self._lock: 

991 return self._user_limits.get( 

992 user_id, (self.max_per_hour, self.max_per_day) 

993 ) 

994 

995 def is_allowed(self, user_id: str) -> bool: 

996 """ 

997 Check if a notification is allowed for a user. 

998 

999 Uses per-user rate limits if configured via set_user_limits(), 

1000 otherwise uses the default limits from __init__. 

1001 

1002 Args: 

1003 user_id: User identifier 

1004 

1005 Returns: 

1006 True if notification is allowed, False if rate limit exceeded 

1007 """ 

1008 with self._lock: 

1009 now = datetime.now(timezone.utc) 

1010 

1011 # Periodic cleanup of inactive users 

1012 self._cleanup_inactive_users_if_needed(now) 

1013 

1014 # Initialize queues for user if needed 

1015 if user_id not in self._hourly_counts: 

1016 self._hourly_counts[user_id] = deque() 

1017 self._daily_counts[user_id] = deque() 

1018 

1019 # Clean old entries 

1020 self._clean_old_entries(user_id, now) 

1021 

1022 # Get user-specific limits or defaults 

1023 max_per_hour, max_per_day = self._user_limits.get( 

1024 user_id, (self.max_per_hour, self.max_per_day) 

1025 ) 

1026 

1027 # Check limits 

1028 hourly_count = len(self._hourly_counts[user_id]) 

1029 daily_count = len(self._daily_counts[user_id]) 

1030 

1031 if hourly_count >= max_per_hour: 

1032 logger.warning( 

1033 f"Hourly rate limit exceeded for user {user_id}: " 

1034 f"{hourly_count}/{max_per_hour}" 

1035 ) 

1036 return False 

1037 

1038 if daily_count >= max_per_day: 

1039 logger.warning(