From 846dfe3d74c1900e3972f9fa081cc88d6d1aecd2 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sun, 1 Nov 2026 13:44:32 +0100 Subject: [PATCH] fix(notifications): throttle failed-login alerts where they are published Every failed login was published on the event bus, and the e-mail, Telegram and Discord notifiers exempt login events from their cooldown so that successful logins are never swallowed. An unauthenticated login flood therefore filled each notifier's queue and crowded out every other alert, a successful login, an Xray crash or a node going down. UserLoginNotify now passes a failed attempt only for the first failure of its source address in a minute, at most ten addresses per minute; successful logins are always published. --- internal/web/service/tgbot/tgbot_report.go | 37 +++++++++- .../service/tgbot/tgbot_reports_usage_test.go | 72 +++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/internal/web/service/tgbot/tgbot_report.go b/internal/web/service/tgbot/tgbot_report.go index dd7a65c3f..1b377896e 100644 --- a/internal/web/service/tgbot/tgbot_report.go +++ b/internal/web/service/tgbot/tgbot_report.go @@ -208,7 +208,40 @@ func (t *Tgbot) prepareServerUsageInfo() string { return info } -// UserLoginNotify publishes a login event to the event bus. +const ( + loginFailAlertWindow = 1 * time.Minute + // Distinct source addresses alerted per window; the address is the caller's + // to vary, so this ceiling keeps a broad flood below the notifier queue size. + loginFailAlertMaxSources = 10 +) + +var loginFailAlerts = &loginFailAlertGate{} + +// loginFailAlertGate throttles failed-login alerts before they reach the bus: the +// notifiers exempt login events from their cooldown, so a flood crowds out every other alert. +type loginFailAlertGate struct { + mu sync.Mutex + windowStart time.Time + sources map[string]struct{} +} + +// allow passes the first failure of each source address in a window. +func (g *loginFailAlertGate) allow(ip string, now time.Time) bool { + g.mu.Lock() + defer g.mu.Unlock() + if now.Sub(g.windowStart) >= loginFailAlertWindow { + g.windowStart = now + g.sources = make(map[string]struct{}, loginFailAlertMaxSources) + } + if _, seen := g.sources[ip]; seen || len(g.sources) >= loginFailAlertMaxSources { + return false + } + g.sources[ip] = struct{}{} + return true +} + +// UserLoginNotify publishes a login event to the event bus. Successful logins +// are always published; failed ones are throttled by loginFailAlerts. func (t *Tgbot) UserLoginNotify(attempt LoginAttempt) { if attempt.Username == "" || attempt.IP == "" || attempt.Time == "" { logger.Warning("UserLoginNotify failed, invalid info!") @@ -222,6 +255,8 @@ func (t *Tgbot) UserLoginNotify(attempt LoginAttempt) { status := "fail" if attempt.Status == LoginSuccess { status = "success" + } else if !loginFailAlerts.allow(attempt.IP, time.Now()) { + return } EventBus.Publish(eventbus.Event{ diff --git a/internal/web/service/tgbot/tgbot_reports_usage_test.go b/internal/web/service/tgbot/tgbot_reports_usage_test.go index e30cae594..bcd3a4b2b 100644 --- a/internal/web/service/tgbot/tgbot_reports_usage_test.go +++ b/internal/web/service/tgbot/tgbot_reports_usage_test.go @@ -2,10 +2,13 @@ package tgbot import ( "encoding/json" + "fmt" "path/filepath" "strconv" "strings" + "sync/atomic" "testing" + "time" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/dbtest" @@ -365,3 +368,72 @@ func TestLoginAlertKeepsUsernameOnOneLine(t *testing.T) { } } } + +// The notifiers exempt login events from their cooldown, so failed logins are throttled where +// they are published, or an unauthenticated flood crowds every other alert out of the queues. +func TestLoginFailAlertGate(t *testing.T) { + g := &loginFailAlertGate{} + now := time.Now() + + if !g.allow("192.0.2.1", now) { + t.Fatal("first failure of a source must be alerted") + } + if g.allow("192.0.2.1", now.Add(loginFailAlertWindow-time.Second)) { + t.Fatal("a source must be alerted once per window") + } + + passed := 1 + for i := range 1000 { + if g.allow(fmt.Sprintf("2001:db8::%x", i), now.Add(time.Second)) { + passed++ + } + } + if passed != loginFailAlertMaxSources { + t.Fatalf("%d sources alerted in one window, want %d", passed, loginFailAlertMaxSources) + } + if len(g.sources) > loginFailAlertMaxSources { + t.Fatalf("gate tracks %d sources, want at most %d", len(g.sources), loginFailAlertMaxSources) + } + + if !g.allow("192.0.2.1", now.Add(loginFailAlertWindow)) { + t.Fatal("a new window must alert again") + } +} + +func TestUserLoginNotifyThrottlesOnlyFailures(t *testing.T) { + bus := eventbus.New(0) + defer bus.Stop() + prevBus, prevGate := EventBus, loginFailAlerts + EventBus, loginFailAlerts = bus, &loginFailAlertGate{} + defer func() { EventBus, loginFailAlerts = prevBus, prevGate }() + + var failures, successes atomic.Int32 + bus.Subscribe("counter", func(e eventbus.Event) { + if e.Data.(*eventbus.LoginEventData).Status == "success" { + successes.Add(1) + } else { + failures.Add(1) + } + }) + + bot := &Tgbot{} + for range 50 { + bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.1", Time: "now", Status: LoginFail, Reason: "too many failed attempts"}) + } + bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.2", Time: "now", Status: LoginFail, Reason: "invalid credentials"}) + for range 3 { + bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.1", Time: "now", Status: LoginSuccess}) + } + + // The bus delivers in publication order, so the last success arrives after every failure. + deadline := time.Now().Add(5 * time.Second) + for successes.Load() < 3 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := successes.Load(); got != 3 { + t.Fatalf("successful logins reported %d times, want 3", got) + } + if got := failures.Load(); got != 2 { + t.Fatalf("failed logins from two sources reported %d times, want 2", got) + } +}