diff --git a/backend/internal/handler/admin/user_handler.go b/backend/internal/handler/admin/user_handler.go index 72f8f36f57..534f774e88 100644 --- a/backend/internal/handler/admin/user_handler.go +++ b/backend/internal/handler/admin/user_handler.go @@ -276,6 +276,7 @@ func (h *UserHandler) Create(c *gin.Context) { Concurrency: req.Concurrency, RPMLimit: req.RPMLimit, AllowedGroups: req.AllowedGroups, + ActorAdminID: getAdminIDFromContext(c), }) if err != nil { response.ErrorFrom(c, err) @@ -320,6 +321,7 @@ func (h *UserHandler) Update(c *gin.Context) { Status: req.Status, AllowedGroups: req.AllowedGroups, GroupRates: req.GroupRates, + ActorAdminID: getAdminIDFromContext(c), }) if err != nil { response.ErrorFrom(c, err) diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index 1152b14132..07b85ab827 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -130,6 +130,8 @@ type CreateUserInput struct { Concurrency int RPMLimit int AllowedGroups []int64 + // ActorAdminID 执行本次操作的管理员ID(来自JWT)，仅用于权限敏感操作的审计日志。 + ActorAdminID int64 } type UpdateUserInput struct { @@ -146,6 +148,8 @@ type UpdateUserInput struct { // GroupRates 用户专属分组倍率配置 // map[groupID]*rate，nil 表示删除该分组的专属倍率 GroupRates map[int64]*float64 + // ActorAdminID 执行本次操作的管理员ID(来自JWT)，仅用于权限敏感操作的审计日志。 + ActorAdminID int64 } type AdminBindAuthIdentityInput struct { diff --git a/backend/internal/service/admin_service_role_test.go b/backend/internal/service/admin_service_role_test.go index 26cf9279b2..956b7fae17 100644 --- a/backend/internal/service/admin_service_role_test.go +++ b/backend/internal/service/admin_service_role_test.go @@ -6,6 +6,7 @@ import ( "context" "testing" + "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" "github.com/stretchr/testify/require" ) @@ -83,3 +84,60 @@ func TestAdminService_UpdateUser_InvalidRoleRejected(t *testing.T) { require.Error(t, err) require.Nil(t, repo.lastUpdated, "非法角色不应触发持久化") } + +// roleGuardUserRepoStub 在 rpmUserRepoStub 之上提供可控的管理员计数， +// 用于测试"最后一个管理员不可降级"守卫。 +type roleGuardUserRepoStub struct { + *rpmUserRepoStub + adminTotal int64 + listCalls int +} + +func (s *roleGuardUserRepoStub) ListWithFilters(_ context.Context, _ pagination.PaginationParams, _ UserListFilters) ([]User, *pagination.PaginationResult, error) { + s.listCalls++ + return nil, &pagination.PaginationResult{Total: s.adminTotal}, nil +} + +func TestAdminService_UpdateUser_DemoteLastAdminRejected(t *testing.T) { + base := &userRepoStub{user: &User{ID: 42, Email: "a@example.com", Role: RoleAdmin}} + repo := &roleGuardUserRepoStub{rpmUserRepoStub: &rpmUserRepoStub{userRepoStub: base}, adminTotal: 1} + svc := &adminServiceImpl{userRepo: repo, redeemCodeRepo: &redeemRepoStub{}} + + _, err := svc.UpdateUser(context.Background(), 42, &UpdateUserInput{Role: RoleUser}) + require.Error(t, err) + require.Contains(t, err.Error(), "last admin") + require.Nil(t, repo.lastUpdated, "最后一个管理员不应被降级持久化") + require.Equal(t, 1, repo.listCalls, "降级路径应触发管理员计数") +} + +func TestAdminService_UpdateUser_DemoteAdminAllowedWhenOthersExist(t *testing.T) { + base := &userRepoStub{user: &User{ID: 42, Email: "a@example.com", Role: RoleAdmin}} + repo := &roleGuardUserRepoStub{rpmUserRepoStub: &rpmUserRepoStub{userRepoStub: base}, adminTotal: 2} + invalidator := &authCacheInvalidatorStub{} + svc := &adminServiceImpl{ + userRepo: repo, + redeemCodeRepo: &redeemRepoStub{}, + authCacheInvalidator: invalidator, + } + + updated, err := svc.UpdateUser(context.Background(), 42, &UpdateUserInput{Role: RoleUser}) + require.NoError(t, err) + require.Equal(t, RoleUser, updated.Role) + require.NotNil(t, repo.lastUpdated) + require.Equal(t, RoleUser, repo.lastUpdated.Role, "存在其他管理员时允许降级") +} + +func TestAdminService_UpdateUser_PromoteDoesNotCountAdmins(t *testing.T) { + base := &userRepoStub{user: &User{ID: 42, Email: "u@example.com", Role: RoleUser}} + repo := &roleGuardUserRepoStub{rpmUserRepoStub: &rpmUserRepoStub{userRepoStub: base}, adminTotal: 1} + svc := &adminServiceImpl{ + userRepo: repo, + redeemCodeRepo: &redeemRepoStub{}, + authCacheInvalidator: &authCacheInvalidatorStub{}, + } + + updated, err := svc.UpdateUser(context.Background(), 42, &UpdateUserInput{Role: RoleAdmin}) + require.NoError(t, err) + require.Equal(t, RoleAdmin, updated.Role) + require.Equal(t, 0, repo.listCalls, "升级路径不应触发管理员计数") +} diff --git a/backend/internal/service/admin_user.go b/backend/internal/service/admin_user.go index 9acd2deba7..862a5ab5c7 100644 --- a/backend/internal/service/admin_user.go +++ b/backend/internal/service/admin_user.go @@ -148,10 +148,33 @@ func (s *adminServiceImpl) CreateUser(ctx context.Context, input *CreateUserInpu if err := s.userRepo.Create(ctx, user); err != nil { return nil, err } + // 创建管理员属权限敏感操作，落审计日志（含操作者），便于事后追溯。 + if user.Role == RoleAdmin { + logger.LegacyPrintf("service.admin", "audit: admin user created actor_admin_id=%d target_user_id=%d", + input.ActorAdminID, user.ID) + } s.assignDefaultSubscriptions(ctx, user.ID) return user, nil } +// ensureNotLastAdmin 降级管理员前确认系统中仍存在其他管理员，防止零 admin 锁死。 +// 注：读取与写入之间存在竞态窗口，极端并发下仍可能双双降级；作为后台低频操作 +// 的兜底保护足够，彻底防护需依赖数据库层约束。 +func (s *adminServiceImpl) ensureNotLastAdmin(ctx context.Context) error { + noSubs := false + _, result, err := s.userRepo.ListWithFilters(ctx, + pagination.PaginationParams{Page: 1, PageSize: 1}, + UserListFilters{Role: RoleAdmin, IncludeSubscriptions: &noSubs}, + ) + if err != nil { + return fmt.Errorf("count admin users: %w", err) + } + if result == nil || result.Total <= 1 { + return errors.New("cannot demote the last admin user") + } + return nil +} + func (s *adminServiceImpl) assignDefaultSubscriptions(ctx context.Context, userID int64) { if s.settingService == nil || s.defaultSubAssigner == nil || userID <= 0 { return @@ -221,6 +244,13 @@ func (s *adminServiceImpl) UpdateUser(ctx context.Context, id int64, input *Upda if err != nil { return nil, err } + // 防锁死保护：不允许降级系统中最后一个管理员（自我降级已在 handler 层拦截， + // 此处兜底覆盖跨管理员互降导致零 admin 的场景）。 + if user.Role == RoleAdmin && role == RoleUser { + if err := s.ensureNotLastAdmin(ctx); err != nil { + return nil, err + } + } user.Role = role } @@ -240,6 +270,12 @@ func (s *adminServiceImpl) UpdateUser(ctx context.Context, id int64, input *Upda return nil, err } + // 角色变更属权限敏感操作，落审计日志（含操作者），便于事后追溯。 + if user.Role != oldRole { + logger.LegacyPrintf("service.admin", "audit: user role changed actor_admin_id=%d target_user_id=%d old_role=%s new_role=%s", + input.ActorAdminID, user.ID, oldRole, user.Role) + } + // 同步用户专属分组倍率 if input.GroupRates != nil && s.userGroupRateRepo != nil { if err := s.userGroupRateRepo.SyncUserGroupRates(ctx, user.ID, input.GroupRates); err != nil { diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 297b7428a8..15c4e47635 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -425,14 +425,11 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) - // response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码，传 0。 - if status, errType, errMsg, matched := applyErrorPassthroughRule( - c, account.Platform, 0, payload, - http.StatusBadGateway, "upstream_error", message, + // response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码；统一走语义 + // 状态推断 + body 归一化（与 /v1/responses 路径一致），使按错误码配置的规则可命中。 + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule( + c, account.Platform, payload, message, ); matched { - if status == 0 { - status = http.StatusBadGateway - } if errMsg == "" { errMsg = message } @@ -597,13 +594,11 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) defaultStatus, defaultErrType, defaultMsg := http.StatusBadGateway, "upstream_error", message - if status, errType, errMsg, matched := applyErrorPassthroughRule( - c, account.Platform, 0, payloadBytes, - defaultStatus, defaultErrType, defaultMsg, + // 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致）， + // 使按错误码配置的透传规则可命中。 + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule( + c, account.Platform, payloadBytes, message, ); matched { - if status == 0 { - status = defaultStatus - } if errMsg == "" { errMsg = defaultMsg } diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 18442939d0..b621a96aee 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -465,13 +465,11 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) - if status, errType, errMsg, matched := applyErrorPassthroughRule( - c, account.Platform, 0, payload, - http.StatusBadGateway, "api_error", message, + // 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致）， + // 使按错误码配置的透传规则可命中。 + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule( + c, account.Platform, payload, message, ); matched { - if status == 0 { - status = http.StatusBadGateway - } if errMsg == "" { errMsg = message } @@ -819,13 +817,11 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) errStatus, errType, errMsg := http.StatusBadGateway, "api_error", message - if status, et, em, matched := applyErrorPassthroughRule( - c, account.Platform, 0, payloadBytes, - errStatus, errType, errMsg, + // 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致）， + // 使按错误码配置的透传规则可命中。 + if status, et, em, matched := applyOpenAIStreamFailedErrorPassthroughRule( + c, account.Platform, payloadBytes, message, ); matched { - if status == 0 { - status = errStatus - } if em == "" { em = errMsg } diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 93588bdd69..635b2ac950 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -694,8 +694,12 @@ func openAIStreamFailedEventPassthroughBody(payload []byte, failedMessage string return body } +// applyOpenAIStreamFailedErrorPassthroughRule 对 response.failed 事件应用错误透传规则： +// 归一化 body 供关键词匹配/消息提取，并推断语义状态码使按错误码配置的规则可以命中。 +// platform 必须传 account.Platform——本服务同时承载 openai 与 grok 平台账号，规则按平台匹配。 func applyOpenAIStreamFailedErrorPassthroughRule( c *gin.Context, + platform string, payload []byte, failedMessage string, ) (status int, errType string, errMsg string, matched bool) { @@ -703,7 +707,7 @@ func applyOpenAIStreamFailedErrorPassthroughRule( upstreamStatus := openAIStreamFailedEventSemanticStatus(payload, failedMessage) return applyErrorPassthroughRule( c, - PlatformOpenAI, + platform, upstreamStatus, ruleBody, http.StatusBadGateway, @@ -922,7 +926,10 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( }) } if !openAIStreamClientOutputStarted(c, clientOutputStarted) { - if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, dataBytes, failedMessage); matched { + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, account.Platform, dataBytes, failedMessage); matched { + // 命中透传规则也要记录 ops 上游错误事件（对齐 CC/Messages 与 + // antigravity 先例），否则透传命中的 failed 在监控中不可见。 + s.recordOpenAIStreamUpstreamError(c, account, true, upstreamRequestID, "http_error", dataBytes, failedMessage) MarkResponseCommitted(c) c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8") c.JSON(status, gin.H{ diff --git a/backend/internal/service/openai_gateway_response_failed_passthrough_test.go b/backend/internal/service/openai_gateway_response_failed_passthrough_test.go index 56fd66c94c..f9439c0551 100644 --- a/backend/internal/service/openai_gateway_response_failed_passthrough_test.go +++ b/backend/internal/service/openai_gateway_response_failed_passthrough_test.go @@ -141,3 +141,84 @@ func TestForwardAsChatCompletions_ResponseFailed_NoRule_Still502(t *testing.T) { require.Error(t, err) require.Equal(t, http.StatusBadGateway, rec.Code, "without passthrough rule should still be 502") } + +// bindStatusCodePassthroughRule 绑定一条按错误码+关键词双条件(MatchModeAll)匹配的规则。 +// 此类规则依赖语义状态码推断才能在协议转换路径命中（response.failed 无真实 HTTP 状态码）。 +func bindStatusCodePassthroughRule(c *gin.Context, platform string, statusCode int, keyword string, responseCode int) { + rule := &model.ErrorPassthroughRule{ + ID: 1, + Name: "status-code-rule", + Enabled: true, + Priority: 1, + Platforms: []string{platform}, + ErrorCodes: []int{statusCode}, + Keywords: []string{keyword}, + MatchMode: model.MatchModeAll, + ResponseCode: &responseCode, + PassthroughBody: true, + } + svc := &ErrorPassthroughService{} + svc.setLocalCache([]*model.ErrorPassthroughRule{rule}) + BindErrorPassthroughService(c, svc) +} + +func TestForwardAsChatCompletions_ResponseFailed_ErrorCodeRuleMatchesViaSemanticStatus(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + bindStatusCodePassthroughRule(c, "openai", http.StatusBadRequest, "context_length_exceeded", http.StatusBadRequest) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(buildContextLengthFailedSSE())), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + + require.Error(t, err) + require.Equal(t, http.StatusBadRequest, rec.Code, "error-code-conditioned rule should match via semantic status inference") + respBody := rec.Body.String() + require.Equal(t, "upstream_error", gjson.Get(respBody, "error.type").String()) + require.Contains(t, gjson.Get(respBody, "error.message").String(), "context window") +} + +func TestForwardAsAnthropic_ResponseFailed_ErrorCodeRuleMatchesViaSemanticStatus(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + bindStatusCodePassthroughRule(c, "openai", http.StatusBadRequest, "context_length_exceeded", http.StatusBadRequest) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(buildContextLengthFailedSSE())), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + + require.Error(t, err) + require.Equal(t, http.StatusBadRequest, rec.Code, "error-code-conditioned rule should match via semantic status inference") + respBody := rec.Body.String() + require.NotEmpty(t, gjson.Get(respBody, "error.message").String()) +} diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 51fc3a621b..b814ff96f3 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -256,8 +256,11 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp }) } if !openAIStreamClientOutputStarted(c, clientOutputStarted) { - if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, dataBytes, failedMessage); matched { + if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, account.Platform, dataBytes, failedMessage); matched { sawFailedEvent = true + // 命中透传规则也要记录 ops 上游错误事件（对齐 CC/Messages 与 + // antigravity 先例），否则透传命中的 failed 在监控中不可见。 + s.recordOpenAIStreamUpstreamError(c, account, false, upstreamRequestID, "http_error", dataBytes, failedMessage) MarkResponseCommitted(c) c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8") c.JSON(status, gin.H{ diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 11d45d9dc5..9f42a82312 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -1504,6 +1504,11 @@ func TestOpenAIStreamingContextWindowResponseFailedBeforeOutputAppliesPassthroug require.Equal(t, upstreamMessage, gjson.Get(body, "error.message").String()) require.NotContains(t, body, "response.failed") require.NotContains(t, body, "Upstream request failed") + // 命中透传规则也应记录 ops 上游错误事件（对齐 CC/Messages 与 antigravity 先例）。 + opsVal, opsRecorded := c.Get(OpsUpstreamErrorsKey) + require.True(t, opsRecorded, "passthrough hit should record an ops upstream error event") + opsEvents, _ := opsVal.([]*OpsUpstreamErrorEvent) + require.NotEmpty(t, opsEvents) } func TestOpenAIStreamingPreambleOnlyMissingTerminalReturnsFailover(t *testing.T) { @@ -1890,6 +1895,11 @@ func TestOpenAIStreamingPassthroughContextWindowResponseFailedBeforeOutputApplie require.Equal(t, upstreamMessage, gjson.Get(body, "error.message").String()) require.NotContains(t, body, "response.failed") require.NotContains(t, body, "Upstream request failed") + // 命中透传规则也应记录 ops 上游错误事件（对齐 CC/Messages 与 antigravity 先例）。 + opsVal, opsRecorded := c.Get(OpsUpstreamErrorsKey) + require.True(t, opsRecorded, "passthrough hit should record an ops upstream error event") + opsEvents, _ := opsVal.([]*OpsUpstreamErrorEvent) + require.NotEmpty(t, opsEvents) } func TestOpenAIStreamingPassthroughContextWindowResponseFailedBeforeOutputWithoutRulePassesThrough(t *testing.T) { diff --git a/frontend/src/i18n/locales/en/admin/overview.ts b/frontend/src/i18n/locales/en/admin/overview.ts index 9ab9eeac9c..0d4a0d0455 100644 --- a/frontend/src/i18n/locales/en/admin/overview.ts +++ b/frontend/src/i18n/locales/en/admin/overview.ts @@ -433,6 +433,7 @@ export default { creating: 'Creating...', updating: 'Updating...', form: { + roleLabel: 'Role', rpmLimit: 'Requests Per Minute (RPM)', rpmLimitPlaceholder: '0 = unlimited', rpmLimitHint: 'Max requests per minute for this user; 0 = unlimited. Acts as a fallback only when the group has no rpm_limit set.'