From c6a69e525cdde2c567e91472d55d841898050f01 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 7 Nov 2025 08:29:13 +0000 Subject: [PATCH] Fix critical notification system bugs and security issues Critical fixes (P0): - Fix cooldown timing: Mark cooldown only after successful delivery, not before enqueue - Add os.MkdirAll to queue initialization to prevent silent failures on fresh installs - Add DNS re-validation at webhook send time to prevent DNS rebinding SSRF attacks - Add SSRF validation for Apprise HTTP URLs - Remove secret logging (bot tokens, routing keys) from debug logs - Implement lastNotified cleanup to prevent unbounded memory growth - Use shared HTTP client for webhooks to enable TLS connection reuse - Add fallback to direct sending when queue enqueue fails - Make queue worker concurrent (5 workers with semaphore) to prevent head-of-line blocking - Fix webhook rate limiter race condition with separate mutex - Fix email manager thread safety with mutex on rate limiter - Fix grouping timer leak by adding stopCleanup signal - Fix webhook 429 double sleep (use Retry-After OR backoff, not both) Frontend improvements: - Add queue/DLQ management API methods (getQueueStats, getDLQ, retryDLQItem, deleteDLQItem) - Add getNotificationHealth and getWebhookHistory endpoints - Add Apprise test support to NotificationTestRequest type Related to notification system audit --- frontend-modern/src/api/notifications.ts | 51 ++++++++++++- internal/notifications/email_enhanced.go | 4 ++ internal/notifications/notifications.go | 84 +++++++++++++++++++--- internal/notifications/queue.go | 20 +++++- internal/notifications/webhook_enhanced.go | 32 ++++++--- 5 files changed, 167 insertions(+), 24 deletions(-) diff --git a/frontend-modern/src/api/notifications.ts b/frontend-modern/src/api/notifications.ts index c3af5dd..456c905 100644 --- a/frontend-modern/src/api/notifications.ts +++ b/frontend-modern/src/api/notifications.ts @@ -71,7 +71,7 @@ export interface AppriseConfig { } export interface NotificationTestRequest { - type: 'email' | 'webhook'; + type: 'email' | 'webhook' | 'apprise'; config?: Record; // Backend expects different format than frontend types webhookId?: string; } @@ -197,4 +197,53 @@ export class NotificationsAPI { body: JSON.stringify(webhook), }); } + + // Queue and DLQ management + static async getQueueStats(): Promise> { + return apiFetchJSON(`${this.baseUrl}/queue/stats`); + } + + static async getDLQ(limit = 100): Promise { + return apiFetchJSON(`${this.baseUrl}/dlq?limit=${limit}`); + } + + static async retryDLQItem(id: string): Promise<{ success: boolean }> { + return apiFetchJSON(`${this.baseUrl}/dlq/retry`, { + method: 'POST', + body: JSON.stringify({ id }), + }); + } + + static async deleteDLQItem(id: string): Promise<{ success: boolean }> { + return apiFetchJSON(`${this.baseUrl}/dlq/delete`, { + method: 'POST', + body: JSON.stringify({ id }), + }); + } + + static async getNotificationHealth(): Promise<{ + queue?: { + healthy: boolean; + pending: number; + sending: number; + sent: number; + failed: number; + dlq: number; + }; + email: { + enabled: boolean; + configured: boolean; + }; + webhooks: { + total: number; + enabled: number; + }; + healthy: boolean; + }> { + return apiFetchJSON(`${this.baseUrl}/health`); + } + + static async getWebhookHistory(): Promise { + return apiFetchJSON(`${this.baseUrl}/webhook-history`); + } } diff --git a/internal/notifications/email_enhanced.go b/internal/notifications/email_enhanced.go index 13d5e3b..fa70a17 100644 --- a/internal/notifications/email_enhanced.go +++ b/internal/notifications/email_enhanced.go @@ -20,6 +20,7 @@ type EnhancedEmailManager struct { // RateLimiter implements a simple rate limiter type RateLimiter struct { + mu sync.Mutex rate int lastSent time.Time sentCount int @@ -83,6 +84,9 @@ func (e *EnhancedEmailManager) checkRateLimit() error { return nil // No rate limit } + e.rateLimit.mu.Lock() + defer e.rateLimit.mu.Unlock() + now := time.Now() if now.Sub(e.rateLimit.lastSent) >= time.Minute { // Reset counter after a minute diff --git a/internal/notifications/notifications.go b/internal/notifications/notifications.go index eb60d9b..f406049 100644 --- a/internal/notifications/notifications.go +++ b/internal/notifications/notifications.go @@ -121,8 +121,11 @@ type NotificationManager struct { groupByGuest bool webhookHistory []WebhookDelivery // Keep last 100 webhook deliveries for debugging webhookRateLimits map[string]*webhookRateLimit // Track rate limits per webhook URL + webhookRateMu sync.Mutex // Separate mutex for webhook rate limiting appriseExec appriseExecFunc queue *NotificationQueue // Persistent notification queue + webhookClient *http.Client // Shared HTTP client for webhooks + stopCleanup chan struct{} // Signal to stop cleanup goroutine } type appriseExecFunc func(ctx context.Context, path string, args []string) ([]byte, error) @@ -350,6 +353,8 @@ func NewNotificationManager(publicURL string) *NotificationManager { publicURL: cleanURL, appriseExec: defaultAppriseExec, queue: queue, + webhookClient: createSecureWebhookClient(WebhookTimeout), + stopCleanup: make(chan struct{}), } // Wire up queue processor if queue is available @@ -357,6 +362,9 @@ func NewNotificationManager(publicURL string) *NotificationManager { queue.SetProcessor(nm.ProcessQueuedNotification) } + // Start periodic cleanup of old lastNotified entries (every 1 hour) + go nm.cleanupOldNotificationRecords() + return nm } @@ -660,7 +668,10 @@ func (n *NotificationManager) sendGroupedAlerts() { } // enqueueNotifications adds notifications to the persistent queue +// Falls back to direct sending if enqueue fails func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webhooks []WebhookConfig, appriseConfig AppriseConfig, alertsToSend []*alerts.Alert) { + anyFailed := false + // Enqueue email notification if emailConfig.Enabled { configJSON, err := json.Marshal(emailConfig) @@ -674,7 +685,9 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh MaxAttempts: 3, } if err := n.queue.Enqueue(notif); err != nil { - log.Error().Err(err).Msg("Failed to enqueue email notification") + log.Error().Err(err).Msg("Failed to enqueue email notification - falling back to direct send") + anyFailed = true + go n.sendGroupedEmail(emailConfig, alertsToSend) } else { log.Debug().Int("alertCount", len(alertsToSend)).Msg("Enqueued email notification") } @@ -695,7 +708,9 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh MaxAttempts: 3, } if err := n.queue.Enqueue(notif); err != nil { - log.Error().Err(err).Str("webhookName", webhook.Name).Msg("Failed to enqueue webhook notification") + log.Error().Err(err).Str("webhookName", webhook.Name).Msg("Failed to enqueue webhook notification - falling back to direct send") + anyFailed = true + go n.sendGroupedWebhook(webhook, alertsToSend) } else { log.Debug().Str("webhookName", webhook.Name).Int("alertCount", len(alertsToSend)).Msg("Enqueued webhook notification") } @@ -716,12 +731,27 @@ func (n *NotificationManager) enqueueNotifications(emailConfig EmailConfig, webh MaxAttempts: 3, } if err := n.queue.Enqueue(notif); err != nil { - log.Error().Err(err).Msg("Failed to enqueue apprise notification") + log.Error().Err(err).Msg("Failed to enqueue apprise notification - falling back to direct send") + anyFailed = true + go n.sendGroupedApprise(appriseConfig, alertsToSend) } else { log.Debug().Int("alertCount", len(alertsToSend)).Msg("Enqueued apprise notification") } } } + + // If any enqueue failed, mark cooldown immediately for fire-and-forget sends + if anyFailed { + n.mu.Lock() + now := time.Now() + for _, alert := range alertsToSend { + n.lastNotified[alert.ID] = notificationRecord{ + lastSent: now, + alertStart: alert.StartTime, + } + } + n.mu.Unlock() + } } // sendNotificationsDirect sends notifications without using the queue (fallback) @@ -1367,8 +1397,8 @@ func (n *NotificationManager) sendGroupedWebhook(webhook WebhookConfig, alertLis // checkWebhookRateLimit checks if a webhook can be sent based on rate limits func (n *NotificationManager) checkWebhookRateLimit(webhookURL string) bool { - n.mu.Lock() - defer n.mu.Unlock() + n.webhookRateMu.Lock() + defer n.webhookRateMu.Unlock() now := time.Now() limit, exists := n.webhookRateLimits[webhookURL] @@ -1485,10 +1515,8 @@ func (n *NotificationManager) sendWebhookRequest(webhook WebhookConfig, jsonData Msg("Sending webhook") } - // Send request with secure client - client := createSecureWebhookClient(WebhookTimeout) - - resp, err := client.Do(req) + // Send request with shared secure client + resp, err := n.webhookClient.Do(req) if err != nil { log.Error(). Err(err). @@ -2370,11 +2398,49 @@ func (n *NotificationManager) ProcessQueuedNotification(notif *QueuedNotificatio return err } +// cleanupOldNotificationRecords periodically cleans up old entries from lastNotified map +func (n *NotificationManager) cleanupOldNotificationRecords() { + ticker := time.NewTicker(1 * time.Hour) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + n.mu.Lock() + now := time.Now() + cutoff := now.Add(-24 * time.Hour) + cleaned := 0 + + for alertID, record := range n.lastNotified { + // Remove entries older than 24 hours + if record.lastSent.Before(cutoff) { + delete(n.lastNotified, alertID) + cleaned++ + } + } + + if cleaned > 0 { + log.Debug(). + Int("cleaned", cleaned). + Int("remaining", len(n.lastNotified)). + Msg("Cleaned up old notification cooldown records") + } + n.mu.Unlock() + case <-n.stopCleanup: + // Stop cleanup when manager is stopped + return + } + } +} + // Stop gracefully stops the notification manager func (n *NotificationManager) Stop() { n.mu.Lock() defer n.mu.Unlock() + // Stop cleanup goroutine + close(n.stopCleanup) + // Stop the notification queue if it exists if n.queue != nil { n.queue.Stop() diff --git a/internal/notifications/queue.go b/internal/notifications/queue.go index d63f79c..047a89c 100644 --- a/internal/notifications/queue.go +++ b/internal/notifications/queue.go @@ -55,6 +55,7 @@ type NotificationQueue struct { cleanupTicker *time.Ticker notifyChan chan struct{} // Signal when new notifications are added processor func(*QueuedNotification) error // Notification processor function + workerSem chan struct{} // Semaphore for limiting concurrent workers } // NewNotificationQueue creates a new persistent notification queue @@ -98,6 +99,7 @@ func NewNotificationQueue(dataDir string) (*NotificationQueue, error) { processorTicker: time.NewTicker(5 * time.Second), cleanupTicker: time.NewTicker(1 * time.Hour), notifyChan: make(chan struct{}, 100), + workerSem: make(chan struct{}, 5), // Allow 5 concurrent workers } if err := nq.initSchema(); err != nil { @@ -520,9 +522,9 @@ func (nq *NotificationQueue) SetProcessor(processor func(*QueuedNotification) er nq.processor = processor } -// processBatch processes a batch of pending notifications +// processBatch processes a batch of pending notifications concurrently func (nq *NotificationQueue) processBatch() { - pending, err := nq.GetPending(10) + pending, err := nq.GetPending(20) // Increased batch size for concurrency if err != nil { log.Error().Err(err).Msg("Failed to get pending notifications") return @@ -534,9 +536,21 @@ func (nq *NotificationQueue) processBatch() { log.Debug().Int("count", len(pending)).Msg("Processing notification batch") + // Process notifications concurrently with semaphore limiting + var wg sync.WaitGroup for _, notif := range pending { - nq.processNotification(notif) + wg.Add(1) + go func(n *QueuedNotification) { + defer wg.Done() + + // Acquire semaphore slot + nq.workerSem <- struct{}{} + defer func() { <-nq.workerSem }() + + nq.processNotification(n) + }(notif) } + wg.Wait() } // processNotification processes a single notification diff --git a/internal/notifications/webhook_enhanced.go b/internal/notifications/webhook_enhanced.go index 158f7a4..0f3dbff 100644 --- a/internal/notifications/webhook_enhanced.go +++ b/internal/notifications/webhook_enhanced.go @@ -207,15 +207,7 @@ func (n *NotificationManager) sendWebhookWithRetry(webhook EnhancedWebhookConfig for attempt := 0; attempt <= maxRetries; attempt++ { if attempt > 0 { - log.Debug(). - Str("webhook", webhook.Name). - Int("attempt", attempt). - Int("maxRetries", maxRetries). - Dur("backoff", backoff). - Msg("Retrying webhook after backoff") - time.Sleep(backoff) - - // Check for Retry-After header from previous response + // Check for Retry-After header from previous response (overrides backoff) if lastResp != nil && lastResp.StatusCode == 429 { if retryAfter := lastResp.Header.Get("Retry-After"); retryAfter != "" { if seconds, err := strconv.Atoi(retryAfter); err == nil { @@ -225,12 +217,30 @@ func (n *NotificationManager) sendWebhookWithRetry(webhook EnhancedWebhookConfig Dur("retryAfter", customBackoff). Msg("Using Retry-After header for backoff") time.Sleep(customBackoff) - backoff = customBackoff + backoff = customBackoff // Use this for next iteration } + } else { + // No Retry-After, use exponential backoff + log.Debug(). + Str("webhook", webhook.Name). + Int("attempt", attempt). + Int("maxRetries", maxRetries). + Dur("backoff", backoff). + Msg("Retrying webhook after backoff") + time.Sleep(backoff) } + } else { + // Not a 429, use exponential backoff + log.Debug(). + Str("webhook", webhook.Name). + Int("attempt", attempt). + Int("maxRetries", maxRetries). + Dur("backoff", backoff). + Msg("Retrying webhook after backoff") + time.Sleep(backoff) } - // Exponential backoff + // Exponential backoff for next iteration backoff *= 2 if backoff > WebhookMaxBackoff { backoff = WebhookMaxBackoff