fix: 幂等缓存过期 + 限流桶清理 + 模型列表按分组过滤

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
 }
 

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