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) + } +}