From ccd4dd9d85ae4afbefdbcdec40b5c2b2b5f757e6 Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Thu, 29 Oct 2026 01:28:12 +0000 Subject: [PATCH] Build the visitor per request instead of sharing it Requests from one IP (anonymous and untiered users) shared a single visitor object with a mutable user, so overlapping requests could be authorized as each other's user. The visitor is now built per request and never changed; the limit service, keyed by visitor key, still shares the buckets and counts. The Firebase penalty becomes a limiter kind, open subscriptions move to a node-local gauge per key, and tier users' stats are enqueued per admitted unit. Fixes #1051 --- limit/service.go | 21 +++ limit/service_test.go | 21 +++ limit/types.go | 3 +- server/server.go | 31 +-- server/server_account.go | 5 +- server/server_account_test.go | 18 +- server/server_firebase_test.go | 2 +- server/server_manager.go | 16 +- server/server_middleware_test.go | 62 ++++++ server/server_payments.go | 6 +- server/server_payments_test.go | 11 +- server/server_test.go | 20 +- server/visitor.go | 312 ++++++++++++------------------- server/visitor_test.go | 50 ++++- 14 files changed, 308 insertions(+), 270 deletions(-) create mode 100644 server/server_middleware_test.go diff --git a/limit/service.go b/limit/service.go index 3085a5da..2a58d8a3 100644 --- a/limit/service.go +++ b/limit/service.go @@ -181,6 +181,27 @@ func (s *Service) Prune(idle time.Duration) { } } +// Touch marks the key as used now, so Prune keeps it, e.g. for a long-lived subscription +func (s *Service) Touch(key string) { + now := s.now() + sh := s.shards[shardIndex(key)] + sh.mu.Lock() + defer sh.mu.Unlock() + sh.entryNoLock(key, now).seen = now.UnixNano() +} + +// Seen returns when the key was last used, or the zero time if the service does not know it +func (s *Service) Seen(key string) time.Time { + sh := s.shards[shardIndex(key)] + sh.mu.Lock() + defer sh.mu.Unlock() + e, ok := sh.entries[key] + if !ok { + return time.Time{} + } + return time.Unix(0, e.seen).UTC() +} + // Limiter returns a util.Limiter view of the key's bucket for the rule, for the readers and // writers in util that meter bytes. A negative AllowN refunds. func (s *Service) Limiter(key string, rule Rule) util.Limiter { diff --git a/limit/service_test.go b/limit/service_test.go index 039a328a..cd83782b 100644 --- a/limit/service_test.go +++ b/limit/service_test.go @@ -381,3 +381,24 @@ func BenchmarkAllow_Parallel(b *testing.B) { } }) } + +func TestService_TouchAndSeen(t *testing.T) { + s, clock := newTestServiceWithClock(t) + require.True(t, s.Seen("k").IsZero()) + s.Touch("k") + require.Equal(t, clock.now(), s.Seen("k")) + clock.advance(10 * time.Minute) + s.Touch("k") + s.Prune(5 * time.Minute) + require.Equal(t, 1, s.Size()) // Touched, so not idle + require.Equal(t, clock.now(), s.Seen("k")) + clock.advance(10 * time.Minute) + s.Prune(5 * time.Minute) + require.True(t, s.Seen("k").IsZero()) +} + +func TestService_FirebaseKind(t *testing.T) { + kind, ok := ParseKind("firebase") + require.True(t, ok) + require.Equal(t, KindFirebase, kind) +} diff --git a/limit/types.go b/limit/types.go index fa25b95d..b4021828 100644 --- a/limit/types.go +++ b/limit/types.go @@ -18,12 +18,13 @@ const ( KindTopics // Topics created in the server's in-memory topic map KindAccount // Account actions: signups and password reset requests KindAuth // Failed authentication attempts + KindFirebase // Firebase sends, paused for a while after Firebase reports the quota exceeded numKinds // Sentinel, keep last ) // kindNames are the stable names of the kinds, used in the usage table and on the wire; they // must never change once released -var kindNames = [numKinds]string{"requests", "messages", "emails", "calls", "bandwidth", "topics", "account", "auth"} +var kindNames = [numKinds]string{"requests", "messages", "emails", "calls", "bandwidth", "topics", "account", "auth", "firebase"} // Rule says how much of a kind a key may use: a token bucket of Burst tokens refilling one token // per Refill, a Daily cap, or both. A zero Rule allows everything. diff --git a/server/server.go b/server/server.go index f109fe4f..da003177 100644 --- a/server/server.go +++ b/server/server.go @@ -64,9 +64,9 @@ type Server struct { smtpServerBackend *smtpBackend mailer mail.Sender topics map[string]*topic - visitors map[string]*visitor // ip: or user: - limiter *limit.Service // Decides on and counts the visitors' usage, keyed by visitor key - ban *ban.Service // Abuse ban-feed; nil when the feature is disabled (no ban file) + limiter *limit.Service // Decides on and counts the visitors' usage, keyed by visitor key + subscriptions *subscriptionGauge // Open subscriptions per visitor key on this node + ban *ban.Service // Abuse ban-feed; nil when the feature is disabled (no ban file) firebaseClient *firebaseClient twilio *twilio.Client messages int64 // Total number of messages (persisted if messageCache enabled) @@ -301,7 +301,7 @@ func New(conf *Config) (*Server, error) { messages: messages, messagesHistory: []int64{messages}, messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas - visitors: make(map[string]*visitor), + subscriptions: newSubscriptionGauge(), limiter: limit.New(&limit.Config{ResetTime: conf.VisitorStatsResetTime}), stripe: stripe, streams: newStreamRegistry(), @@ -1116,10 +1116,6 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess return nil, err } } - u := v.User() - if s.userManager != nil && u != nil && u.Tier != nil { - go v.EnqueueUserStatsDelta() - } s.mu.Lock() s.messages++ s.mu.Unlock() @@ -2164,9 +2160,6 @@ func (s *Server) resetStats() { s.mu.Lock() defer s.mu.Unlock() // Includes the database query to avoid races with other processes s.limiter.ResetUsage() - for _, v := range s.visitors { - v.ResetStats() - } if s.userManager != nil { if err := s.userManager.ResetStats(); err != nil { log.Tag(tagResetter).Warn("Failed to write to database: %s", err.Error()) @@ -2178,7 +2171,7 @@ func (s *Server) runFirebaseKeepaliver() { if s.firebaseClient == nil { return } - v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0 + v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0 for { select { case <-time.After(s.config.FirebaseKeepaliveInterval): @@ -2357,18 +2350,10 @@ func (s *Server) transformMatrixJSON(next handleFunc) handleFunc { } } +// visitor returns the visitor for this request. It is built per request, so concurrent requests from +// the same IP (which share a limiter key) cannot see each other's user (#1051). func (s *Server) visitor(ip netip.Addr, user *user.User) *visitor { - s.mu.Lock() - defer s.mu.Unlock() - id := visitorID(ip, user, s.config) - v, exists := s.visitors[id] - if !exists { - s.visitors[id] = newVisitor(s.config, s.limiter, s.messageCache, s.userManager, ip, user) - return s.visitors[id] - } - v.Keepalive() - v.SetUser(user) // Always update with the latest user, may be nil! - return v + return newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, ip, user) } func (s *Server) writeJSON(w http.ResponseWriter, v any) error { diff --git a/server/server_account.go b/server/server_account.go index 03466571..24f542ea 100644 --- a/server/server_account.go +++ b/server/server_account.go @@ -326,9 +326,8 @@ func (s *Server) handleAccountTokenCreate(w http.ResponseWriter, r *http.Request func (s *Server) handleAccountTokenUpdate(w http.ResponseWriter, r *http.Request, v *visitor) error { u := v.User() if u == nil { - // ensureUser already checked, but the user can be nil'd between that check and this - // re-read: IP-keyed visitors are shared, and a concurrent anonymous request on the - // same IP calls SetUser(nil) on this very object (recurring production panic) + // ensureUser already checked; kept as a guard, since a nil user here was a recurring + // production panic while visitors were shared between requests (#1051) return errHTTPUnauthorized } req, err := readJSONWithLimit[apiAccountTokenUpdateRequest](r.Body, jsonBodyBytesLimit, true) // Allow empty body! diff --git a/server/server_account_test.go b/server/server_account_test.go index 13d251a8..d7f89977 100644 --- a/server/server_account_test.go +++ b/server/server_account_test.go @@ -928,25 +928,25 @@ func TestAccount_Reservation_Delete_Messages_And_Attachments(t *testing.T) { require.Equal(t, int64(2), account.Stats.Messages) // Is not reset! }*/ -func TestAccount_TokenUpdate_NilUserRace_NoPanic(t *testing.T) { +func TestAccount_TokenUpdate_NilUser_NoPanic(t *testing.T) { forEachBackend(t, func(t *testing.T, databaseURL string) { - // Regression for the recurring production panic in handleAccountTokenUpdate: the - // handler re-reads v.User() after ensureUser already checked it, and a concurrent - // anonymous request on the same IP-keyed visitor can SetUser(nil) in between. The - // handler must return 401, not dereference a nil user. + // Regression for a recurring production panic in handleAccountTokenUpdate: the handler + // re-reads v.User() after ensureUser checked it, and must return 401 for a nil user, not + // dereference it. Visitors were shared between requests then (#1051). conf := newTestConfigWithAuthFile(t, databaseURL) conf.EnableLogin = true s := newTestServer(t, conf) + ip := netip.MustParseAddr("9.9.9.9") require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false)) u, err := s.userManager.User("phil") require.Nil(t, err) - - v := s.visitor(netip.MustParseAddr("9.9.9.9"), u) - v.SetUser(nil) // The concurrent anonymous request, made deterministic + authed := s.visitor(ip, u) + anon := s.visitor(ip, nil) // A concurrent anonymous request from the same IP + require.NotNil(t, authed.User()) r, _ := http.NewRequest("PATCH", "/v1/account/token", strings.NewReader("{}")) rr := httptest.NewRecorder() - err = s.handleAccountTokenUpdate(rr, r, v) + err = s.handleAccountTokenUpdate(rr, r, anon) require.Error(t, err) httpErr, ok := err.(*errHTTP) require.True(t, ok) diff --git a/server/server_firebase_test.go b/server/server_firebase_test.go index 90dc5d35..1bedfcb5 100644 --- a/server/server_firebase_test.go +++ b/server/server_firebase_test.go @@ -346,7 +346,7 @@ func TestMaybeTruncateFCMMessage_NotTooLong(t *testing.T) { func TestToFirebaseSender_Abuse(t *testing.T) { sender := &testFirebaseSender{allowed: 2} client := newFirebaseClient(sender, &testAuther{}) - visitor := newVisitor(newTestConfig(t, ""), limit.New(&limit.Config{}), newMemTestCache(t), nil, netip.MustParseAddr("1.2.3.4"), nil) + visitor := newVisitor(newTestConfig(t, ""), limit.New(&limit.Config{}), newSubscriptionGauge(), newMemTestCache(t), nil, netip.MustParseAddr("1.2.3.4"), nil) require.Nil(t, client.Send(visitor, &model.Message{Topic: "mytopic"})) require.Equal(t, 1, len(sender.Messages())) diff --git a/server/server_manager.go b/server/server_manager.go index 1042f831..a0885cba 100644 --- a/server/server_manager.go +++ b/server/server_manager.go @@ -75,8 +75,9 @@ func (s *Server) execManager() { // Print stats s.mu.RLock() - messagesCount, topicsCount, visitorsCount := s.messages, len(s.topics), len(s.visitors) + messagesCount, topicsCount := s.messages, len(s.topics) s.mu.RUnlock() + visitorsCount := s.limiter.Size() // Update stats s.updateAndWriteStats(messagesCount) @@ -110,23 +111,12 @@ func (s *Server) execManager() { } func (s *Server) pruneVisitors() { - staleVisitors := 0 log. Tag(tagManager). Timing(func() { - s.mu.Lock() - defer s.mu.Unlock() - for ip, v := range s.visitors { - if v.Stale() { - log.Tag(tagManager).With(v).Trace("Deleting stale visitor") - delete(s.visitors, ip) - staleVisitors++ - } - } + s.limiter.Prune(visitorExpungeAfter) // Visitors are per request; only their keys' buckets and counts stay }). - Field("stale_visitors", staleVisitors). Debug("Finished deleting stale visitors") - s.limiter.Prune(visitorExpungeAfter) // The keys' buckets and counts go with the visitors } func (s *Server) pruneTokens() { diff --git a/server/server_middleware_test.go b/server/server_middleware_test.go new file mode 100644 index 00000000..c9a95472 --- /dev/null +++ b/server/server_middleware_test.go @@ -0,0 +1,62 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "net/netip" + "testing" + + "github.com/stretchr/testify/require" + "heckel.io/ntfy/v2/user" +) + +func TestServer_EnsureUser_AnonymousNotAuthorizedAsConcurrentUser(t *testing.T) { + // An anonymous request that overlaps with an authenticated request from the same IP must not + // see that request's user, or it could mint a token for the victim's account (#1051) + c := newTestConfigWithAuthFile(t, "") + c.AuthDefault = user.PermissionDenyAll + s := newTestServer(t, c) + require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false)) + u, err := s.userManager.User("phil") + require.Nil(t, err) + ip := netip.MustParseAddr("9.9.9.9") + + anon := s.visitor(ip, nil) + s.visitor(ip, u) // The concurrent authenticated request, made deterministic + + nextCalled := false + next := func(w http.ResponseWriter, r *http.Request, v *visitor) error { + nextCalled = true + return nil + } + r, _ := http.NewRequest("POST", "/v1/account/token", nil) + err = s.ensureUser(next)(httptest.NewRecorder(), r, anon) + require.Equal(t, errHTTPUnauthorized, err) + require.False(t, nextCalled) +} + +func TestServer_AuthorizeTopic_UserSurvivesConcurrentAnonymousRequest(t *testing.T) { + // The reverse direction: an authenticated request must not be authorized as nobody (and be + // rejected under deny-all) because an anonymous request from the same IP overlapped with it + // (#1051, #2113) + c := newTestConfigWithAuthFile(t, "") + c.AuthDefault = user.PermissionDenyAll + s := newTestServer(t, c) + require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false)) + require.Nil(t, s.userManager.AllowAccess("phil", "mytopic", user.PermissionReadWrite)) + u, err := s.userManager.User("phil") + require.Nil(t, err) + ip := netip.MustParseAddr("9.9.9.9") + + authed := s.visitor(ip, u) + s.visitor(ip, nil) // The concurrent anonymous request, made deterministic + + nextCalled := false + next := func(w http.ResponseWriter, r *http.Request, v *visitor) error { + nextCalled = true + return nil + } + r, _ := http.NewRequest("PUT", "/mytopic", nil) + require.Nil(t, s.authorizeTopicWrite(next)(httptest.NewRecorder(), r, authed)) + require.True(t, nextCalled) +} diff --git a/server/server_payments.go b/server/server_payments.go index 7a8c8d2e..dadf20e8 100644 --- a/server/server_payments.go +++ b/server/server_payments.go @@ -211,7 +211,7 @@ func (s *Server) handleAccountBillingSubscriptionCreateSuccess(w http.ResponseWr if err != nil { return err } - v.SetUser(u) + v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user logvr(v, r). With(tier). Tag(tagStripe). @@ -435,7 +435,7 @@ func (s *Server) handleAccountBillingWebhookSubscriptionUpdated(r *http.Request, if err != nil { return err } - v.SetUser(u) + v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user tier, err := s.userManager.TierByStripePrice(priceID) if err != nil { return err @@ -458,7 +458,7 @@ func (s *Server) handleAccountBillingWebhookSubscriptionDeleted(r *http.Request, if err != nil { return err } - v.SetUser(u) + v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user logvr(v, r). Tag(tagStripe). Field("stripe_webhook_type", event.Type). diff --git a/server/server_payments_test.go b/server/server_payments_test.go index 4ddb61da..19bf92e2 100644 --- a/server/server_payments_test.go +++ b/server/server_payments_test.go @@ -12,7 +12,6 @@ import ( "heckel.io/ntfy/v2/user" "heckel.io/ntfy/v2/util" "io" - "net/netip" "os" "path/filepath" "strings" @@ -395,12 +394,10 @@ func TestPayments_Checkout_Success_And_Increase_Rate_Limits_Reset_Visitor(t *tes }) require.Equal(t, 429, rr.Code) - // Now let's test the message limiter by faking a ridiculously generous rate limiter - v := s.visitor(netip.MustParseAddr("9.9.9.9"), u) - v.mu.Lock() - v.limits.RequestLimitBurst = 1000000 - v.limits.RequestLimitReplenish = time.Nanosecond - v.mu.Unlock() + // Now let's test the message limiter by faking a ridiculously generous request limit; the + // tier's request burst never goes below the configured one, and visitors are built per request + s.config.VisitorRequestLimitBurst = 1000000 + s.config.VisitorRequestLimitReplenish = time.Nanosecond var wg sync.WaitGroup for i := 0; i < 209; i++ { diff --git a/server/server_test.go b/server/server_test.go index 4f624a72..f58a3f60 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -3627,25 +3627,25 @@ func TestServer_SubscriberRateLimiting_VisitorExpiration(t *testing.T) { rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, subscriberFn) require.Equal(t, 200, rr.Code) require.Equal(t, "1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.ip.String()) - require.Equal(t, s.visitors["ip:1.2.3.4"], s.topics["upAAAAAAAAAAAA"].rateVisitor) + require.Equal(t, "ip:1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.key) // Publish message, observe rate visitor tokens being decreased + publisher := s.visitor(netip.MustParseAddr("9.9.9.9"), nil) response := request(t, s, "POST", "/upAAAAAAAAAAAA", "some message", nil) require.Equal(t, 200, response.Code) - require.Equal(t, int64(0), s.visitors["ip:9.9.9.9"].Stats().Messages) + require.Equal(t, int64(0), publisher.Stats().Messages) require.Equal(t, int64(1), s.topics["upAAAAAAAAAAAA"].rateVisitor.Stats().Messages) - require.Equal(t, s.visitors["ip:1.2.3.4"], s.topics["upAAAAAAAAAAAA"].rateVisitor) - // Expire visitor - s.visitors["ip:1.2.3.4"].seen = time.Now().Add(-1 * 25 * time.Hour) - s.pruneVisitors() + // Expire visitor: its key is pruned from the limiter after a day without use + s.limiter.Prune(-time.Hour) // Everything is idle for longer than -1h + require.True(t, s.topics["upAAAAAAAAAAAA"].rateVisitor.Stale()) // Publish message again, observe that rateVisitor is not used anymore and is reset response = request(t, s, "POST", "/upAAAAAAAAAAAA", "some message", nil) require.Equal(t, 200, response.Code) - require.Equal(t, int64(1), s.visitors["ip:9.9.9.9"].Stats().Messages) + require.Equal(t, int64(1), publisher.Stats().Messages) require.Nil(t, s.topics["upAAAAAAAAAAAA"].rateVisitor) - require.Nil(t, s.visitors["ip:1.2.3.4"]) + require.Equal(t, int64(0), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages) }) } @@ -5575,7 +5575,7 @@ func TestServer_SubscribeHTTP_NoWriteAfterHandlerReturn(t *testing.T) { // that copied the subscriber list before Unsubscribe ran. In production, this is exactly // how the panic occurs: the goroutine spawned by topic.Publish calls sub() after the // handler has already returned and Go has cleaned up the response writer. - v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.MustParseAddr("9.9.9.9"), nil) + v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.MustParseAddr("9.9.9.9"), nil) msg := model.NewDefaultMessage("mytopic", "straggler message") _ = copiedSub(v, msg) @@ -5595,7 +5595,7 @@ func TestServer_HandleError_SkipsWriteHeaderOnHijackedConnection(t *testing.T) { r, _ := http.NewRequest("GET", "/mytopic/ws", nil) r.Header.Set("Upgrade", "websocket") r.Header.Set("Connection", "Upgrade") - v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.MustParseAddr("1.2.3.4"), nil) + v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.MustParseAddr("1.2.3.4"), nil) // Test post-upgrade errors wrapped with errWebSocketPostUpgrade (should NOT call WriteHeader) postUpgradeErr := &errWebSocketPostUpgrade{errors.New("websocket: close 1000 (normal)")} diff --git a/server/visitor.go b/server/visitor.go index 3a766be8..b0c379e7 100644 --- a/server/visitor.go +++ b/server/visitor.go @@ -3,7 +3,6 @@ package server import ( "net/netip" "sync" - "sync/atomic" "time" "heckel.io/ntfy/v2/limit" @@ -24,27 +23,28 @@ const ( visitorExpungeAfter = oneDay ) -// visitor represents an API user: its identity (the IP, or the user) and the limits that apply to it. -// The counting itself happens in the limit service, keyed by the visitor key, so two visitor objects -// with the same key share one set of buckets and counts. +// visitor represents one request's API user: its identity (the IP, or the user) and the limits that +// apply to it. It is built per request and never changed afterwards, so concurrent requests from the +// same IP cannot see each other's user (#1051). The counting happens in the limit service, keyed by +// key, so two visitors with the same key share one set of buckets and counts. type visitor struct { config *Config limitsConfig *vis.Config // The parts of config that keying and limit resolution need (see visitorConfigFrom) limiter *limit.Service // Decides on and counts usage, keyed by key + subscriptions *subscriptionGauge messageCache *message.Cache - userManager *user.Manager // May be nil - key string // The limiter key: "ip:" or "user:", see vis.Key - ip netip.Addr // Visitor IP address - user *user.User // Only set if authenticated user, otherwise nil - limits *vis.Limits // Resolved from the config or the user's tier; replaced by SetUser when the tier changes - subscriptions *util.FixedLimiter // Gauge of this node's open subscriptions for the visitor; never shared, it protects this node - firebase time.Time // Next allowed Firebase message - seen time.Time // Last seen time of this visitor (needed for removal of stale visitors) - localMessages atomic.Int64 // Messages this visitor object admitted; the user stats deltas are measured on these, not on the shared counts (see enqueueUserStatsDeltaNoLock) - localEmails atomic.Int64 // Emails this visitor object admitted - localCalls atomic.Int64 // Calls this visitor object admitted - statsEnqueued user.Stats // The local counts already handed to the user manager's write queue - mu sync.RWMutex + userManager *user.Manager // May be nil + key string // The limiter key: "ip:" or "user:", see vis.Key + ip netip.Addr // Visitor IP address + user *user.User // Only set if authenticated user, otherwise nil + limits *vis.Limits // Resolved from the config or the user's tier +} + +// subscriptionGauge counts the open subscriptions per visitor key on this node. It is never shared +// with other nodes: it protects this node's connections. +type subscriptionGauge struct { + counts map[string]int64 + mu sync.Mutex // Protects counts } type visitorInfo struct { @@ -65,35 +65,68 @@ type visitorStats struct { AttachmentTotalSizeRemaining int64 } -func newVisitor(conf *Config, limiter *limit.Service, messageCache *message.Cache, userManager *user.Manager, ip netip.Addr, user *user.User) *visitor { +func newVisitor(conf *Config, limiter *limit.Service, subscriptions *subscriptionGauge, messageCache *message.Cache, userManager *user.Manager, ip netip.Addr, u *user.User) *visitor { + limitsConfig := visitorConfigFrom(conf) + var tier *user.Tier + if u != nil { + tier = u.Tier // May be nil, which resolves the config-based limits + } v := &visitor{ config: conf, - limitsConfig: visitorConfigFrom(conf), + limitsConfig: limitsConfig, limiter: limiter, + subscriptions: subscriptions, messageCache: messageCache, userManager: userManager, // May be nil + key: vis.Key(ip, u, limitsConfig), ip: ip, - user: user, - firebase: time.Unix(0, 0), - seen: time.Now(), - subscriptions: util.NewFixedLimiter(int64(conf.VisitorSubscriptionLimit)), + user: u, + limits: vis.LimitsFor(limitsConfig, tier), + } + if tier != nil { + // Carry the persisted stats into the key's daily counts, if the key is new to the limiter + var usage limit.Usage + usage.Set(limit.KindMessages, u.Stats.Messages) + usage.Set(limit.KindEmails, u.Stats.Emails) + usage.Set(limit.KindCalls, u.Stats.Calls) + limiter.Seed(v.key, usage) } - v.resetLimitsNoLock() return v } -func (v *visitor) Context() log.Context { - v.mu.RLock() - defer v.mu.RUnlock() - return v.contextNoLock() +func newSubscriptionGauge() *subscriptionGauge { + return &subscriptionGauge{ + counts: make(map[string]int64), + } } -func (v *visitor) contextNoLock() log.Context { - info := v.infoLightNoLock() +// Acquire counts one more subscription for the key, unless the key already has limit of them +func (g *subscriptionGauge) Acquire(key string, limit int64) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.counts[key] >= limit { + return false + } + g.counts[key]++ + return true +} + +// Release counts one subscription less for the key, and forgets the key when it has none +func (g *subscriptionGauge) Release(key string) { + g.mu.Lock() + defer g.mu.Unlock() + g.counts[key]-- + if g.counts[key] <= 0 { + delete(g.counts, key) + } +} + +func (v *visitor) Context() log.Context { + info := v.infoLight() fields := log.Context{ "visitor_id": v.key, "visitor_ip": v.ip.String(), - "visitor_seen": util.FormatTime(v.seen), + "visitor_seen": util.FormatTime(v.limiter.Seen(v.key)), "visitor_messages": info.Stats.Messages, "visitor_messages_limit": info.Limits.MessageLimit, "visitor_messages_remaining": info.Stats.MessagesRemaining, @@ -146,76 +179,85 @@ func visitorExtendedInfoContext(info *visitorInfo) log.Context { } func (v *visitor) RequestAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.RequestLimitBurst <= 0 { return false // A bucket of nothing admits nothing, as before } return v.limiter.Allow(v.key, v.requestRule(), 1).OK } +// FirebaseAllowed returns false while Firebase sends are paused for the key (see FirebaseTemporarilyDeny) func (v *visitor) FirebaseAllowed() bool { - v.mu.RLock() - defer v.mu.RUnlock() - return !time.Now().Before(v.firebase) + rule := v.firebaseRule() + if rule.Refill <= 0 { + return true + } + return v.limiter.Tokens(v.key, rule) >= 1 } +// FirebaseTemporarilyDeny pauses Firebase sends for the key for the configured penalty duration func (v *visitor) FirebaseTemporarilyDeny() { - v.mu.Lock() - defer v.mu.Unlock() - v.firebase = time.Now().Add(v.config.FirebaseQuotaExceededPenaltyDuration) + rule := v.firebaseRule() + if rule.Refill <= 0 { + return + } + v.limiter.Allow(v.key, rule, 1) } func (v *visitor) MessageAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.MessageLimit <= 0 { return false // A daily limit of zero admits nothing; the limit service reads it as "no cap" } allowed := v.limiter.Allow(v.key, v.messageRule(), 1).OK if allowed { - v.localMessages.Add(1) + v.enqueueUserStats(&user.Stats{ + Messages: 1, + }) } return allowed } func (v *visitor) EmailAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.EmailLimitBurst <= 0 { return false // No email bucket means no emails, as before } allowed := v.limiter.Allow(v.key, v.emailRule(), 1).OK if allowed { - v.localEmails.Add(1) + v.enqueueUserStats(&user.Stats{ + Emails: 1, + }) } return allowed } func (v *visitor) CallAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.CallLimit <= 0 { return false // A daily limit of zero admits nothing (e.g. a tier without calls, verified number or not) } allowed := v.limiter.Allow(v.key, v.callRule(), 1).OK if allowed { - v.localCalls.Add(1) + v.enqueueUserStats(&user.Stats{ + Calls: 1, + }) } return allowed } +// enqueueUserStats hands what a tier user's request admitted to the user manager's write queue, which +// adds it to the persisted stats (deltas add up across requests and cluster nodes) +func (v *visitor) enqueueUserStats(delta *user.Stats) { + if v.userManager == nil || v.user == nil || v.user.Tier == nil { + return + } + v.userManager.EnqueueUserStats(v.user.ID, delta) +} + func (v *visitor) SubscriptionAllowed() bool { - v.mu.RLock() - defer v.mu.RUnlock() - return v.subscriptions.Allow() + return v.subscriptions.Acquire(v.key, int64(v.config.VisitorSubscriptionLimit)) } // TopicCreationAllowed returns true if the visitor is allowed to cause a new topic to be // inserted into the server's in-memory topic map. Returns true if no limiter is configured. func (v *visitor) TopicCreationAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if !v.topicCreationLimited() { return true } @@ -224,8 +266,6 @@ func (v *visitor) TopicCreationAllowed() bool { // AuthAllowed returns true if an auth request can be attempted (a token is available) func (v *visitor) AuthAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.user != nil { return true // Users are already logged in, no need to limit requests } @@ -234,8 +274,6 @@ func (v *visitor) AuthAllowed() bool { // AuthFailed records an auth failure func (v *visitor) AuthFailed() { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.user == nil { v.limiter.Allow(v.key, v.authRule(), 1) } @@ -244,8 +282,6 @@ func (v *visitor) AuthFailed() { // AccountActionAllowed returns true if a rate-limited account action (signup or password-reset // request) is currently allowed for this visitor func (v *visitor) AccountActionAllowed() bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.user != nil { return false // Users cannot create accounts when logged in } @@ -255,16 +291,12 @@ func (v *visitor) AccountActionAllowed() bool { // AccountActionPerformed decreases the account limiter. This is to be called after a rate-limited // account action (signup or password-reset request). func (v *visitor) AccountActionPerformed() { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.user == nil { v.limiter.Allow(v.key, v.accountRule(), 1) } } func (v *visitor) BandwidthAllowed(bytes int64) bool { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.AttachmentBandwidthLimit <= 0 { return false // A bandwidth limit of zero admits nothing, as before } @@ -272,41 +304,29 @@ func (v *visitor) BandwidthAllowed(bytes int64) bool { } func (v *visitor) RemoveSubscription() { - v.mu.RLock() - defer v.mu.RUnlock() - v.subscriptions.AllowN(-1) + v.subscriptions.Release(v.key) } +// Keepalive marks the visitor's key as used, so a long-lived subscription keeps its limiter state func (v *visitor) Keepalive() { - v.mu.Lock() - defer v.mu.Unlock() - v.seen = time.Now() + v.limiter.Touch(v.key) } // BandwidthLimiter returns the visitor's bandwidth bucket as a util.Limiter, for the readers and // writers that meter bytes. A revert on the limiter (a negative AllowN) refunds the bytes. func (v *visitor) BandwidthLimiter() util.Limiter { - v.mu.RLock() // limits could be replaced! - defer v.mu.RUnlock() if v.limits.AttachmentBandwidthLimit <= 0 { return util.NewFixedLimiter(0) // A bandwidth limit of zero admits nothing, as BandwidthAllowed } return v.limiter.Limiter(v.key, v.bandwidthRule()) } +// Stale returns true if the visitor's key has not been used for visitorExpungeAfter func (v *visitor) Stale() bool { - v.mu.RLock() - defer v.mu.RUnlock() - return time.Since(v.seen) > visitorExpungeAfter + return time.Since(v.limiter.Seen(v.key)) > visitorExpungeAfter } func (v *visitor) Stats() *user.Stats { - v.mu.RLock() - defer v.mu.RUnlock() - return v.statsNoLock() -} - -func (v *visitor) statsNoLock() *user.Stats { usage := v.limiter.Usage(v.key) return &user.Stats{ Messages: usage.Count(limit.KindMessages), @@ -315,143 +335,36 @@ func (v *visitor) statsNoLock() *user.Stats { } } -// ResetStats zeroes the visitor's local counts and with them the enqueued seeds its deltas are -// measured against, at the daily rollover, because the caller zeroes the user's database row and -// the limiter's counts in the same pass (in a cluster that row is reset once, by the leader). -// Only the stats resetter calls this. -func (v *visitor) ResetStats() { - v.mu.Lock() // Also protects statsEnqueued - defer v.mu.Unlock() - v.resetLocalStatsNoLock() // The database row is zeroed by the (leader's) daily reset -} - -// EnqueueUserStatsDelta enqueues the visitor's not-yet-persisted stats increments to the user -// manager's write queue. Deltas add up across cluster nodes (see Manager.EnqueueUserStats). -func (v *visitor) EnqueueUserStatsDelta() { - v.mu.Lock() - defer v.mu.Unlock() - v.enqueueUserStatsDeltaNoLock() -} - -// enqueueUserStatsDeltaNoLock measures the deltas on what this visitor object admitted, not on -// the limiter's counts: those are shared by every visitor object with the same key, and would be -// enqueued by each of them. The caller must hold v.mu. -func (v *visitor) enqueueUserStatsDeltaNoLock() { - u := v.user // Pinned for the whole function: SetUser(nil) must not turn this into a nil deref - if v.userManager == nil || u == nil { - return - } - current := user.Stats{ - Messages: v.localMessages.Load(), - Emails: v.localEmails.Load(), - Calls: v.localCalls.Load(), - } - delta := user.Stats{ - Messages: current.Messages - v.statsEnqueued.Messages, - Emails: current.Emails - v.statsEnqueued.Emails, - Calls: current.Calls - v.statsEnqueued.Calls, - } - if delta == (user.Stats{}) { - return - } - v.statsEnqueued = current - v.userManager.EnqueueUserStats(u.ID, &delta) -} - -// resetLocalStatsNoLock zeroes the local counts and what was enqueued of them. The caller must hold v.mu. -func (v *visitor) resetLocalStatsNoLock() { - v.localMessages.Store(0) - v.localEmails.Store(0) - v.localCalls.Store(0) - v.statsEnqueued = user.Stats{} -} - // User returns the visitor user, or nil if there is none func (v *visitor) User() *user.User { - v.mu.RLock() - defer v.mu.RUnlock() return v.user // May be nil } // IP returns the visitor IP address func (v *visitor) IP() netip.Addr { - v.mu.RLock() - defer v.mu.RUnlock() return v.ip } // Authenticated returns true if a user successfully authenticated func (v *visitor) Authenticated() bool { - v.mu.RLock() - defer v.mu.RUnlock() return v.user != nil } -// SetUser sets the visitors user to the given value. A changed tier (gained, lost, or a tier -// with other limits) re-resolves the key and the limits; the next call to the limiter uses them, -// with the buckets as they are: nothing is handed back. -func (v *visitor) SetUser(u *user.User) { - v.mu.Lock() - defer v.mu.Unlock() - shouldResetLimits := v.user.TierID() != u.TierID() // TierID works with nil receiver - if shouldResetLimits && v.user != nil && v.user.Tier != nil { - // Flush increments counted under the old key, under the old account, before the key is - // replaced below; the new key is re-seeded from the new user's persisted stats. Usage of - // an IP-keyed visitor (no user, or no tier) is never persisted, so it must not land on - // the new account. - v.enqueueUserStatsDeltaNoLock() - } - tierChanged := shouldResetLimits || (u != nil && u.Tier != nil && *v.user.Tier != *u.Tier) - v.user = u // u may be nil! - if tierChanged { - v.resetLimitsNoLock() - } -} - // MaybeUserID returns the user ID of the visitor (if any). If this is an anonymous visitor, // an empty string is returned. func (v *visitor) MaybeUserID() string { - v.mu.RLock() - defer v.mu.RUnlock() if v.user != nil { return v.user.ID } return "" } -// resetLimitsNoLock re-resolves the key and the limits from the current user, and seeds the key's -// daily counts from the user's persisted stats if the key is new to the limiter. The caller must -// hold v.mu. -func (v *visitor) resetLimitsNoLock() { - v.key = vis.Key(v.ip, v.user, v.limitsConfig) - v.limits = v.limitsNoLock() - if v.user != nil && v.user.Tier != nil { - var usage limit.Usage - usage.Set(limit.KindMessages, v.user.Stats.Messages) - usage.Set(limit.KindEmails, v.user.Stats.Emails) - usage.Set(limit.KindCalls, v.user.Stats.Calls) - v.limiter.Seed(v.key, usage) - } - v.resetLocalStatsNoLock() // What was admitted so far belongs to the previous key and user, and was flushed by SetUser if it was owed - log.Fields(v.contextNoLock()).Debug("Rate limits reset for visitor") // Must be after function, because contextNoLock() describes rate limits -} - func (v *visitor) Limits() *vis.Limits { - v.mu.RLock() - defer v.mu.RUnlock() return v.limits } -func (v *visitor) limitsNoLock() *vis.Limits { - var tier *user.Tier - if v.user != nil { - tier = v.user.Tier // May be nil, which resolves the config-based limits - } - return vis.LimitsFor(v.limitsConfig, tier) -} - // The rules the visitor's limits translate to, built per call (they are small values); the -// limiter keeps the state. The caller must hold v.mu. +// limiter keeps the state. func (v *visitor) requestRule() limit.Rule { return limit.Rule{ @@ -524,10 +437,17 @@ func (v *visitor) authRule() limit.Rule { } } +// firebaseRule is a single token that takes the penalty duration to come back: taking it pauses sends +func (v *visitor) firebaseRule() limit.Rule { + return limit.Rule{ + Kind: limit.KindFirebase, + Burst: 1, + Refill: v.config.FirebaseQuotaExceededPenaltyDuration, + } +} + func (v *visitor) Info() (*visitorInfo, error) { - v.mu.RLock() - info := v.infoLightNoLock() - v.mu.RUnlock() + info := v.infoLight() // Attachment stats from database var attachmentsBytesUsed int64 @@ -558,8 +478,8 @@ func (v *visitor) Info() (*visitorInfo, error) { return info, nil } -func (v *visitor) infoLightNoLock() *visitorInfo { - stats := v.statsNoLock() +func (v *visitor) infoLight() *visitorInfo { + stats := v.Stats() limits := v.limits return &visitorInfo{ Limits: limits, @@ -567,16 +487,16 @@ func (v *visitor) infoLightNoLock() *visitorInfo { Messages: stats.Messages, MessagesRemaining: zeroIfNegative(limits.MessageLimit - stats.Messages), Emails: stats.Emails, - EmailsRemaining: v.emailsRemainingNoLock(stats.Emails), + EmailsRemaining: v.emailsRemaining(stats.Emails), Calls: stats.Calls, CallsRemaining: zeroIfNegative(limits.CallLimit - stats.Calls), }, } } -// emailsRemainingNoLock returns what the visitor can send now: the day's remaining emails, but no +// emailsRemaining returns what the visitor can send now: the day's remaining emails, but no // more than the email bucket holds (it may be empty while the day has room, e.g. after the reset) -func (v *visitor) emailsRemainingNoLock(emails int64) int64 { +func (v *visitor) emailsRemaining(emails int64) int64 { remaining := zeroIfNegative(v.limits.EmailLimit - emails) if rule := v.emailRule(); rule.Burst > 0 { remaining = min(remaining, v.limiter.Tokens(v.key, rule)) diff --git a/server/visitor_test.go b/server/visitor_test.go index d8241cae..87138d12 100644 --- a/server/visitor_test.go +++ b/server/visitor_test.go @@ -40,9 +40,9 @@ func TestVisitor_UserStatsDeltaFlush(t *testing.T) { }) } -func TestVisitor_SetUser_DoesNotChargeAnonymousUsage(t *testing.T) { - // A user who gets a tier takes over their IP-keyed visitor (e.g. the billing callback). What - // that visitor counted anonymously must not land on the account. +func TestVisitor_TierUserDoesNotGetAnonymousUsage(t *testing.T) { + // What a user's IP counted anonymously must not land on the account once the user (with a + // tier) shows up from that IP. c := newTestConfigWithAuthFile(t, "") c.AuthStatsQueueWriterInterval = 100 * time.Millisecond s := newTestServer(t, c) @@ -56,7 +56,7 @@ func TestVisitor_SetUser_DoesNotChargeAnonymousUsage(t *testing.T) { require.True(t, v.MessageAllowed()) } - v.SetUser(u) + require.True(t, s.visitor(netip.MustParseAddr("9.9.9.9"), u).RequestAllowed()) time.Sleep(500 * time.Millisecond) u, err = s.userManager.User("phil") require.Nil(t, err) @@ -229,3 +229,45 @@ func TestVisitor_ZeroLimitsAdmitNothing(t *testing.T) { require.False(t, v.BandwidthLimiter().AllowN(1)) }) } + +func TestVisitor_UserIsPerRequest(t *testing.T) { + // Untiered users and anonymous requests from one IP share a limiter key, but each request + // must keep its own user (#1051) + s := newTestServer(t, newTestConfigWithAuthFile(t, "")) + require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false)) + u, err := s.userManager.User("phil") + require.Nil(t, err) + ip := netip.MustParseAddr("9.9.9.9") + + // The concurrent requests, made deterministic + authed := s.visitor(ip, u) + anon := s.visitor(ip, nil) + require.NotNil(t, authed.User()) + require.Equal(t, u.ID, authed.User().ID) + require.Nil(t, anon.User()) + authedAgain := s.visitor(ip, u) + require.Nil(t, anon.User()) + require.NotNil(t, authedAgain.User()) + require.Equal(t, u.ID, authedAgain.User().ID) + + // Limits are still shared per IP: requests and the node-local subscription gauge + for authed.RequestAllowed() { + } + require.False(t, anon.RequestAllowed()) + for authed.SubscriptionAllowed() { + } + require.False(t, anon.SubscriptionAllowed()) + authed.RemoveSubscription() + require.True(t, anon.SubscriptionAllowed()) +} + +func TestVisitor_FirebasePenaltyIsSharedPerKey(t *testing.T) { + s := newTestServer(t, newTestConfig(t, "")) + ip := netip.MustParseAddr("9.9.9.9") + v1 := s.visitor(ip, nil) + v2 := s.visitor(ip, nil) + require.True(t, v2.FirebaseAllowed()) + v1.FirebaseTemporarilyDeny() + require.False(t, v2.FirebaseAllowed()) + require.True(t, s.visitor(netip.MustParseAddr("1.2.3.4"), nil).FirebaseAllowed()) +}