mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 03:52:26 +08:00
feat: add risk control user allowlist
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -135,6 +135,7 @@ type SettingService struct {
|
||||
codexRestrictionPolicyCache atomic.Value // *cachedCodexRestrictionPolicy
|
||||
codexRestrictionPolicySF singleflight.Group
|
||||
|
||||
cyberSessionBlockRuntimeMu sync.Mutex
|
||||
cyberSessionBlockRuntimeCache atomic.Value // *cachedCyberSessionBlockRuntime
|
||||
cyberSessionBlockRuntimeSF singleflight.Group
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -169,6 +169,7 @@ type SystemSettings struct {
|
||||
DefaultBalance float64
|
||||
RiskControlEnabled bool
|
||||
CyberSessionBlockEnabled bool
|
||||
CyberPolicyUserAllowlist string
|
||||
CyberSessionBlockTTLSeconds int
|
||||
AffiliateEnabled bool
|
||||
AffiliateRebateRate float64
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -84,6 +84,8 @@ export default {
|
||||
enabledHint: '关闭后管理员侧边栏入口隐藏,网关内容审计不会执行。',
|
||||
cyberSessionBlock: 'cyber 会话自动屏蔽',
|
||||
cyberSessionBlockHint: '开启后,被上游网络安全策略(cyber_policy)拦截的会话将在 TTL 内被本地屏蔽,不再发往上游。仅屏蔽该会话,不影响同 Key 其他会话。',
|
||||
riskControlUserAllowlist: '风控白名单',
|
||||
riskControlUserAllowlistHint: '输入任意邮箱关键词进行模糊搜索。 白名单中的用户不会触发封号或本地屏蔽,但仍然无法突破上游拦截。该功能通常用于可信的下游中转站。',
|
||||
cyberSessionBlockTTL: '屏蔽时长(秒)',
|
||||
},
|
||||
affiliate: {
|
||||
|
||||
@@ -7461,6 +7461,18 @@
|
||||
<Toggle v-model="form.risk_control_enabled" />
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<label class="input-label">
|
||||
{{ t('admin.settings.features.riskControl.riskControlUserAllowlist') }}
|
||||
</label>
|
||||
<OpenAIFastPolicyUserSelector
|
||||
v-model="riskControlAllowlistedUserIds"
|
||||
/>
|
||||
<p class="mt-1 text-xs text-gray-500 dark:text-gray-400">
|
||||
{{ t('admin.settings.features.riskControl.riskControlUserAllowlistHint') }}
|
||||
</p>
|
||||
</div>
|
||||
|
||||
<div class="flex items-center justify-between">
|
||||
<div>
|
||||
<label class="text-sm font-medium text-gray-700 dark:text-gray-300">
|
||||
@@ -9779,6 +9791,7 @@ const form = reactive<SettingsForm>({
|
||||
hide_ccs_import_button: false,
|
||||
payment_enabled: false,
|
||||
risk_control_enabled: false,
|
||||
cyber_policy_user_allowlist: "",
|
||||
cyber_session_block_enabled: false,
|
||||
cyber_session_block_ttl_seconds: 3600,
|
||||
payment_min_amount: 1,
|
||||
@@ -10040,6 +10053,19 @@ function applyCaptchaSelection(provider: CaptchaProviderSelection | null): void
|
||||
form.aliyun_captcha_enabled = provider === "aliyun";
|
||||
}
|
||||
|
||||
// Keep the settings API representation as user IDs; the selector displays emails.
|
||||
const riskControlAllowlistedUserIds = computed<number[]>({
|
||||
get: () => Array.from(new Set(
|
||||
form.cyber_policy_user_allowlist
|
||||
.split(/[,\s]+/)
|
||||
.map(Number)
|
||||
.filter((id) => Number.isSafeInteger(id) && id > 0),
|
||||
)),
|
||||
set: (ids) => {
|
||||
form.cyber_policy_user_allowlist = ids.join(",");
|
||||
},
|
||||
});
|
||||
|
||||
const captchaMasterEnabled = computed({
|
||||
get: () =>
|
||||
form.turnstile_enabled ||
|
||||
@@ -11638,6 +11664,7 @@ async function saveSettings() {
|
||||
// Payment configuration
|
||||
payment_enabled: form.payment_enabled,
|
||||
risk_control_enabled: form.risk_control_enabled,
|
||||
cyber_policy_user_allowlist: form.cyber_policy_user_allowlist,
|
||||
cyber_session_block_enabled: form.cyber_session_block_enabled,
|
||||
cyber_session_block_ttl_seconds:
|
||||
Number(form.cyber_session_block_ttl_seconds) || 3600,
|
||||
|
||||
Reference in New Issue
Block a user