package notify import ( "context" "fmt" "log/slog" "net/http" "strings" "sync" "time" "github.com/fonu/fonu/internal/secret" "github.com/fonu/fonu/internal/settings" ) type Service struct { settings *settings.Store store *Store logger *slog.Logger client *http.Client accessMu sync.Mutex accessHits map[string][]time.Time accessAlerted map[string]time.Time loginMu sync.Mutex loginHits map[string][]time.Time loginAlerted map[string]time.Time } func New(settingsStore *settings.Store, secretBox *secret.Box, logger *slog.Logger) *Service { if logger == nil { logger = slog.Default() } return &Service{ settings: settingsStore, store: NewStore(settingsStore, secretBox), logger: logger, client: &http.Client{Timeout: 10 * time.Second}, accessHits: map[string][]time.Time{}, accessAlerted: map[string]time.Time{}, loginHits: map[string][]time.Time{}, loginAlerted: map[string]time.Time{}, } } func (s *Service) Alert(ctx context.Context, event, title, message string) { if s == nil || s.store == nil { return } cfg, err := s.store.LoadRuntime(ctx) if err != nil { s.logger.Warn("notify load config failed", "error", err.Error()) return } if !s.enabledFor(cfg, event) { return } if err := s.send(ctx, cfg, event, title, message); err != nil { s.logger.Warn("notify send failed", "event", event, "error", err.Error()) } } func (s *Service) RecordAccessHit(ctx context.Context, ip string) { s.recordThresholdHit(ctx, ip, EventIPFrequentAccess, func(cfg RuntimeConfig) (threshold, windowSec, cooldownSec int, enabled bool) { enabled = s.enabledFor(cfg, EventIPFrequentAccess) threshold = cfg.IPFrequentThreshold if threshold <= 0 { threshold = DefaultIPFrequentThreshold } windowSec = cfg.IPFrequentWindowSec if windowSec <= 0 { windowSec = DefaultIPFrequentWindowSec } return threshold, windowSec, DefaultIPFrequentAlertCooldown, enabled }, &s.accessMu, s.accessHits, s.accessAlerted, func(ip string, windowSec, count, threshold int) string { return fmt.Sprintf("IP %s 在 %d 秒内访问 %d 次(阈值 %d)", ip, windowSec, count, threshold) }, "IP 频繁访问") } func (s *Service) RecordLoginFailure(ctx context.Context, ip string) { s.recordThresholdHit(ctx, ip, EventLoginFailure, func(cfg RuntimeConfig) (threshold, windowSec, cooldownSec int, enabled bool) { enabled = s.enabledFor(cfg, EventLoginFailure) threshold = cfg.LoginFailureThreshold if threshold <= 0 { threshold = DefaultLoginFailureThreshold } windowSec = cfg.LoginFailureWindowSec if windowSec <= 0 { windowSec = DefaultLoginFailureWindowSec } return threshold, windowSec, DefaultLoginFailureAlertCooldown, enabled }, &s.loginMu, s.loginHits, s.loginAlerted, func(ip string, windowSec, count, threshold int) string { return fmt.Sprintf("IP %s 在 %d 秒内登录失败 %d 次(阈值 %d)", ip, windowSec, count, threshold) }, "登录异常") } func (s *Service) recordThresholdHit( ctx context.Context, ip string, event string, cfgFn func(RuntimeConfig) (threshold, windowSec, cooldownSec int, enabled bool), mu *sync.Mutex, hits map[string][]time.Time, alerted map[string]time.Time, messageFn func(ip string, windowSec, count, threshold int) string, title string, ) { if s == nil || s.store == nil { return } ip = strings.TrimSpace(ip) if ip == "" || ip == "-" { return } cfg, err := s.store.LoadRuntime(ctx) if err != nil { return } threshold, windowSec, cooldownSec, enabled := cfgFn(cfg) if !enabled { return } window := time.Duration(windowSec) * time.Second now := time.Now() cutoff := now.Add(-window) mu.Lock() prev := hits[ip] pruned := make([]time.Time, 0, len(prev)+1) for _, at := range prev { if !at.Before(cutoff) { pruned = append(pruned, at) } } pruned = append(pruned, now) hits[ip] = pruned count := len(pruned) lastAlert := alerted[ip] cooldown := time.Duration(cooldownSec) * time.Second shouldAlert := count >= threshold && now.Sub(lastAlert) >= cooldown if shouldAlert { alerted[ip] = now } mu.Unlock() if !shouldAlert { return } s.Alert(ctx, event, title, messageFn(ip, windowSec, count, threshold)) } func (s *Service) SendTest(ctx context.Context, in TestInput) error { runtime, err := s.resolveTestRuntime(ctx, in) if err != nil { return err } if runtime.Type == "" { return fmt.Errorf("请先选择通知方式") } if err := s.send(ctx, runtime, "test", "测试通知", "配置验证通过,当前通知通道可正常使用。"); err != nil { s.logger.Warn("notify test send failed", "error", err.Error()) return err } return nil } func (s *Service) resolveTestRuntime(ctx context.Context, in TestInput) (RuntimeConfig, error) { smtpPassword := strings.TrimSpace(in.SMTPPassword) webhookSecret := strings.TrimSpace(in.WebhookSecret) telegramToken := strings.TrimSpace(in.TelegramToken) if key := strings.TrimSpace(in.Webhook.Key); key != "" && key != MaskedSecret { webhookSecret = key } needStored := smtpPassword == "" || smtpPassword == MaskedSecret || webhookSecret == "" || webhookSecret == MaskedSecret || telegramToken == "" || telegramToken == MaskedSecret if needStored && s.store != nil { stored, err := s.store.LoadRuntime(ctx) if err != nil { return RuntimeConfig{}, err } if (smtpPassword == "" || smtpPassword == MaskedSecret) && in.Email.HasPassword { smtpPassword = stored.SMTPPassword } if (webhookSecret == "" || webhookSecret == MaskedSecret) && in.Webhook.HasSecret { webhookSecret = stored.WebhookSecret } if (telegramToken == "" || telegramToken == MaskedSecret) && in.Telegram.HasBotToken { telegramToken = stored.TelegramToken } } return in.Config.Runtime(smtpPassword, webhookSecret, telegramToken), nil } func (s *Service) Store() *Store { if s == nil { return nil } return s.store } func (s *Service) send(ctx context.Context, cfg RuntimeConfig, event, title, message string) error { content := FormatAlert(event, title, message) switch cfg.Type { case NotifyTypeEmail: return SendEmail(cfg, content.Subject, content.PlainBody) case NotifyTypeWebhook: return SendWebhook(ctx, s.client, cfg, content) case NotifyTypeTelegram: return SendTelegram(ctx, s.client, cfg, content) default: return nil } } func (s *Service) enabledFor(cfg RuntimeConfig, event string) bool { if cfg.Type == "" { return false } switch event { case EventDDNSIPChange: return cfg.OnDDNSIPChange case EventDDNSFailure: return cfg.OnDDNSFailure case EventCertExpiry: return cfg.OnCertExpiry case EventCertRenewSuccess: return cfg.OnCertRenewSuccess case EventCertRenewFailure: return cfg.OnCertRenewFailure case EventIPFrequentAccess: return cfg.OnIPFrequentAccess case EventLoginFailure: return cfg.OnLoginFailure case EventNginxReloadFailure: return cfg.OnNginxReloadFailure case EventResourceThreshold: return cfg.OnResourceThreshold case EventTunnelDown: return cfg.OnTunnelDown case EventFrpsServerDown: return cfg.OnFrpsDown case EventAgentUnreachable: return cfg.OnAgentUnreachable case EventRouteDrift: return cfg.OnRouteDrift case EventBackupFailure: return cfg.OnBackupFailure default: return true } }