From 0625ce120fd80fa5b3aad817a4e502f8e692070c Mon Sep 17 00:00:00 2001 From: Catie Date: Thu, 10 Sep 2026 17:36:42 +0800 Subject: [PATCH] fix: expire the idempotency cache, sweep rate limit buckets, scope the model list Three defects found while reviewing the group work: - idempotencyCache never expired. CachedResponse.CreatedAt was written but never read, so a reused Idempotency-Key replayed its original body forever and the cache (holding whole response bodies) grew without bound. - rateLimitBuckets accumulated one entry per key per active minute for the lifetime of the process; pruneOperationalHistoryLocked only covered logs and the quota ledger. - /v1/models and /api/catalog/models ignored group visibility, advertising models whose only channels the caller's group cannot reach. A user would see the model and then get model_not_available on the call. The catalog stays fully visible to anonymous visitors; signed-in callers see only what their group can route to. The new check avoids channelCandidatesLocked because that expands key pools and decrypts secrets per model. Also adds a race-detector note: it needs cgo, which is unavailable on this box. Co-Authored-By: Claude Code --- cmd/capi/main.go | 128 +++++++++++++++++++++++++++++++--- cmd/capi/main_test.go | 155 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 275 insertions(+), 8 deletions(-) diff --git a/cmd/capi/main.go b/cmd/capi/main.go index 0761ae3..b8cd67e 100644 --- a/cmd/capi/main.go +++ b/cmd/capi/main.go @@ -67,6 +67,12 @@ var ( var imageJSONKeepaliveInterval = 8 * time.Second +// idempotencyCacheTTL bounds how long a client-supplied Idempotency-Key replays +// its original response. Without an expiry a key could be reused indefinitely +// and would return a stale body, and the cache (which holds whole response +// bodies) would grow without bound. +var idempotencyCacheTTL = 24 * time.Hour + type AppState struct { Users []User `json:"users"` Groups []UserGroup `json:"groups"` @@ -420,6 +426,7 @@ type Server struct { sessionTTL time.Duration accountHealthInterval time.Duration rateLimitBuckets map[string]int + rateLimitMinute int64 idempotencyCache map[string]CachedResponse authStates map[string]time.Time sessions map[string]Session @@ -2225,13 +2232,33 @@ func (s *Server) updateAccountProfile(c *gin.Context) { } func (s *Server) publicModelCatalog(c *gin.Context) { + // Resolve the session before taking the lock: sessionFromRequest locks s.mu + // itself, and s.mu is not reentrant. + session, hasSession := s.sessionFromRequest(c) s.mu.Lock() defer s.mu.Unlock() + + // Signed-in callers only see models their group can actually reach, so the + // catalog cannot advertise something that would fail with + // model_not_available at call time. Anonymous visitors see the full catalog. + groupID := "" + filterByGroup := false + if hasSession { + if user := s.findUser(session.UserID); user != nil { + groupID = user.GroupID + filterByGroup = true + } + } + models := []Model{} for _, model := range s.state.Models { - if model.Status == "available" { - models = append(models, model) + if model.Status != "available" { + continue + } + if filterByGroup && !s.channelServesModelForGroupLocked(model.ID, groupID) { + continue } + models = append(models, model) } c.JSON(http.StatusOK, gin.H{"models": models}) } @@ -4142,9 +4169,20 @@ func (s *Server) openAIModels(c *gin.Context) { } data := []gin.H{} for _, model := range s.state.Models { - if model.Status == "available" && (auth == nil || apiKeyAllowsModel(auth.Key, model.ID)) { - data = append(data, toOpenAIModel(model)) + if model.Status != "available" { + continue } + if auth != nil { + if !apiKeyAllowsModel(auth.Key, model.ID) { + continue + } + // Only advertise models the caller's group can actually route to; + // otherwise the listing promises a model that fails at call time. + if !s.channelServesModelForGroupLocked(model.ID, auth.User.GroupID) { + continue + } + } + data = append(data, toOpenAIModel(model)) } c.JSON(http.StatusOK, gin.H{"object": "list", "data": data}) } @@ -4526,7 +4564,7 @@ func (s *Server) chatCompletions(c *gin.Context) { s.mu.Lock() if idempotencyKey != "" { - if cached, ok := s.idempotencyCache[idempotencyKey]; ok { + if cached, ok := s.cachedResponseLocked(idempotencyKey); ok { s.mu.Unlock() c.Header("x-capi-cache", "idempotency") c.JSON(cached.Status, cached.Body) @@ -4550,7 +4588,7 @@ func (s *Server) imageGenerations(c *gin.Context) { s.mu.Lock() if idempotencyKey != "" { - if cached, ok := s.idempotencyCache[idempotencyKey]; ok { + if cached, ok := s.cachedResponseLocked(idempotencyKey); ok { s.mu.Unlock() c.JSON(cached.Status, cached.Body) return @@ -4572,7 +4610,7 @@ func (s *Server) imageEdits(c *gin.Context) { s.mu.Lock() if idempotencyKey != "" { - if cached, ok := s.idempotencyCache[idempotencyKey]; ok { + if cached, ok := s.cachedResponseLocked(idempotencyKey); ok { s.mu.Unlock() c.JSON(cached.Status, cached.Body) return @@ -9338,6 +9376,27 @@ func channelAllowsUserGroup(channel Channel, groupID string) bool { return containsString(channel.AllowedGroupIDs, strings.TrimSpace(groupID)) } +// channelServesModelForGroupLocked reports whether any enabled channel that is +// visible to the group actually serves the model. It answers the same question +// as channelCandidatesLocked but without expanding key pools or decrypting +// secrets, so it is cheap enough to run across an entire model catalog. +// Callers must hold s.mu. +func (s *Server) channelServesModelForGroupLocked(modelID, groupID string) bool { + for i := range s.state.Channels { + channel := &s.state.Channels[i] + if channel.Status == "disabled" { + continue + } + if !channelAllowsUserGroup(*channel, groupID) { + continue + } + if containsString(channel.Models, modelID) { + return true + } + } + return false +} + func (s *Server) findAPIKeyByID(id string) *APIKey { for i := range s.state.APIKeys { if s.state.APIKeys[i].ID == id { @@ -9952,8 +10011,33 @@ func isTokenInvalidatedProviderError(providerErr *ProviderError) bool { strings.Contains(text, "try signing in again") } +// cachedResponseLocked returns a cached idempotent response that is still +// within its TTL. Expired entries are dropped so a reused Idempotency-Key +// cannot replay an indefinitely old body, and so the cache stays bounded. +// Callers must hold s.mu. +func (s *Server) cachedResponseLocked(key string) (CachedResponse, bool) { + entry, ok := s.idempotencyCache[key] + if !ok { + return CachedResponse{}, false + } + createdAt, err := time.Parse(time.RFC3339Nano, entry.CreatedAt) + if err != nil || time.Since(createdAt) > idempotencyCacheTTL { + delete(s.idempotencyCache, key) + return CachedResponse{}, false + } + return entry, true +} + func (s *Server) checkRateLimitLocked(key *APIKey) bool { - bucket := fmt.Sprintf("%s:%d", key.ID, time.Now().Unix()/60) + minute := time.Now().Unix() / 60 + // Buckets are keyed per key per minute and would otherwise accumulate one + // entry per key per active minute for the lifetime of the process. Sweeping + // once per minute keeps the map bounded to the current window. + if minute != s.rateLimitMinute { + s.rateLimitMinute = minute + s.sweepRateLimitBucketsLocked(minute) + } + bucket := fmt.Sprintf("%s:%d", key.ID, minute) current := s.rateLimitBuckets[bucket] limit := s.requestLimitPerMinute if key.RateLimitPerMinute > 0 { @@ -9966,6 +10050,29 @@ func (s *Server) checkRateLimitLocked(key *APIKey) bool { return true } +// sweepRateLimitBucketsLocked drops every bucket that does not belong to the +// given minute. Callers must hold s.mu. +func (s *Server) sweepRateLimitBucketsLocked(minute int64) { + current := fmt.Sprintf(":%d", minute) + for bucket := range s.rateLimitBuckets { + if !strings.HasSuffix(bucket, current) { + delete(s.rateLimitBuckets, bucket) + } + } +} + +// pruneExpiredIdempotencyLocked evicts idempotency entries past their TTL. +// Unlike the operator-visible history this is in-memory only, so callers must +// not treat a change here as a reason to persist state. Callers must hold s.mu. +func (s *Server) pruneExpiredIdempotencyLocked() { + for key, entry := range s.idempotencyCache { + createdAt, err := time.Parse(time.RFC3339Nano, entry.CreatedAt) + if err != nil || time.Since(createdAt) > idempotencyCacheTTL { + delete(s.idempotencyCache, key) + } + } +} + func (s *Server) loadState() { if s.persistence == "postgres" { s.loadPostgresState() @@ -10309,6 +10416,11 @@ func (s *Server) pruneOperationalHistoryLocked() bool { s.state.QuotaLedger = append([]QuotaEntry{}, s.state.QuotaLedger[len(s.state.QuotaLedger)-settings.MaxQuotaEntries:]...) changed = true } + + // These caches are in-memory only, so evicting from them must not mark the + // state as changed and trigger a write. + s.pruneExpiredIdempotencyLocked() + s.sweepRateLimitBucketsLocked(time.Now().Unix() / 60) return changed } diff --git a/cmd/capi/main_test.go b/cmd/capi/main_test.go index 7b0f261..8c7f0da 100644 --- a/cmd/capi/main_test.go +++ b/cmd/capi/main_test.go @@ -5044,3 +5044,158 @@ func TestDefaultRegistrationGroupAppliesToNewUsers(t *testing.T) { t.Fatalf("new user group = %q, want %q", registered.GroupID, groupID) } } + +func TestIdempotencyCacheExpires(t *testing.T) { + withEnv(t, map[string]string{"PERSISTENCE": "memory"}) + server, _ := testServerRouter(t) + + previousTTL := idempotencyCacheTTL + idempotencyCacheTTL = time.Hour + t.Cleanup(func() { idempotencyCacheTTL = previousTTL }) + + server.mu.Lock() + server.idempotencyCache["fresh"] = CachedResponse{Status: http.StatusOK, Body: gin.H{"ok": true}, CreatedAt: now()} + server.idempotencyCache["stale"] = CachedResponse{ + Status: http.StatusOK, + Body: gin.H{"ok": true}, + CreatedAt: time.Now().Add(-2 * time.Hour).UTC().Format(time.RFC3339Nano), + } + server.idempotencyCache["unparsable"] = CachedResponse{Status: http.StatusOK, Body: gin.H{"ok": true}, CreatedAt: "not-a-timestamp"} + + if _, ok := server.cachedResponseLocked("fresh"); !ok { + t.Fatal("a fresh idempotency entry was rejected") + } + if _, ok := server.cachedResponseLocked("stale"); ok { + t.Fatal("an expired idempotency entry was replayed") + } + if _, ok := server.cachedResponseLocked("unparsable"); ok { + t.Fatal("an entry with an unparsable timestamp was replayed") + } + + // Expired entries must be evicted, not merely skipped. + if _, exists := server.idempotencyCache["stale"]; exists { + t.Fatal("expired entry was not evicted") + } + if _, exists := server.idempotencyCache["unparsable"]; exists { + t.Fatal("unparsable entry was not evicted") + } + + server.idempotencyCache["stale2"] = CachedResponse{ + Status: http.StatusOK, + Body: gin.H{"ok": true}, + CreatedAt: time.Now().Add(-72 * time.Hour).UTC().Format(time.RFC3339Nano), + } + server.pruneExpiredIdempotencyLocked() + if len(server.idempotencyCache) != 1 { + t.Fatalf("cache after prune = %#v, want only the fresh entry", server.idempotencyCache) + } + server.mu.Unlock() +} + +func TestRateLimitBucketsAreSweptPerMinute(t *testing.T) { + withEnv(t, map[string]string{"PERSISTENCE": "memory"}) + server, _ := testServerRouter(t) + + minute := time.Now().Unix() / 60 + server.mu.Lock() + server.rateLimitBuckets[fmt.Sprintf("key_a:%d", minute)] = 3 + server.rateLimitBuckets[fmt.Sprintf("key_b:%d", minute-5)] = 9 + server.rateLimitBuckets[fmt.Sprintf("key_c:%d", minute-1)] = 1 + + key := &APIKey{ID: "key_a", RateLimitPerMinute: 10} + if !server.checkRateLimitLocked(key) { + t.Fatal("request under the limit was rejected") + } + + if len(server.rateLimitBuckets) != 1 { + t.Fatalf("buckets after sweep = %#v, want only the current minute", server.rateLimitBuckets) + } + if _, ok := server.rateLimitBuckets[fmt.Sprintf("key_a:%d", minute)]; !ok { + t.Fatalf("the current minute bucket was dropped: %#v", server.rateLimitBuckets) + } + if server.rateLimitBuckets[fmt.Sprintf("key_a:%d", minute)] != 4 { + t.Fatalf("current bucket count = %d, want the incremented value", server.rateLimitBuckets[fmt.Sprintf("key_a:%d", minute)]) + } + server.mu.Unlock() +} + +func TestOpenAIModelListHidesModelsTheGroupCannotReach(t *testing.T) { + withEnv(t, map[string]string{"PERSISTENCE": "memory"}) + server, router := testServerRouter(t) + seedGatewayFixtures(server) + + auth := map[string]string{"Authorization": "Bearer cat_fixture_live_secret"} + before := perform(router, http.MethodGet, "/v1/models", "", auth) + if before.Code != http.StatusOK || !bytes.Contains(before.Body.Bytes(), []byte(`"id":"deepseek-v4"`)) { + t.Fatalf("baseline model list = %d body = %s", before.Code, before.Body.String()) + } + + created := perform(router, http.MethodPost, "/api/groups", `{"name":"gated"}`, nil) + var groupPayload struct { + Group UserGroup `json:"group"` + } + if err := json.Unmarshal(created.Body.Bytes(), &groupPayload); err != nil { + t.Fatalf("decode group: %v", err) + } + // chn_1002 is the only channel serving deepseek-v4; gate it behind a group + // the calling user is not in. + restricted := perform(router, http.MethodPatch, "/api/channels/chn_1002", `{"allowedGroupIds":["`+groupPayload.Group.ID+`"]}`, nil) + if restricted.Code != http.StatusOK { + t.Fatalf("restrict channel status = %d body = %s", restricted.Code, restricted.Body.String()) + } + + after := perform(router, http.MethodGet, "/v1/models", "", auth) + if after.Code != http.StatusOK { + t.Fatalf("model list status = %d body = %s", after.Code, after.Body.String()) + } + if bytes.Contains(after.Body.Bytes(), []byte(`"id":"deepseek-v4"`)) { + t.Fatalf("a model the group cannot route to is still advertised: %s", after.Body.String()) + } + if !bytes.Contains(after.Body.Bytes(), []byte(`"id":"gpt-5.5"`)) { + t.Fatalf("a model on an unrestricted channel disappeared: %s", after.Body.String()) + } +} + +func TestPublicCatalogFiltersBySessionGroup(t *testing.T) { + withEnv(t, map[string]string{"PERSISTENCE": "memory"}) + server, router := testServerRouter(t) + seedGatewayFixtures(server) + + // An anonymous visitor sees the whole catalog. + anonymous := perform(router, http.MethodGet, "/api/catalog/models", "", nil) + if anonymous.Code != http.StatusOK || !bytes.Contains(anonymous.Body.Bytes(), []byte(`"id":"deepseek-v4"`)) { + t.Fatalf("anonymous catalog = %d body = %s", anonymous.Code, anonymous.Body.String()) + } + + server.mu.Lock() + server.state.Users = append(server.state.Users, User{ID: "usr_cat", Name: "Catalog User", Role: "user", Status: "active"}) + account := Account{ID: "acct_cat", UserID: "usr_cat", Username: "catalog", Role: "user", Status: "active"} + server.state.Accounts = append(server.state.Accounts, account) + server.mu.Unlock() + + created := perform(router, http.MethodPost, "/api/groups", `{"name":"catalog_gated"}`, nil) + var groupPayload struct { + Group UserGroup `json:"group"` + } + if err := json.Unmarshal(created.Body.Bytes(), &groupPayload); err != nil { + t.Fatalf("decode group: %v", err) + } + restricted := perform(router, http.MethodPatch, "/api/channels/chn_1002", `{"allowedGroupIds":["`+groupPayload.Group.ID+`"]}`, nil) + if restricted.Code != http.StatusOK { + t.Fatalf("restrict channel status = %d body = %s", restricted.Code, restricted.Body.String()) + } + + session := server.createAccountSession(account, "Catalog User", "password") + headers := map[string]string{"Cookie": "capi_session=" + session.ID} + + signedIn := perform(router, http.MethodGet, "/api/catalog/models", "", headers) + if signedIn.Code != http.StatusOK { + t.Fatalf("signed-in catalog status = %d body = %s", signedIn.Code, signedIn.Body.String()) + } + if bytes.Contains(signedIn.Body.Bytes(), []byte(`"id":"deepseek-v4"`)) { + t.Fatalf("catalog advertised an unreachable model: %s", signedIn.Body.String()) + } + if !bytes.Contains(signedIn.Body.Bytes(), []byte(`"id":"gpt-5.5"`)) { + t.Fatalf("catalog dropped a reachable model: %s", signedIn.Body.String()) + } +}