From 6a69dd051a66a37ab2f0a9427f4b2bffcb878012 Mon Sep 17 00:00:00 2001 From: PMExtra Date: Thu, 24 Sep 2026 21:27:57 +0800 Subject: [PATCH] feat: add risk control user allowlist --- .../internal/handler/admin/setting_handler.go | 1 + .../handler/admin/setting_handler_audit.go | 3 + .../handler/admin/setting_handler_update.go | 19 ++- backend/internal/handler/dto/settings.go | 5 +- .../handler/openai_cyber_allowlist.go | 18 +++ .../handler/openai_cyber_allowlist_test.go | 79 +++++++++ .../handler/openai_gateway_handler.go | 14 +- .../openai_ws_v2_passthrough_cyber_test.go | 10 +- .../repository/content_moderation_repo.go | 4 + .../content_moderation_repo_test.go | 15 ++ backend/internal/server/api_contract_test.go | 2 + .../internal/service/content_moderation.go | 77 ++++++--- .../service/cyber_policy_allowlist.go | 50 ++++++ .../service/cyber_policy_allowlist_test.go | 62 +++++++ backend/internal/service/domain_constants.go | 1 + .../service/risk_control_allowlist_test.go | 153 ++++++++++++++++++ .../service/setting_cyber_allowlist_test.go | 42 +++++ .../service/setting_gateway_runtime.go | 42 +++-- backend/internal/service/setting_parse.go | 2 + backend/internal/service/setting_service.go | 1 + backend/internal/service/setting_update.go | 9 ++ backend/internal/service/settings_view.go | 1 + frontend/src/api/admin/settings.ts | 2 + .../src/i18n/locales/en/admin/settings.ts | 2 + .../src/i18n/locales/zh/admin/settings.ts | 2 + frontend/src/views/admin/SettingsView.vue | 27 ++++ 26 files changed, 598 insertions(+), 45 deletions(-) create mode 100644 backend/internal/handler/openai_cyber_allowlist.go create mode 100644 backend/internal/handler/openai_cyber_allowlist_test.go create mode 100644 backend/internal/service/cyber_policy_allowlist.go create mode 100644 backend/internal/service/cyber_policy_allowlist_test.go create mode 100644 backend/internal/service/risk_control_allowlist_test.go create mode 100644 backend/internal/service/setting_cyber_allowlist_test.go diff --git a/backend/internal/handler/admin/setting_handler.go b/backend/internal/handler/admin/setting_handler.go index 036fbbfca2..7bff15c48a 100644 --- a/backend/internal/handler/admin/setting_handler.go +++ b/backend/internal/handler/admin/setting_handler.go @@ -265,6 +265,7 @@ func (h *SettingHandler) GetSettings(c *gin.Context) { DefaultBalance: settings.DefaultBalance, RiskControlEnabled: settings.RiskControlEnabled, CyberSessionBlockEnabled: settings.CyberSessionBlockEnabled, + CyberPolicyUserAllowlist: settings.CyberPolicyUserAllowlist, CyberSessionBlockTTLSeconds: settings.CyberSessionBlockTTLSeconds, AffiliateRebateRate: settings.AffiliateRebateRate, AffiliateRebateFreezeHours: settings.AffiliateRebateFreezeHours, diff --git a/backend/internal/handler/admin/setting_handler_audit.go b/backend/internal/handler/admin/setting_handler_audit.go index b67eb88c49..81e48f6772 100644 --- a/backend/internal/handler/admin/setting_handler_audit.go +++ b/backend/internal/handler/admin/setting_handler_audit.go @@ -603,6 +603,9 @@ func diffSettings(before *service.SystemSettings, after *service.SystemSettings, if before.RiskControlEnabled != after.RiskControlEnabled { changed = append(changed, "risk_control_enabled") } + if before.CyberPolicyUserAllowlist != after.CyberPolicyUserAllowlist { + changed = append(changed, "cyber_policy_user_allowlist") + } if before.CyberSessionBlockEnabled != after.CyberSessionBlockEnabled { changed = append(changed, "cyber_session_block_enabled") } diff --git a/backend/internal/handler/admin/setting_handler_update.go b/backend/internal/handler/admin/setting_handler_update.go index 8ca112e7f7..618f0bde5c 100644 --- a/backend/internal/handler/admin/setting_handler_update.go +++ b/backend/internal/handler/admin/setting_handler_update.go @@ -364,8 +364,9 @@ type UpdateSettingsRequest struct { RiskControlEnabled *bool `json:"risk_control_enabled"` // cyber 会话屏蔽开关 + TTL - CyberSessionBlockEnabled *bool `json:"cyber_session_block_enabled"` - CyberSessionBlockTTLSeconds *int `json:"cyber_session_block_ttl_seconds"` + CyberSessionBlockEnabled *bool `json:"cyber_session_block_enabled"` + CyberPolicyUserAllowlist *string `json:"cyber_policy_user_allowlist"` + CyberSessionBlockTTLSeconds *int `json:"cyber_session_block_ttl_seconds"` // OpenAI fast/flex policy (optional, only updated when provided) OpenAIFastPolicySettings *dto.OpenAIFastPolicySettings `json:"openai_fast_policy_settings,omitempty"` @@ -1503,6 +1504,13 @@ func (h *SettingHandler) UpdateSettings(c *gin.Context) { } } + if req.CyberPolicyUserAllowlist != nil { + if _, err := service.ParseCyberPolicyUserAllowlist(*req.CyberPolicyUserAllowlist); err != nil { + response.BadRequest(c, err.Error()) + return + } + } + // cyber 会话屏蔽 TTL 校验：提供时必须 > 0 if req.CyberSessionBlockTTLSeconds != nil && *req.CyberSessionBlockTTLSeconds <= 0 { response.BadRequest(c, "cyber_session_block_ttl_seconds must be > 0") @@ -2008,6 +2016,12 @@ func (h *SettingHandler) UpdateSettings(c *gin.Context) { } return previousSettings.RiskControlEnabled }(), + CyberPolicyUserAllowlist: func() string { + if req.CyberPolicyUserAllowlist != nil { + return *req.CyberPolicyUserAllowlist + } + return previousSettings.CyberPolicyUserAllowlist + }(), CyberSessionBlockEnabled: func() bool { if req.CyberSessionBlockEnabled != nil { return *req.CyberSessionBlockEnabled @@ -2429,6 +2443,7 @@ func (h *SettingHandler) UpdateSettings(c *gin.Context) { RiskControlEnabled: updatedSettings.RiskControlEnabled, CyberSessionBlockEnabled: updatedSettings.CyberSessionBlockEnabled, + CyberPolicyUserAllowlist: updatedSettings.CyberPolicyUserAllowlist, CyberSessionBlockTTLSeconds: updatedSettings.CyberSessionBlockTTLSeconds, AccountSchedulingThresholds: updatedSettings.AccountSchedulingThresholds, AllowUserViewErrorRequests: updatedSettings.AllowUserViewErrorRequests, diff --git a/backend/internal/handler/dto/settings.go b/backend/internal/handler/dto/settings.go index 72f2a56cbc..4f16f11d45 100644 --- a/backend/internal/handler/dto/settings.go +++ b/backend/internal/handler/dto/settings.go @@ -335,8 +335,9 @@ type SystemSettings struct { RiskControlEnabled bool `json:"risk_control_enabled"` // cyber 会话屏蔽开关 + TTL - CyberSessionBlockEnabled bool `json:"cyber_session_block_enabled"` - CyberSessionBlockTTLSeconds int `json:"cyber_session_block_ttl_seconds"` + CyberSessionBlockEnabled bool `json:"cyber_session_block_enabled"` + CyberPolicyUserAllowlist string `json:"cyber_policy_user_allowlist"` + CyberSessionBlockTTLSeconds int `json:"cyber_session_block_ttl_seconds"` // Affiliate (邀请返利) feature switch AffiliateEnabled bool `json:"affiliate_enabled"` diff --git a/backend/internal/handler/openai_cyber_allowlist.go b/backend/internal/handler/openai_cyber_allowlist.go new file mode 100644 index 0000000000..7112bff0ad --- /dev/null +++ b/backend/internal/handler/openai_cyber_allowlist.go @@ -0,0 +1,18 @@ +package handler + +import ( + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" +) + +func (h *OpenAIGatewayHandler) cyberPolicyLogOnly(c *gin.Context, apiKey *service.APIKey) bool { + return h != nil && c != nil && c.Request != nil && h.gatewayService.CyberPolicyLogOnly(c.Request.Context(), apiKey) +} + +// Both HTTP and WebSocket admission skip existing blocks for trusted users. +func (h *OpenAIGatewayHandler) findBlockedCyberSessionForAPIKey(c *gin.Context, apiKey *service.APIKey, body []byte) string { + if apiKey == nil || h.cyberPolicyLogOnly(c, apiKey) { + return "" + } + return findBlockedCyberSessionKey(c.Request.Context(), h.gatewayService, apiKey.ID, c, body) +} diff --git a/backend/internal/handler/openai_cyber_allowlist_test.go b/backend/internal/handler/openai_cyber_allowlist_test.go new file mode 100644 index 0000000000..22bb674586 --- /dev/null +++ b/backend/internal/handler/openai_cyber_allowlist_test.go @@ -0,0 +1,79 @@ +package handler + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + coderws "github.com/coder/websocket" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestCyberAllowlistedUserBypassesExistingBlocksAndContinuesWebSocket(t *testing.T) { + gin.SetMode(gin.TestMode) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := coderws.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.CloseNow() }() + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + for _, response := range []string{ + `{"type":"response.failed","response":{"id":"resp_cyber","model":"gpt-5.1","error":{"code":"cyber_policy","message":"blocked by upstream"}}}`, + `{"type":"response.completed","response":{"id":"resp_ok","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`, + } { + if _, _, err := conn.Read(ctx); err != nil { + return + } + if err := conn.Write(ctx, coderws.MessageText, []byte(response)); err != nil { + return + } + } + // Let the downstream close so the completed frame can drain normally. + _, _, _ = conn.Read(ctx) + })) + defer upstream.Close() + harness := newOpenAIWSPassthroughHandlerHarness(t, upstream.URL, map[string]string{ + service.SettingKeyCyberPolicyUserAllowlist: "1751", + }) + payload := []byte(`{"type":"response.create","model":"gpt-5.1","prompt_cache_key":"trusted-session","input":"test"}`) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", strings.NewReader(string(payload))) + explicitKey := service.CyberSessionExplicitBlockKey(harness.apiKey.ID, c, payload) + store, ok := harness.gatewayCache.(service.CyberSessionBlockStore) + require.True(t, ok) + require.NoError(t, store.SetCyberSessionBlocked(context.Background(), "", []string{explicitKey}, time.Minute)) + for _, format := range []cyberSessionBlockFormat{cyberBlockFormatResponses, cyberBlockFormatChat, cyberBlockFormatAnthropic} { + require.False(t, harness.handler.rejectIfCyberSessionBlocked(c, harness.apiKey, payload, "gpt-5.1", format)) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, harness.clientConn.Write(ctx, coderws.MessageText, payload)) + _, event, err := harness.clientConn.Read(ctx) + require.NoError(t, err) + require.Equal(t, "cyber_policy", gjson.GetBytes(event, "response.error.code").String()) + require.Eventually(t, func() bool { + logs := harness.moderationRepo.logSnapshot() + return len(logs) == 1 && logs[0].Mode == service.ContentModerationModeCyberLogOnly + }, 3*time.Second, 10*time.Millisecond) + matched, err := store.FindCyberSessionBlocked(ctx, service.CyberSessionTranscriptBlockKeys(harness.apiKey.ID, payload)) + require.NoError(t, err) + require.Empty(t, matched, "the cyber response must not write new transcript blocks") + require.NoError(t, harness.clientConn.Write(ctx, coderws.MessageText, payload)) + _, event, err = harness.clientConn.Read(ctx) + require.NoError(t, err) + require.Equal(t, "response.completed", gjson.GetBytes(event, "type").String()) + _ = harness.clientConn.CloseNow() + select { + case <-harness.handlerDone: + case <-ctx.Done(): + t.Fatal("handler did not exit") + } +} diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index bf4563f2f2..305f369a53 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -2440,7 +2440,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { // The first response.create frame is available here, so explicit IDs are // checked directly and body-derived sessions use the coarse scope gate. - if cyberBlockKey := findBlockedCyberSessionKey(c.Request.Context(), h.gatewayService, apiKey.ID, c, firstMessage); cyberBlockKey != "" { + if cyberBlockKey := h.findBlockedCyberSessionForAPIKey(c, apiKey, firstMessage); cyberBlockKey != "" { writeCyberSessionBlockedWSError(c.Request.Context(), wsConn) closeOpenAIClientWS(wsConn, coderws.StatusPolicyViolation, "session blocked by cyber-security policy") h.enqueueCyberSessionBlockedOpsEntry(c, apiKey, reqModel, cyberBlockKey) @@ -2817,7 +2817,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { // 连接级 cyber session gate 也在 BeforeRequest 先执行，使 native 与 // passthrough ingress 都能在 BeforeTurn 及上游写入前无副作用地拒绝。 // BeforeTurn 中保留同一检查作为防御式兜底。 - if cyberBlockedThisConn { + if cyberBlockedThisConn && !h.cyberPolicyLogOnly(c, apiKey) { return service.NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, cyberSessionBlockedClientMsg, nil) } if turn == 1 { @@ -2869,7 +2869,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { }, BeforeTurn: func(turn int) error { // turn==1 的会话屏蔽已由握手层检查覆盖；连接内 flag 只拦截后续 turn。 - if cyberBlockedThisConn { + if cyberBlockedThisConn && !h.cyberPolicyLogOnly(c, apiKey) { return service.NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, cyberSessionBlockedClientMsg, nil) } // 长连接跨峰谷/倍率刷新防护：每个 turn 按当前时刻重装门并复核 @@ -2949,7 +2949,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { cyberBlockedThisConn, cyberBlockPendingAfterFailover = advanceOpenAIWSCyberBlockState( cyberBlockedThisConn, cyberBlockPendingAfterFailover, - cyberMarked, + cyberMarked && !h.cyberPolicyLogOnly(c, apiKey), turnErr, ) if turnErr != nil { @@ -4045,7 +4045,7 @@ func (h *OpenAIGatewayHandler) rejectIfCyberSessionBlocked(c *gin.Context, apiKe if enabled, _ := h.gatewayService.CyberSessionBlockRuntime(c.Request.Context()); !enabled { return false } - key := findBlockedCyberSessionKey(c.Request.Context(), h.gatewayService, apiKey.ID, c, body) + key := h.findBlockedCyberSessionForAPIKey(c, apiKey, body) if key == "" { return false } @@ -4242,7 +4242,8 @@ func (h *OpenAIGatewayHandler) recordCyberPolicyIfMarked(c *gin.Context, apiKey ClientIP: clientIPStr, CreatedAt: time.Now(), } - if gwSvc != nil && apiKey != nil { + cyberLogOnly := h.cyberPolicyLogOnly(c, apiKey) + if gwSvc != nil && apiKey != nil && !cyberLogOnly { plan := buildCyberSessionBlockWritePlan(apiKey.ID, c, cyberBlockBody) if len(plan.keys) > 0 { blockCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) @@ -4255,6 +4256,7 @@ func (h *OpenAIGatewayHandler) recordCyberPolicyIfMarked(c *gin.Context, apiKey defer cancel() if cmSvc != nil { cmSvc.RecordCyberPolicyEvent(ctx, service.CyberPolicyRecordInput{ + LogOnly: cyberLogOnly, RequestID: requestID, UserID: userID, UserEmail: userEmail, diff --git a/backend/internal/handler/openai_ws_v2_passthrough_cyber_test.go b/backend/internal/handler/openai_ws_v2_passthrough_cyber_test.go index 49e17c3e75..61ba54488a 100644 --- a/backend/internal/handler/openai_ws_v2_passthrough_cyber_test.go +++ b/backend/internal/handler/openai_ws_v2_passthrough_cyber_test.go @@ -19,6 +19,7 @@ import ( ) type openAIWSPassthroughHandlerHarness struct { + handler *OpenAIGatewayHandler clientConn *coderws.Conn handlerDone <-chan struct{} moderationRepo *contentModerationHandlerTestRepo @@ -26,7 +27,7 @@ type openAIWSPassthroughHandlerHarness struct { apiKey *service.APIKey } -func newOpenAIWSPassthroughHandlerHarness(t *testing.T, upstreamURL string) *openAIWSPassthroughHandlerHarness { +func newOpenAIWSPassthroughHandlerHarness(t *testing.T, upstreamURL string, settings ...map[string]string) *openAIWSPassthroughHandlerHarness { t.Helper() gatewayCache := testutil.NewRedisGatewayCache(t) @@ -35,6 +36,11 @@ func newOpenAIWSPassthroughHandlerHarness(t *testing.T, upstreamURL string) *ope service.SettingKeyCyberSessionBlockEnabled: "true", service.SettingKeyCyberSessionBlockTTLSeconds: "60", }} + for _, overrides := range settings { + for key, value := range overrides { + settingRepo.values[key] = value + } + } moderationRepo := &contentModerationHandlerTestRepo{} moderationSvc := service.NewContentModerationService(settingRepo, moderationRepo, nil, nil, nil, nil, nil, nil) settingSvc := service.NewSettingService(settingRepo, nil) @@ -90,6 +96,7 @@ func newOpenAIWSPassthroughHandlerHarness(t *testing.T, upstreamURL string) *ope apiKey := &service.APIKey{ ID: 1851, + UserID: 1751, Name: "ws-cyber-key", Key: "sk-handler-cyber-test", GroupID: &groupID, @@ -116,6 +123,7 @@ func newOpenAIWSPassthroughHandlerHarness(t *testing.T, upstreamURL string) *ope t.Cleanup(func() { _ = clientConn.CloseNow() }) return &openAIWSPassthroughHandlerHarness{ + handler: h, clientConn: clientConn, handlerDone: handlerDone, moderationRepo: moderationRepo, diff --git a/backend/internal/repository/content_moderation_repo.go b/backend/internal/repository/content_moderation_repo.go index 32b8e9150f..20142bfd5e 100644 --- a/backend/internal/repository/content_moderation_repo.go +++ b/backend/internal/repository/content_moderation_repo.go @@ -197,6 +197,8 @@ func (r *contentModerationRepository) CountFlaggedByUserSince(ctx context.Contex return 0, nil } // SQL 中的 'cyber_policy' 字面量须与 service.ContentModerationActionCyberPolicy 保持一致。 + // 'cyber_log_only' matches service.ContentModerationModeCyberLogOnly; these + // events remain evidence but never become penalties after allowlist removal. var count int err := r.db.QueryRowContext(ctx, ` WITH last_auto_ban AS ( @@ -209,6 +211,8 @@ FROM content_moderation_logs WHERE user_id = $1 AND flagged = TRUE AND action <> 'hash_block' + AND mode <> 'cyber_log_only' + AND mode <> 'risk_control_log_only' AND ($3::bool IS FALSE OR action <> 'cyber_policy') AND created_at >= $2 AND created_at > COALESCE((SELECT at FROM last_auto_ban), '-infinity'::timestamptz) diff --git a/backend/internal/repository/content_moderation_repo_test.go b/backend/internal/repository/content_moderation_repo_test.go index 2b32d09f97..7be148c149 100644 --- a/backend/internal/repository/content_moderation_repo_test.go +++ b/backend/internal/repository/content_moderation_repo_test.go @@ -102,3 +102,18 @@ func TestContentModerationRepositoryCountFlaggedByUserSince_ExcludesCyberPolicyW require.Equal(t, 3, count) require.NoError(t, mock.ExpectationsWereMet()) } + +func TestContentModerationRepositoryCountFlaggedByUserSince_ExcludesCyberLogOnly(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer func() { _ = db.Close() }() + repo := NewContentModerationRepository(db) + since := time.Now().Add(-time.Hour) + mock.ExpectQuery(regexp.QuoteMeta("AND mode <> 'cyber_log_only'\n AND mode <> 'risk_control_log_only'")). + WithArgs(int64(12), since, false). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0)) + count, err := repo.CountFlaggedByUserSince(context.Background(), 12, since, false) + require.NoError(t, err) + require.Zero(t, count) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/backend/internal/server/api_contract_test.go b/backend/internal/server/api_contract_test.go index 219c83b149..f4d4c4a019 100644 --- a/backend/internal/server/api_contract_test.go +++ b/backend/internal/server/api_contract_test.go @@ -1004,6 +1004,7 @@ func TestAPIContracts(t *testing.T) { "model_plaza_description": "", "plugin_management_enabled": false, "risk_control_enabled": false, + "cyber_policy_user_allowlist": "", "cyber_session_block_enabled": false, "cyber_session_block_ttl_seconds": 3600, "affiliate_enabled": false, @@ -1322,6 +1323,7 @@ func TestAPIContracts(t *testing.T) { "model_plaza_description": "", "plugin_management_enabled": false, "risk_control_enabled": false, + "cyber_policy_user_allowlist": "", "cyber_session_block_enabled": false, "cyber_session_block_ttl_seconds": 3600, "affiliate_enabled": false, diff --git a/backend/internal/service/content_moderation.go b/backend/internal/service/content_moderation.go index cb000e2a0a..3c6df48ed2 100644 --- a/backend/internal/service/content_moderation.go +++ b/backend/internal/service/content_moderation.go @@ -316,18 +316,19 @@ type ContentModerationModelFilter struct { } type ContentModerationCheckInput struct { - RequestID string - UserID int64 - UserEmail string - APIKeyID int64 - APIKeyName string - GroupID *int64 - GroupName string - Endpoint string - Provider string - Model string - Protocol string - Body []byte + riskControlLogOnly bool // Captured at admission and retained by asynchronous tasks. + RequestID string + UserID int64 + UserEmail string + APIKeyID int64 + APIKeyName string + GroupID *int64 + GroupName string + Endpoint string + Provider string + Model string + Protocol string + Body []byte } type ContentModerationInput struct { @@ -545,6 +546,7 @@ type ContentModerationService struct { } type contentModerationRuntimeSnapshot struct { + allowlistedUsers map[int64]struct{} riskControlEnabled bool config *ContentModerationConfig keywordMatcher *contentModerationKeywordMatcher @@ -807,7 +809,7 @@ func (s *ContentModerationService) TestAPIKeys(ctx context.Context, input TestCo return &TestContentModerationAPIKeysResult{Items: items, AuditResult: auditResult, ImageCount: imageCount}, nil } -func (s *ContentModerationService) Check(ctx context.Context, input ContentModerationCheckInput) (*ContentModerationDecision, error) { +func (s *ContentModerationService) Check(ctx context.Context, input ContentModerationCheckInput) (decision *ContentModerationDecision, err error) { allow := &ContentModerationDecision{Allowed: true, Action: ContentModerationActionAllow} if s == nil || s.settingRepo == nil || s.repo == nil { slog.Info("content_moderation.skip_unavailable", @@ -838,6 +840,15 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer "protocol", input.Protocol) return allow, nil } + _, input.riskControlLogOnly = runtimeSnapshot.allowlistedUsers[input.UserID] + // Keep audit evidence while ensuring all local rejection paths allow trusted users. + defer func() { + if input.riskControlLogOnly && decision != nil { + decision.Allowed, decision.Blocked = true, false + decision.Action = ContentModerationActionAllow + decision.Message, decision.StatusCode = "", 0 + } + }() cfg := runtimeSnapshot.config inGroupScope := cfg.includesGroup(input.GroupID) inModelScope := cfg.includesModel(input.Model) @@ -909,7 +920,11 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer if cfg.KeywordBlockingMode != ContentModerationKeywordModeAPIOnly && len(cfg.BlockedKeywords) > 0 { keywordText := extractContentModerationKeywordText(input.Protocol, input.Body) if keyword, hit := runtimeSnapshot.matchBlockedKeyword(keywordText); hit { - s.recordPreBlockSyncMetric(0, ContentModerationActionKeywordBlock) + metricAction := ContentModerationActionKeywordBlock + if input.riskControlLogOnly { + metricAction = ContentModerationActionAllow + } + s.recordPreBlockSyncMetric(0, metricAction) slog.Info("content_moderation.keyword_block", "user_id", input.UserID, "api_key_id", input.APIKeyID, @@ -974,7 +989,11 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer } if matched { if cfg.Mode == ContentModerationModePreBlock { - s.recordPreBlockSyncMetric(0, ContentModerationActionHashBlock) + metricAction := ContentModerationActionHashBlock + if input.riskControlLogOnly { + metricAction = ContentModerationActionAllow + } + s.recordPreBlockSyncMetric(0, metricAction) } slog.Info("content_moderation.hash_block", "user_id", input.UserID, @@ -1081,7 +1100,7 @@ func (s *ContentModerationService) checkSync(ctx context.Context, input ContentM flagged, highestCategory, highestScore := evaluateModerationScores(result.CategoryScores, cfg.Thresholds) action := ContentModerationActionAllow blocked := false - if allowBlock && flagged && cfg.Mode == ContentModerationModePreBlock { + if allowBlock && !input.riskControlLogOnly && flagged && cfg.Mode == ContentModerationModePreBlock { action = ContentModerationActionBlock blocked = true } @@ -1572,14 +1591,20 @@ func (s *ContentModerationService) refreshRuntimeSnapshot(ctx context.Context) ( values, err := s.settingRepo.GetMultiple(ctx, []string{ SettingKeyRiskControlEnabled, SettingKeyContentModerationConfig, + SettingKeyCyberPolicyUserAllowlist, }) if err != nil { return nil, fmt.Errorf("get content moderation runtime settings: %w", err) } + allowlistedUsers, err := ParseCyberPolicyUserAllowlist(values[SettingKeyCyberPolicyUserAllowlist]) + if err != nil { + return nil, err + } rawConfig := values[SettingKeyContentModerationConfig] configDigest := sha256.Sum256([]byte(rawConfig)) if current := s.runtimeSnapshot.Load(); current != nil && current.configDigest == configDigest { snapshot := &contentModerationRuntimeSnapshot{ + allowlistedUsers: allowlistedUsers, riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true", config: current.config, keywordMatcher: current.keywordMatcher, @@ -1596,6 +1621,7 @@ func (s *ContentModerationService) refreshRuntimeSnapshot(ctx context.Context) ( } cfg = cfg.effectiveEngine(cfg.Engine) snapshot := &contentModerationRuntimeSnapshot{ + allowlistedUsers: allowlistedUsers, riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true", config: cfg, keywordMatcher: newContentModerationKeywordMatcher(cfg.BlockedKeywords), @@ -1628,6 +1654,7 @@ func (s *ContentModerationService) replaceRuntimeConfig(cfg *ContentModerationCo return } s.runtimeSnapshot.Store(&contentModerationRuntimeSnapshot{ + allowlistedUsers: current.allowlistedUsers, riskControlEnabled: current.riskControlEnabled, config: config, keywordMatcher: keywordMatcher, @@ -1874,6 +1901,11 @@ func (s *ContentModerationService) buildLog(input ContentModerationCheckInput, c if input.APIKeyID > 0 { apiKeyID = &input.APIKeyID } + mode := cfg.Mode + if input.riskControlLogOnly { + mode = ContentModerationModeRiskControlLogOnly + action = ContentModerationActionAllow + } return &ContentModerationLog{ RequestID: input.RequestID, UserID: userID, @@ -1885,7 +1917,7 @@ func (s *ContentModerationService) buildLog(input ContentModerationCheckInput, c Endpoint: input.Endpoint, Provider: input.Provider, Model: input.Model, - Mode: cfg.Mode, + Mode: mode, Action: action, Flagged: flagged, HighestCategory: highestCategory, @@ -1903,6 +1935,9 @@ func (s *ContentModerationService) persistContentModerationLog(ctx context.Conte if s == nil || log == nil { return } + if log.Mode == ContentModerationModeRiskControlLogOnly { + recordHash, applySideEffects = false, false + } if recordHash && s.hashCache != nil { if err := s.hashCache.RecordFlaggedInputHash(ctx, hashText); err != nil { slog.Warn("content_moderation.record_hash_failed", "user_id", contentModerationEmailUserID(log), "endpoint", log.Endpoint, "error", err) @@ -2990,6 +3025,7 @@ func maskSecretTail(secret string) string { // CyberPolicyRecordInput 是一次 cyber_policy 硬阻断的风控记录入参。 type CyberPolicyRecordInput struct { + LogOnly bool // Trusted platform user: retain evidence without penalties or notifications. RequestID string UserID int64 UserEmail string @@ -3064,7 +3100,10 @@ func (s *ContentModerationService) RecordCyberPolicyEvent(ctx context.Context, i // 开关开时 cyber_policy 不参与封号计数：当次不判定（此处跳过）， // 历史行由 CountFlaggedByUserSince 的 excludeCyberPolicy 排除。 autoBanned := false - if !cfg.CyberPolicyExcludeFromBanCount { + if in.LogOnly { + log.Mode = ContentModerationModeCyberLogOnly + } + if !in.LogOnly && !cfg.CyberPolicyExcludeFromBanCount { autoBanned = s.applyFlaggedAccountSideEffects(ctx, cfg, log) } log.EmailSent = false @@ -3074,7 +3113,7 @@ func (s *ContentModerationService) RecordCyberPolicyEvent(ctx context.Context, i slog.Warn("content_moderation.cyber_create_log_failed", "user_id", in.UserID, "error", err) } emailSent := false - if s.emailService != nil && strings.TrimSpace(log.UserEmail) != "" { + if !in.LogOnly && s.emailService != nil && strings.TrimSpace(log.UserEmail) != "" { if err := s.sendCyberPolicyEmail(ctx, log); err != nil { slog.Warn("content_moderation.cyber_email_failed", "user_id", in.UserID, "error", err) } else { diff --git a/backend/internal/service/cyber_policy_allowlist.go b/backend/internal/service/cyber_policy_allowlist.go new file mode 100644 index 0000000000..ed3fb59ef0 --- /dev/null +++ b/backend/internal/service/cyber_policy_allowlist.go @@ -0,0 +1,50 @@ +package service + +import ( + "context" + "fmt" + "strconv" + "strings" + "unicode" +) + +// ContentModerationModeCyberLogOnly preserves upstream evidence while excluding +// these events from subsequent automatic-ban counts, even after whitelist removal. +const ContentModerationModeCyberLogOnly = "cyber_log_only" +const ContentModerationModeRiskControlLogOnly = "risk_control_log_only" + +// The historical cyber_policy_user_allowlist storage/API key also governs ordinary moderation. + +// ParseCyberPolicyUserAllowlist accepts positive platform user IDs separated by +// commas or whitespace. Reject the whole list on invalid input; never accept a +// partially parsed privilege configuration. +func ParseCyberPolicyUserAllowlist(raw string) (map[int64]struct{}, error) { + if len(raw) > 16384 { + return nil, fmt.Errorf("cyber_policy_user_allowlist must not exceed 16384 bytes") + } + ids := make(map[int64]struct{}) + for _, field := range strings.FieldsFunc(raw, func(r rune) bool { return r == ',' || unicode.IsSpace(r) }) { + id, err := strconv.ParseInt(field, 10, 64) + if err != nil || id <= 0 { + return nil, fmt.Errorf("cyber_policy_user_allowlist must contain positive user IDs separated by commas or whitespace") + } + ids[id] = struct{}{} + } + return ids, nil +} + +func (s *SettingService) IsCyberPolicyUserAllowlisted(ctx context.Context, userID int64) bool { + if s == nil || s.settingRepo == nil || userID <= 0 { + return false + } + s.GetCyberSessionBlockRuntime(ctx) + if cached, ok := s.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime); ok && cached != nil { + _, allowed := cached.allowlistedUsers[userID] + return allowed + } + return false +} + +func (s *OpenAIGatewayService) CyberPolicyLogOnly(ctx context.Context, apiKey *APIKey) bool { + return s != nil && apiKey != nil && s.settingService.IsCyberPolicyUserAllowlisted(ctx, apiKey.UserID) +} diff --git a/backend/internal/service/cyber_policy_allowlist_test.go b/backend/internal/service/cyber_policy_allowlist_test.go new file mode 100644 index 0000000000..32c9451825 --- /dev/null +++ b/backend/internal/service/cyber_policy_allowlist_test.go @@ -0,0 +1,62 @@ +package service + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestParseCyberPolicyUserAllowlist(t *testing.T) { + ids, err := ParseCyberPolicyUserAllowlist("12, 34\n12\t56") + require.NoError(t, err) + require.Len(t, ids, 3) + for _, raw := range []string{"0", "-1", "12,invalid", "9223372036854775808", strings.Repeat("1", 16385)} { + ids, err := ParseCyberPolicyUserAllowlist(raw) + require.Error(t, err) + require.Nil(t, ids) + } + ids, err = ParseCyberPolicyUserAllowlist("") + require.NoError(t, err) + require.Empty(t, ids) +} + +func TestCyberPolicyAllowlistCoversAllUserKeysAndRefreshes(t *testing.T) { + repo := &fakeSettingRepo{vals: map[string]string{SettingKeyCyberPolicyUserAllowlist: "12"}} + settings := &SettingService{settingRepo: repo} + svc := &OpenAIGatewayService{settingService: settings} + ctx := context.Background() + // The allowlist also applies when session blocking is disabled. + require.True(t, svc.CyberPolicyLogOnly(ctx, &APIKey{ID: 1, UserID: 12})) + require.True(t, svc.CyberPolicyLogOnly(ctx, &APIKey{ID: 2, UserID: 12})) + require.False(t, svc.CyberPolicyLogOnly(ctx, &APIKey{ID: 3, UserID: 34})) + require.False(t, svc.CyberPolicyLogOnly(ctx, nil)) + repo.vals[SettingKeyCyberPolicyUserAllowlist] = "34" + settings.cyberSessionBlockRuntimeCache.Store(&cachedCyberSessionBlockRuntime{}) + require.False(t, svc.CyberPolicyLogOnly(ctx, &APIKey{ID: 1, UserID: 12})) + require.True(t, svc.CyberPolicyLogOnly(ctx, &APIKey{ID: 3, UserID: 34})) +} + +func TestRecordCyberPolicyEventLogOnlyPreservesEvidenceWithoutSideEffects(t *testing.T) { + repo := &banCountArgsTestRepo{} + svc := NewContentModerationService( + &contentModerationTestSettingRepo{values: map[string]string{SettingKeyRiskControlEnabled: "true"}}, + repo, nil, nil, nil, nil, nil, &EmailService{}, + ) + // An unconfigured email service must never be invoked in log-only mode. + svc.RecordCyberPolicyEvent(context.Background(), CyberPolicyRecordInput{ + LogOnly: true, UserID: 12, UserEmail: "test@example.com", Model: "gpt-5", + UpstreamMessage: "blocked", UpstreamBody: `{"error":{"code":"cyber_policy"}}`, + }) + logs := repo.snapshotLogs() + require.Len(t, logs, 1) + require.Equal(t, ContentModerationModeCyberLogOnly, logs[0].Mode) + require.Equal(t, ContentModerationActionCyberPolicy, logs[0].Action) + require.True(t, logs[0].Flagged) + require.Contains(t, logs[0].Error, "cyber_policy") + require.False(t, logs[0].AutoBanned) + require.False(t, logs[0].EmailSent) + require.Zero(t, logs[0].ViolationCount) + require.Empty(t, repo.snapshotCountCalls()) +} diff --git a/backend/internal/service/domain_constants.go b/backend/internal/service/domain_constants.go index 9118d28316..82477d3808 100644 --- a/backend/internal/service/domain_constants.go +++ b/backend/internal/service/domain_constants.go @@ -240,6 +240,7 @@ const ( SettingKeyRiskControlEnabled = "risk_control_enabled" // 是否启用风控中心入口与审计链路 SettingKeyContentModerationConfig = "content_moderation_config" // 内容审计配置（JSON） SettingKeyCyberSessionBlockEnabled = "cyber_session_block_enabled" // cyber 命中后会话级自动屏蔽总开关(默认关) + SettingKeyCyberPolicyUserAllowlist = "cyber_policy_user_allowlist" // Platform user IDs with log-only cyber handling SettingKeyCyberSessionBlockTTLSeconds = "cyber_session_block_ttl_seconds" // 会话屏蔽 TTL 秒数(默认 3600) SettingKeyLoginAgreementEnabled = "login_agreement_enabled" // 登录前是否要求同意条款 SettingKeyLoginAgreementMode = "login_agreement_mode" // 条款确认展示模式：modal / checkbox diff --git a/backend/internal/service/risk_control_allowlist_test.go b/backend/internal/service/risk_control_allowlist_test.go new file mode 100644 index 0000000000..8019c1cdd8 --- /dev/null +++ b/backend/internal/service/risk_control_allowlist_test.go @@ -0,0 +1,153 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type failingAllowlistRepo struct { + fakeSettingRepo + failKey string +} + +func (r *failingAllowlistRepo) GetValue(ctx context.Context, key string) (string, error) { + if key == r.failKey || r.failKey == "all" { + return "", errors.New("database unavailable") + } + return r.fakeSettingRepo.GetValue(ctx, key) +} + +func TestRiskControlAllowlistRetainsLastGoodValueAndRetries(t *testing.T) { + for _, failKey := range []string{SettingKeyCyberPolicyUserAllowlist, "all"} { + t.Run(failKey, func(t *testing.T) { + repo := &failingAllowlistRepo{fakeSettingRepo: fakeSettingRepo{vals: map[string]string{SettingKeyCyberPolicyUserAllowlist: "12"}}} + svc := &SettingService{settingRepo: repo} + ctx := context.Background() + require.True(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12)) + expire := func() { + cached, ok := svc.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime) + require.True(t, ok) + old := *cached + old.expiresAt = 0 + svc.cyberSessionBlockRuntimeCache.Store(&old) + } + expire() + repo.failKey = failKey + require.True(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12)) + cached, ok := svc.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime) + require.True(t, ok) + require.LessOrEqual(t, time.Until(time.Unix(0, cached.expiresAt)), cyberSessionBlockRuntimeErrorTTL) + repo.failKey = "" + repo.vals[SettingKeyCyberPolicyUserAllowlist] = "" + expire() + require.False(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12), "successful removal must replace stale membership") + }) + } +} + +func TestRiskControlAllowlistAuditsWithoutLocalPenalties(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(moderationAPIResponse{Results: []moderationAPIResult{{CategoryScores: map[string]float64{"sexual": 0.99}}}}) + })) + defer server.Close() + for _, path := range []string{"keyword", "hash", "pre_block", "observe"} { + t.Run(path, func(t *testing.T) { + cfg := defaultContentModerationConfig() + cfg.Enabled, cfg.AutoBanEnabled, cfg.EmailOnHit = true, true, true + cfg.BanThreshold = 1 + cfg.Mode = ContentModerationModePreBlock + cfg.BaseURL, cfg.APIKeys = server.URL, []string{"sk-test"} + if path == "observe" { + cfg.Mode = ContentModerationModeObserve + } + if path == "keyword" { + cfg.BlockedKeywords = []string{"blocked prompt"} + } + hashes := &contentModerationTestHashCache{hashes: map[string]struct{}{}} + if path == "hash" { + cfg.PreHashCheckEnabled = true + hashes.hashes[(ContentModerationInput{Text: "blocked prompt"}).Hash()] = struct{}{} + } + raw, err := json.Marshal(cfg) + require.NoError(t, err) + repo := &banCountArgsTestRepo{} + svc := &ContentModerationService{ + settingRepo: &contentModerationTestSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", SettingKeyContentModerationConfig: string(raw), SettingKeyCyberPolicyUserAllowlist: "12", + }}, + repo: repo, hashCache: hashes, emailService: &EmailService{}, + httpClient: server.Client(), asyncQueue: make(chan contentModerationTask, 4), keyHealth: make(map[string]*contentModerationKeyHealth), + } + decision, err := svc.Check(context.Background(), ContentModerationCheckInput{ + UserID: 12, UserEmail: "trusted@example.com", Protocol: ContentModerationProtocolOpenAIChat, + Body: []byte(`{"messages":[{"role":"user","content":"blocked prompt"}]}`), + }) + require.NoError(t, err) + require.True(t, decision.Allowed) + require.False(t, decision.Blocked) + var task contentModerationTask + select { + case task = <-svc.asyncQueue: + case <-time.After(time.Second): + t.Fatal("expected an audit task") + } + if task.log != nil { + svc.persistContentModerationLog(context.Background(), task.config, task.log, task.inputHash, task.recordHash, task.applySideEffects) + } else { + delay := 1 + svc.checkSync(context.Background(), task.input, svc.runtimeSnapshot.Load().config, task.content, task.inputHash, &delay, false) + } + logs := repo.snapshotLogs() + require.Len(t, logs, 1) + require.True(t, logs[0].Flagged) + require.Equal(t, ContentModerationModeRiskControlLogOnly, logs[0].Mode) + require.False(t, logs[0].AutoBanned) + require.False(t, logs[0].EmailSent) + require.Zero(t, logs[0].ViolationCount) + require.Empty(t, repo.snapshotCountCalls()) + require.Zero(t, svc.preBlockBlocked.Load()) + if path != "hash" { + require.Empty(t, hashes.hashes) + } + }) + } +} + +type failingModerationAllowlistRepo struct { + contentModerationTestSettingRepo + fail bool +} + +func (r *failingModerationAllowlistRepo) GetMultiple(ctx context.Context, keys []string) (map[string]string, error) { + if r.fail { + return nil, errors.New("database unavailable") + } + return r.contentModerationTestSettingRepo.GetMultiple(ctx, keys) +} + +func TestRiskControlAllowlistSnapshotRetainsMembershipOnFailure(t *testing.T) { + repo := &failingModerationAllowlistRepo{contentModerationTestSettingRepo: contentModerationTestSettingRepo{values: map[string]string{ + SettingKeyRiskControlEnabled: "true", SettingKeyCyberPolicyUserAllowlist: "12", + }}} + svc := &ContentModerationService{settingRepo: repo} + ctx := context.Background() + _, err := svc.refreshRuntimeSnapshot(ctx) + require.NoError(t, err) + repo.fail = true + _, err = svc.refreshRuntimeSnapshot(ctx) + require.Error(t, err) + require.Contains(t, svc.runtimeSnapshot.Load().allowlistedUsers, int64(12)) + repo.fail = false + repo.values[SettingKeyCyberPolicyUserAllowlist] = "34" + _, err = svc.refreshRuntimeSnapshot(ctx) + require.NoError(t, err) + require.NotContains(t, svc.runtimeSnapshot.Load().allowlistedUsers, int64(12)) + require.Contains(t, svc.runtimeSnapshot.Load().allowlistedUsers, int64(34)) +} diff --git a/backend/internal/service/setting_cyber_allowlist_test.go b/backend/internal/service/setting_cyber_allowlist_test.go new file mode 100644 index 0000000000..7c0c6752ac --- /dev/null +++ b/backend/internal/service/setting_cyber_allowlist_test.go @@ -0,0 +1,42 @@ +//go:build unit + +package service + +import ( + "context" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/stretchr/testify/require" +) + +type cyberAllowlistSettingsRepo struct { + authSourceDefaultsRepoStub +} + +func (r *cyberAllowlistSettingsRepo) GetValue(_ context.Context, key string) (string, error) { + value, ok := r.values[key] + if !ok { + return "", ErrSettingNotFound + } + return value, nil +} + +func TestUpdateSettingsCyberAllowlistRoundTripAndImmediateRefresh(t *testing.T) { + ctx := context.Background() + repo := &cyberAllowlistSettingsRepo{authSourceDefaultsRepoStub{values: map[string]string{}}} + svc := NewSettingService(repo, &config.Config{}) + resetGatewayForwardingSettingsCacheForTest(t) + defer svc.refreshCachedSettings(&SystemSettings{}) + require.False(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12)) + settings := &SystemSettings{CyberPolicyUserAllowlist: "12, 34"} + require.NoError(t, svc.UpdateSettings(ctx, settings)) + require.True(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12)) + require.Equal(t, "12, 34", svc.parseSettings(repo.values).CyberPolicyUserAllowlist) + settings.CyberPolicyUserAllowlist = "12, invalid" + require.Error(t, svc.UpdateSettings(ctx, settings)) + require.Equal(t, "12, 34", repo.values[SettingKeyCyberPolicyUserAllowlist]) + settings.CyberPolicyUserAllowlist = "" + require.NoError(t, svc.UpdateSettings(ctx, settings)) + require.False(t, svc.IsCyberPolicyUserAllowlisted(ctx, 12)) +} diff --git a/backend/internal/service/setting_gateway_runtime.go b/backend/internal/service/setting_gateway_runtime.go index 8f81c1c0bb..ec6d21d502 100644 --- a/backend/internal/service/setting_gateway_runtime.go +++ b/backend/internal/service/setting_gateway_runtime.go @@ -152,12 +152,13 @@ type cachedCodexRestrictionPolicy struct { expiresAt int64 // unix nano } -// cachedCyberSessionBlockRuntime cyber 会话屏蔽开关+TTL 进程内缓存（60s TTL）。 +// cachedCyberSessionBlockRuntime cyber 会话屏蔽开关、TTL 与用户白名单缓存（60s TTL）。 // GetCyberSessionBlockRuntime 在网关请求热路径上被调用，避免每次访问 DB。 type cachedCyberSessionBlockRuntime struct { - enabled bool - ttl time.Duration - expiresAt int64 // unix nano + allowlistedUsers map[int64]struct{} + enabled bool + ttl time.Duration + expiresAt int64 // unix nano } const cyberSessionBlockRuntimeCacheTTL = 60 * time.Second @@ -172,7 +173,7 @@ const openAIQuotaAutoPauseSettingsRefreshKey = "openai_quota_auto_pause_settings // GetCyberSessionBlockRuntime 返回 (开关, TTL)，进程内缓存 ~60s， // 供网关热路径读取时避免 DB 往返。 -// 两个 setting key 在单次 singleflight 里一起读取，减少 DB 往返。 +// 屏蔽设置与用户白名单在单次 singleflight 中一起读取。 // 默认值：开关 false，TTL 1h（与粘性会话对齐）。 func (s *SettingService) GetCyberSessionBlockRuntime(ctx context.Context) (bool, time.Duration) { if cached, ok := s.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime); ok && cached != nil { @@ -181,6 +182,8 @@ func (s *SettingService) GetCyberSessionBlockRuntime(ctx context.Context) (bool, } } result, _, _ := s.cyberSessionBlockRuntimeSF.Do("cyber_session_block_runtime", func() (any, error) { + s.cyberSessionBlockRuntimeMu.Lock() + defer s.cyberSessionBlockRuntimeMu.Unlock() if cached, ok := s.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime); ok && cached != nil { if time.Now().UnixNano() < cached.expiresAt { return cached, nil @@ -192,15 +195,11 @@ func (s *SettingService) GetCyberSessionBlockRuntime(ctx context.Context) (bool, enabledVal, enabledErr := s.settingRepo.GetValue(dbCtx, SettingKeyCyberSessionBlockEnabled) ttlVal, ttlErr := s.settingRepo.GetValue(dbCtx, SettingKeyCyberSessionBlockTTLSeconds) + previous, _ := s.cyberSessionBlockRuntimeCache.Load().(*cachedCyberSessionBlockRuntime) + cacheTTL := cyberSessionBlockRuntimeCacheTTL if enabledErr != nil && !errors.Is(enabledErr, ErrSettingNotFound) { slog.Warn("failed to get cyber_session_block_enabled setting", "error", enabledErr) - entry := &cachedCyberSessionBlockRuntime{ - enabled: false, - ttl: time.Hour, - expiresAt: time.Now().Add(cyberSessionBlockRuntimeErrorTTL).UnixNano(), - } - s.cyberSessionBlockRuntimeCache.Store(entry) - return entry, nil + cacheTTL = cyberSessionBlockRuntimeErrorTTL } enabled := enabledErr == nil && strings.TrimSpace(enabledVal) == "true" @@ -212,10 +211,23 @@ func (s *SettingService) GetCyberSessionBlockRuntime(ctx context.Context) (bool, } } + allowlistVal, allowlistErr := s.settingRepo.GetValue(dbCtx, SettingKeyCyberPolicyUserAllowlist) + var allowlistedUsers map[int64]struct{} + if allowlistErr == nil { + allowlistedUsers, allowlistErr = ParseCyberPolicyUserAllowlist(allowlistVal) + } + if allowlistErr != nil && !errors.Is(allowlistErr, ErrSettingNotFound) { + slog.Warn("failed to load risk control user allowlist", "error", allowlistErr) + cacheTTL = cyberSessionBlockRuntimeErrorTTL + if previous != nil { + allowlistedUsers = previous.allowlistedUsers + } + } entry := &cachedCyberSessionBlockRuntime{ - enabled: enabled, - ttl: ttl, - expiresAt: time.Now().Add(cyberSessionBlockRuntimeCacheTTL).UnixNano(), + allowlistedUsers: allowlistedUsers, + enabled: enabled, + ttl: ttl, + expiresAt: time.Now().Add(cacheTTL).UnixNano(), } s.cyberSessionBlockRuntimeCache.Store(entry) return entry, nil diff --git a/backend/internal/service/setting_parse.go b/backend/internal/service/setting_parse.go index 7c62451d33..f93f4b1b30 100644 --- a/backend/internal/service/setting_parse.go +++ b/backend/internal/service/setting_parse.go @@ -220,6 +220,7 @@ func (s *SettingService) InitializeDefaultSettings(ctx context.Context) error { // cyber 会话屏蔽（默认关闭，TTL 默认 3600s） SettingKeyCyberSessionBlockEnabled: "false", + SettingKeyCyberPolicyUserAllowlist: "", SettingKeyCyberSessionBlockTTLSeconds: "3600", // Claude Code version check (default: empty = disabled) @@ -843,6 +844,7 @@ func (s *SettingService) parseSettings(settings map[string]string) *SystemSettin // cyber 会话屏蔽（默认关闭，TTL 默认 3600s） result.CyberSessionBlockEnabled = settings[SettingKeyCyberSessionBlockEnabled] == "true" + result.CyberPolicyUserAllowlist = settings[SettingKeyCyberPolicyUserAllowlist] if v, err := strconv.Atoi(strings.TrimSpace(settings[SettingKeyCyberSessionBlockTTLSeconds])); err == nil && v > 0 { result.CyberSessionBlockTTLSeconds = v } else { diff --git a/backend/internal/service/setting_service.go b/backend/internal/service/setting_service.go index 6559249ada..7aea7c8ae9 100644 --- a/backend/internal/service/setting_service.go +++ b/backend/internal/service/setting_service.go @@ -135,6 +135,7 @@ type SettingService struct { codexRestrictionPolicyCache atomic.Value // *cachedCodexRestrictionPolicy codexRestrictionPolicySF singleflight.Group + cyberSessionBlockRuntimeMu sync.Mutex cyberSessionBlockRuntimeCache atomic.Value // *cachedCyberSessionBlockRuntime cyberSessionBlockRuntimeSF singleflight.Group diff --git a/backend/internal/service/setting_update.go b/backend/internal/service/setting_update.go index 421b226524..caf3dd87e9 100644 --- a/backend/internal/service/setting_update.go +++ b/backend/internal/service/setting_update.go @@ -449,6 +449,10 @@ func (s *SettingService) buildSystemSettingsUpdates(ctx context.Context, setting // cyber 会话屏蔽开关 + TTL updates[SettingKeyCyberSessionBlockEnabled] = strconv.FormatBool(settings.CyberSessionBlockEnabled) + if _, err := ParseCyberPolicyUserAllowlist(settings.CyberPolicyUserAllowlist); err != nil { + return nil, err + } + updates[SettingKeyCyberPolicyUserAllowlist] = settings.CyberPolicyUserAllowlist if settings.CyberSessionBlockTTLSeconds > 0 { updates[SettingKeyCyberSessionBlockTTLSeconds] = strconv.Itoa(settings.CyberSessionBlockTTLSeconds) } @@ -800,6 +804,11 @@ func (s *SettingService) refreshCachedSettings(settings *SystemSettings) { // codex_cli_only 加固策略缓存：设置更新后强制下次重载（涉及 4 个键 + JSON 解析，直接置过期）。 s.codexRestrictionPolicySF.Forget("codex_restriction_policy") s.codexRestrictionPolicyCache.Store(&cachedCodexRestrictionPolicy{expiresAt: 0}) + // Retain the successfully saved allowlist if the next DB refresh fails. + s.cyberSessionBlockRuntimeMu.Lock() + allowlistedUsers, _ := ParseCyberPolicyUserAllowlist(settings.CyberPolicyUserAllowlist) + s.cyberSessionBlockRuntimeCache.Store(&cachedCyberSessionBlockRuntime{allowlistedUsers: allowlistedUsers}) + s.cyberSessionBlockRuntimeMu.Unlock() if s.onUpdate != nil { s.onUpdate() // Invalidate cache after settings update } diff --git a/backend/internal/service/settings_view.go b/backend/internal/service/settings_view.go index 883d680319..bf2e840baa 100644 --- a/backend/internal/service/settings_view.go +++ b/backend/internal/service/settings_view.go @@ -169,6 +169,7 @@ type SystemSettings struct { DefaultBalance float64 RiskControlEnabled bool CyberSessionBlockEnabled bool + CyberPolicyUserAllowlist string CyberSessionBlockTTLSeconds int AffiliateEnabled bool AffiliateRebateRate float64 diff --git a/frontend/src/api/admin/settings.ts b/frontend/src/api/admin/settings.ts index 7e551c9c76..1fcc0021b1 100644 --- a/frontend/src/api/admin/settings.ts +++ b/frontend/src/api/admin/settings.ts @@ -656,6 +656,7 @@ export interface SystemSettings { risk_control_enabled: boolean; // Cyber session block + cyber_policy_user_allowlist: string; cyber_session_block_enabled: boolean; cyber_session_block_ttl_seconds: number; @@ -976,6 +977,7 @@ export interface UpdateSettingsRequest { risk_control_enabled?: boolean; // Cyber session block + cyber_policy_user_allowlist?: string; cyber_session_block_enabled?: boolean; cyber_session_block_ttl_seconds?: number; diff --git a/frontend/src/i18n/locales/en/admin/settings.ts b/frontend/src/i18n/locales/en/admin/settings.ts index ced6771230..680669024c 100644 --- a/frontend/src/i18n/locales/en/admin/settings.ts +++ b/frontend/src/i18n/locales/en/admin/settings.ts @@ -84,6 +84,8 @@ export default { enabledHint: 'When off, the admin sidebar entry is hidden and gateway moderation is skipped.', cyberSessionBlock: 'Cyber session auto-block', cyberSessionBlockHint: 'When enabled, sessions hit by upstream cyber_policy are blocked locally for the TTL and no longer forwarded. Only the offending session is blocked; other sessions on the same key are unaffected.', + riskControlUserAllowlist: 'Risk control allowlist', + riskControlUserAllowlistHint: 'Enter any email keyword to search for matching users. Allowlisted users will not trigger account bans or local blocking, but upstream restrictions still apply. This feature is typically used for trusted downstream relays.', cyberSessionBlockTTL: 'Block TTL (seconds)', }, affiliate: { diff --git a/frontend/src/i18n/locales/zh/admin/settings.ts b/frontend/src/i18n/locales/zh/admin/settings.ts index 39007377c4..e31e3dc6a7 100644 --- a/frontend/src/i18n/locales/zh/admin/settings.ts +++ b/frontend/src/i18n/locales/zh/admin/settings.ts @@ -84,6 +84,8 @@ export default { enabledHint: '关闭后管理员侧边栏入口隐藏，网关内容审计不会执行。', cyberSessionBlock: 'cyber 会话自动屏蔽', cyberSessionBlockHint: '开启后,被上游网络安全策略(cyber_policy)拦截的会话将在 TTL 内被本地屏蔽,不再发往上游。仅屏蔽该会话,不影响同 Key 其他会话。', + riskControlUserAllowlist: '风控白名单', + riskControlUserAllowlistHint: '输入任意邮箱关键词进行模糊搜索。 白名单中的用户不会触发封号或本地屏蔽，但仍然无法突破上游拦截。该功能通常用于可信的下游中转站。', cyberSessionBlockTTL: '屏蔽时长(秒)', }, affiliate: { diff --git a/frontend/src/views/admin/SettingsView.vue b/frontend/src/views/admin/SettingsView.vue index 9f04a341da..1b1c914ec0 100644 --- a/frontend/src/views/admin/SettingsView.vue +++ b/frontend/src/views/admin/SettingsView.vue @@ -7461,6 +7461,18 @@ +
+ + +

+ {{ t('admin.settings.features.riskControl.riskControlUserAllowlistHint') }} +

+
+