diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index dedf06d46..92300337b 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -46,10 +46,8 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { return nil, err } userRepository := repository.NewUserRepository(client, db) - passkeyRepository := repository.NewPasskeyRepository(db) redeemCodeRepository := repository.NewRedeemCodeRepository(client) redisClient := repository.ProvideRedis(configConfig) - passkeySessionStore := repository.NewPasskeySessionStore(redisClient) refreshTokenCache := repository.NewRefreshTokenCache(redisClient) settingRepository := repository.NewSettingRepository(client) groupRepository := repository.NewGroupRepository(client, db) @@ -81,10 +79,6 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { affiliateRepository := repository.NewAffiliateRepository(client, db) affiliateService := service.NewAffiliateService(affiliateRepository, settingService, apiKeyAuthCacheInvalidator, billingCacheService) authService := service.NewAuthService(client, userRepository, redeemCodeRepository, refreshTokenCache, configConfig, settingService, emailService, turnstileService, emailQueueService, promoService, subscriptionService, affiliateService, serviceUserPlatformQuotaRepository) - passkeyService, err := service.NewPasskeyService(configConfig, passkeyRepository, passkeySessionStore, userRepository) - if err != nil { - return nil, err - } userService := service.NewUserService(userRepository, settingRepository, apiKeyAuthCacheInvalidator, billingCache) redeemCache := repository.NewRedeemCache(redisClient) redeemService := service.NewRedeemService(redeemCodeRepository, userRepository, subscriptionService, redeemCache, billingCacheService, client, apiKeyAuthCacheInvalidator, affiliateService) @@ -98,7 +92,6 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { userAttributeValueRepository := repository.NewUserAttributeValueRepository(client) userAttributeService := service.NewUserAttributeService(userAttributeDefinitionRepository, userAttributeValueRepository) authHandler := handler.NewAuthHandler(configConfig, authService, userService, settingService, promoService, redeemService, totpService, userAttributeService) - passkeyHandler := handler.NewPasskeyHandler(passkeyService, authService, settingService) userHandler := handler.NewUserHandler(userService, authService, emailService, emailCache, affiliateService, serviceUserPlatformQuotaRepository) apiKeyHandler := handler.NewAPIKeyHandler(apiKeyService) usageLogRepository := repository.NewUsageLogRepository(client, db) @@ -260,7 +253,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { channelMonitorRequestTemplateHandler := admin.NewChannelMonitorRequestTemplateHandler(channelMonitorRequestTemplateService) contentModerationRepository := repository.NewContentModerationRepository(db) contentModerationHashCache := repository.NewContentModerationHashCache(redisClient) - contentModerationService := service.NewContentModerationService(settingRepository, contentModerationRepository, contentModerationHashCache, groupRepository, userRepository, apiKeyAuthCacheInvalidator, emailService) + contentModerationService := service.NewContentModerationService(settingRepository, contentModerationRepository, contentModerationHashCache, groupRepository, userRepository, proxyRepository, apiKeyAuthCacheInvalidator, emailService) contentModerationHandler := admin.NewContentModerationHandler(contentModerationService) configManager := securityaudit.NewConfigManager(db, settingRepository, redisClient, secretEncryptor, configConfig) postgreSQLRepository := securityaudit.NewPostgreSQLRepository(db) @@ -287,6 +280,13 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { openAIGatewayHandler := handler.ProvideOpenAIGatewayHandler(openAIGatewayService, concurrencyService, billingCacheService, apiKeyService, usageRecordWorkerPool, errorPassthroughService, contentModerationService, opsService, grokQuotaService, configConfig, coordinator) handlerSettingHandler := handler.ProvideSettingHandler(settingService, buildInfo, notificationEmailService) totpHandler := handler.NewTotpHandler(totpService) + passkeyRepository := repository.NewPasskeyRepository(db) + passkeySessionStore := repository.NewPasskeySessionStore(redisClient) + passkeyService, err := service.NewPasskeyService(configConfig, passkeyRepository, passkeySessionStore, userRepository) + if err != nil { + return nil, err + } + passkeyHandler := handler.NewPasskeyHandler(passkeyService, authService, settingService) handlerPaymentHandler := handler.NewPaymentHandler(paymentService, paymentConfigService) paymentWebhookHandler := handler.NewPaymentWebhookHandler(paymentService, registry) availableChannelHandler := handler.NewAvailableChannelHandler(channelService, apiKeyService, settingService) diff --git a/backend/internal/handler/admin/content_moderation_handler.go b/backend/internal/handler/admin/content_moderation_handler.go index 00a8dca05..cafcaeec4 100644 --- a/backend/internal/handler/admin/content_moderation_handler.go +++ b/backend/internal/handler/admin/content_moderation_handler.go @@ -20,10 +20,12 @@ func NewContentModerationHandler(svc *service.ContentModerationService) *Content } type contentModerationConfigRequest struct { - Enabled *bool `json:"enabled"` - Mode *string `json:"mode"` - BaseURL *string `json:"base_url"` - Model *string `json:"model"` + Enabled *bool `json:"enabled"` + Mode *string `json:"mode"` + BaseURL *string `json:"base_url"` + Model *string `json:"model"` + // 审计请求使用的代理服务器:null 不修改;0 清除(直连);>0 指定代理。 + ProxyID *int64 `json:"proxy_id"` APIKey *string `json:"api_key"` APIKeys *[]string `json:"api_keys"` APIKeysMode string `json:"api_keys_mode"` @@ -60,6 +62,7 @@ type contentModerationAPIKeyTestRequest struct { BaseURL string `json:"base_url"` Model string `json:"model"` TimeoutMS int `json:"timeout_ms"` + ProxyID *int64 `json:"proxy_id"` Prompt string `json:"prompt"` Images []string `json:"images"` } @@ -88,6 +91,7 @@ func (h *ContentModerationHandler) UpdateConfig(c *gin.Context) { Mode: req.Mode, BaseURL: req.BaseURL, Model: req.Model, + ProxyID: req.ProxyID, APIKey: req.APIKey, APIKeys: req.APIKeys, APIKeysMode: req.APIKeysMode, @@ -134,6 +138,7 @@ func (h *ContentModerationHandler) TestAPIKeys(c *gin.Context) { BaseURL: req.BaseURL, Model: req.Model, TimeoutMS: req.TimeoutMS, + ProxyID: req.ProxyID, Prompt: req.Prompt, Images: req.Images, }) diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 3d01bd15e..0f0964413 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -1266,6 +1266,7 @@ func TestOpenAIResponsesWebSocket_ContentModerationBlocksFirstFrame(t *testing.T nil, nil, nil, + nil, ) decision, err := moderationSvc.Check(context.Background(), service.ContentModerationCheckInput{ UserID: 1, diff --git a/backend/internal/service/content_moderation.go b/backend/internal/service/content_moderation.go index 62f12eeab..acf38bf49 100644 --- a/backend/internal/service/content_moderation.go +++ b/backend/internal/service/content_moderation.go @@ -21,6 +21,7 @@ import ( "time" infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/pkg/httpclient" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" "github.com/Wei-Shaw/sub2api/internal/pkg/servertiming" ) @@ -138,10 +139,12 @@ func ContentModerationCategories() []string { } type ContentModerationConfig struct { - Enabled bool `json:"enabled"` - Mode string `json:"mode"` - BaseURL string `json:"base_url"` - Model string `json:"model"` + Enabled bool `json:"enabled"` + Mode string `json:"mode"` + BaseURL string `json:"base_url"` + Model string `json:"model"` + // ProxyID 指定审计请求使用的代理服务器(IP管理-代理服务器),nil 表示直连。 + ProxyID *int64 `json:"proxy_id,omitempty"` APIKey string `json:"api_key,omitempty"` APIKeys []string `json:"api_keys,omitempty"` TimeoutMS int `json:"timeout_ms"` @@ -176,6 +179,7 @@ type ContentModerationConfigView struct { Mode string `json:"mode"` BaseURL string `json:"base_url"` Model string `json:"model"` + ProxyID *int64 `json:"proxy_id"` APIKeyConfigured bool `json:"api_key_configured"` APIKeyMasked string `json:"api_key_masked"` APIKeyCount int `json:"api_key_count"` @@ -240,8 +244,10 @@ type TestContentModerationAPIKeysInput struct { BaseURL string `json:"base_url"` Model string `json:"model"` TimeoutMS int `json:"timeout_ms"` - Prompt string `json:"prompt"` - Images []string `json:"images"` + // ProxyID nil 表示沿用已保存配置的代理;<=0 表示强制直连测试;>0 表示指定代理测试。 + ProxyID *int64 `json:"proxy_id"` + Prompt string `json:"prompt"` + Images []string `json:"images"` } type TestContentModerationAPIKeysResult struct { @@ -260,10 +266,12 @@ type ContentModerationTestAuditResult struct { } type UpdateContentModerationConfigInput struct { - Enabled *bool `json:"enabled"` - Mode *string `json:"mode"` - BaseURL *string `json:"base_url"` - Model *string `json:"model"` + Enabled *bool `json:"enabled"` + Mode *string `json:"mode"` + BaseURL *string `json:"base_url"` + Model *string `json:"model"` + // ProxyID nil 表示不修改;<=0 表示清除代理(恢复直连);>0 表示指定代理。 + ProxyID *int64 `json:"proxy_id"` APIKey *string `json:"api_key"` APIKeys *[]string `json:"api_keys"` APIKeysMode string `json:"api_keys_mode"` @@ -495,9 +503,11 @@ type ContentModerationService struct { hashCache ContentModerationHashCache groupRepo GroupRepository userRepo UserRepository + proxyRepo ProxyRepository authCacheInvalidator APIKeyAuthCacheInvalidator emailService *EmailService httpClient *http.Client + moderationProxyCache atomic.Pointer[moderationProxyURLCacheEntry] asyncQueue chan contentModerationTask workerCount int apiKeyCursor atomic.Uint64 @@ -566,6 +576,7 @@ func NewContentModerationService( hashCache ContentModerationHashCache, groupRepo GroupRepository, userRepo UserRepository, + proxyRepo ProxyRepository, authCacheInvalidator APIKeyAuthCacheInvalidator, emailService *EmailService, ) *ContentModerationService { @@ -575,6 +586,7 @@ func NewContentModerationService( hashCache: hashCache, groupRepo: groupRepo, userRepo: userRepo, + proxyRepo: proxyRepo, authCacheInvalidator: authCacheInvalidator, emailService: emailService, httpClient: servertiming.InstrumentClient(nil), @@ -616,6 +628,14 @@ func (s *ContentModerationService) UpdateConfig(ctx context.Context, input Updat if input.Model != nil { cfg.Model = strings.TrimSpace(*input.Model) } + if input.ProxyID != nil { + if *input.ProxyID > 0 { + id := *input.ProxyID + cfg.ProxyID = &id + } else { + cfg.ProxyID = nil + } + } if input.TimeoutMS != nil { cfg.TimeoutMS = *input.TimeoutMS } @@ -716,6 +736,8 @@ func (s *ContentModerationService) UpdateConfig(ctx context.Context, input Updat return nil, fmt.Errorf("save content moderation config: %w", err) } s.replaceRuntimeConfig(cfg, raw) + // 代理选择可能已变化,丢弃已解析的代理 URL 缓存,下次调用即时生效。 + s.moderationProxyCache.Store(nil) return s.configView(cfg), nil } @@ -739,6 +761,14 @@ func (s *ContentModerationService) TestAPIKeys(ctx context.Context, input TestCo if input.TimeoutMS > 0 { cfg.TimeoutMS = input.TimeoutMS } + if input.ProxyID != nil { + if *input.ProxyID > 0 { + id := *input.ProxyID + cfg.ProxyID = &id + } else { + cfg.ProxyID = nil + } + } cfg.normalize() testInput, imageCount, err := buildModerationTestInput(input.Prompt, input.Images) if err != nil { @@ -1631,6 +1661,11 @@ func (s *ContentModerationService) validateConfig(ctx context.Context, cfg *Cont if _, err := url.ParseRequestURI(cfg.BaseURL); err != nil { return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_BASE_URL", "OpenAI Base URL 无效") } + if cfg.ProxyID != nil && s.proxyRepo != nil { + if _, err := s.proxyRepo.GetByID(ctx, *cfg.ProxyID); err != nil { + return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_PROXY", fmt.Sprintf("代理服务器不存在: %d", *cfg.ProxyID)) + } + } if cfg.BlockStatus < 400 || cfg.BlockStatus > 599 { return infraerrors.BadRequest("INVALID_CONTENT_MODERATION_BLOCK_STATUS", "拦截 HTTP 状态码必须在 400-599 之间") } @@ -1723,9 +1758,9 @@ func (s *ContentModerationService) callModerationOnceWithInput(ctx context.Conte req.Header.Set("Authorization", "Bearer "+apiKey) req.Header.Set("Content-Type", "application/json") - client := s.httpClient - if client == nil { - client = http.DefaultClient + client, err := s.moderationHTTPClient(ctx, cfg) + if err != nil { + return nil, err } resp, err := client.Do(req) if err != nil { @@ -1750,6 +1785,73 @@ func (s *ContentModerationService) callModerationOnceWithInput(ctx context.Conte return &out.Results[0], nil } +// moderationProxyURLCacheEntry 缓存 proxy_id 到代理 URL 的解析结果, +// 避免审计热路径上每次调用都查询数据库。 +type moderationProxyURLCacheEntry struct { + proxyID int64 + url string + expiresAt time.Time +} + +const contentModerationProxyURLCacheTTL = time.Minute + +// moderationHTTPClient 返回本次审计调用应使用的 HTTP 客户端。 +// 未配置代理时沿用默认客户端;配置了代理时通过共享客户端池构建, +// 代理解析/构建失败直接返回错误,绝不回退直连(避免 IP 关联风险)。 +func (s *ContentModerationService) moderationHTTPClient(ctx context.Context, cfg *ContentModerationConfig) (*http.Client, error) { + if cfg == nil || cfg.ProxyID == nil { + if s.httpClient == nil { + return http.DefaultClient, nil + } + return s.httpClient, nil + } + proxyURL, err := s.resolveModerationProxyURL(ctx, *cfg.ProxyID) + if err != nil { + return nil, err + } + client, err := httpclient.GetClient(httpclient.Options{ProxyURL: proxyURL}) + if err != nil { + return nil, fmt.Errorf("build moderation proxy client: %w", err) + } + return client, nil +} + +func (s *ContentModerationService) resolveModerationProxyURL(ctx context.Context, proxyID int64) (string, error) { + now := time.Now() + prev := s.moderationProxyCache.Load() + if prev != nil && prev.proxyID == proxyID && now.Before(prev.expiresAt) { + return prev.url, nil + } + if s.proxyRepo == nil { + return "", errors.New("moderation proxy repository unavailable") + } + px, err := s.proxyRepo.GetByID(ctx, proxyID) + if err != nil { + return "", fmt.Errorf("resolve moderation proxy %d: %w", proxyID, err) + } + if !px.IsActive() || px.IsExpired(now) { + slog.Warn("content_moderation.proxy_not_active", + "proxy_id", proxyID, + "proxy_name", px.Name, + "status", px.Status, + "expired", px.IsExpired(now)) + } + proxyURL := px.URL() + if prev == nil || prev.proxyID != proxyID || prev.url != proxyURL { + // 不打印完整 URL(可能含认证信息),仅记录可定位的地址。 + slog.Info("content_moderation.proxy_enabled", + "proxy_id", proxyID, + "proxy_name", px.Name, + "proxy_addr", fmt.Sprintf("%s://%s:%d", px.Protocol, px.Host, px.Port)) + } + s.moderationProxyCache.Store(&moderationProxyURLCacheEntry{ + proxyID: proxyID, + url: proxyURL, + expiresAt: now.Add(contentModerationProxyURLCacheTTL), + }) + return proxyURL, nil +} + func (s *ContentModerationService) buildLog(input ContentModerationCheckInput, cfg *ContentModerationConfig, action string, flagged bool, highestCategory string, highestScore float64, scores map[string]float64, text string, latency *int, queueDelay *int, errText string) *ContentModerationLog { var userID *int64 if input.UserID > 0 { @@ -2013,6 +2115,7 @@ func cloneContentModerationConfig(cfg *ContentModerationConfig) *ContentModerati return nil } clone := *cfg + clone.ProxyID = cloneInt64Ptr(cfg.ProxyID) clone.APIKeys = append([]string(nil), cfg.APIKeys...) clone.GroupIDs = append([]int64(nil), cfg.GroupIDs...) clone.BlockedKeywords = append([]string(nil), cfg.BlockedKeywords...) @@ -2042,6 +2145,9 @@ func (cfg *ContentModerationConfig) normalize() { cfg.Model = defaultContentModerationModel } cfg.Model = strings.TrimSpace(cfg.Model) + if cfg.ProxyID != nil && *cfg.ProxyID <= 0 { + cfg.ProxyID = nil + } if cfg.TimeoutMS <= 0 { cfg.TimeoutMS = defaultContentModerationTimeoutMS } @@ -2305,6 +2411,7 @@ func (s *ContentModerationService) configView(cfg *ContentModerationConfig) *Con Mode: cfg.Mode, BaseURL: cfg.BaseURL, Model: cfg.Model, + ProxyID: cloneInt64Ptr(cfg.ProxyID), APIKeyConfigured: len(keys) > 0, APIKeyMasked: apiKeyMasked, APIKeyCount: len(keys), diff --git a/backend/internal/service/content_moderation_cyber_test.go b/backend/internal/service/content_moderation_cyber_test.go index 07ef2812d..3574b2b8f 100644 --- a/backend/internal/service/content_moderation_cyber_test.go +++ b/backend/internal/service/content_moderation_cyber_test.go @@ -76,6 +76,7 @@ func TestRecordCyberPolicyEvent_DisabledWhenRiskControlOff(t *testing.T) { nil, nil, nil, + nil, ) svc.RecordCyberPolicyEvent(context.Background(), CyberPolicyRecordInput{ @@ -102,6 +103,7 @@ func TestRecordCyberPolicyEvent_WritesLogWhenEnabled(t *testing.T) { nil, nil, nil, + nil, nil, // emailService=nil: email path safely skipped ) @@ -179,6 +181,7 @@ func TestRecordCyberPolicyEvent_CreateLogBeforeEmail(t *testing.T) { nil, nil, nil, + nil, nil, // emailService=nil: email path safely skipped; see doc comment above ) @@ -233,7 +236,7 @@ func TestApplyFlaggedAccountSideEffects_PassesExcludeCyberFlag(t *testing.T) { repo := &banCountArgsTestRepo{} svc := NewContentModerationService( &contentModerationTestSettingRepo{values: map[string]string{}}, - repo, nil, nil, nil, nil, nil, + repo, nil, nil, nil, nil, nil, nil, ) userID := int64(42) @@ -255,7 +258,7 @@ func TestRecordCyberPolicyEvent_ExcludeFromBanCount_SkipsBanJudgment(t *testing. SettingKeyRiskControlEnabled: "true", SettingKeyContentModerationConfig: `{"cyber_policy_exclude_from_ban_count":true}`, }}, - repo, nil, nil, nil, nil, nil, + repo, nil, nil, nil, nil, nil, nil, ) svc.RecordCyberPolicyEvent(context.Background(), CyberPolicyRecordInput{ @@ -282,7 +285,7 @@ func TestRecordCyberPolicyEvent_DefaultCountsTowardBan(t *testing.T) { &contentModerationTestSettingRepo{values: map[string]string{ SettingKeyRiskControlEnabled: "true", }}, - repo, nil, nil, nil, nil, nil, + repo, nil, nil, nil, nil, nil, nil, ) svc.RecordCyberPolicyEvent(context.Background(), CyberPolicyRecordInput{ diff --git a/backend/internal/service/content_moderation_proxy_test.go b/backend/internal/service/content_moderation_proxy_test.go new file mode 100644 index 000000000..0778530fc --- /dev/null +++ b/backend/internal/service/content_moderation_proxy_test.go @@ -0,0 +1,303 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" +) + +// contentModerationTestProxyRepo 仅实现审计代理路径用到的 GetByID,其余方法不应被调用。 +type contentModerationTestProxyRepo struct { + proxies map[int64]*Proxy + getByIDErr error + getCalls atomic.Int64 +} + +func (r *contentModerationTestProxyRepo) GetByID(ctx context.Context, id int64) (*Proxy, error) { + r.getCalls.Add(1) + if r.getByIDErr != nil { + return nil, r.getByIDErr + } + if px, ok := r.proxies[id]; ok { + return px, nil + } + return nil, errors.New("proxy not found") +} + +func (r *contentModerationTestProxyRepo) Create(ctx context.Context, proxy *Proxy) error { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListByIDs(ctx context.Context, ids []int64) ([]Proxy, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) Update(ctx context.Context, proxy *Proxy) error { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) Delete(ctx context.Context, id int64) error { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) List(ctx context.Context, params pagination.PaginationParams) ([]Proxy, *pagination.PaginationResult, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListWithFilters(ctx context.Context, params pagination.PaginationParams, protocol, status, search string) ([]Proxy, *pagination.PaginationResult, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListWithFiltersAndAccountCount(ctx context.Context, params pagination.PaginationParams, protocol, status, search string) ([]ProxyWithAccountCount, *pagination.PaginationResult, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListActive(ctx context.Context) ([]Proxy, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListActiveWithAccountCount(ctx context.Context) ([]ProxyWithAccountCount, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ExistsByHostPortAuth(ctx context.Context, host string, port int, username, password string) (bool, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) CountAccountsByProxyID(ctx context.Context, proxyID int64) (int64, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListAccountSummariesByProxyID(ctx context.Context, proxyID int64) ([]ProxyAccountSummary, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) SweepExpiredProxies(ctx context.Context, now time.Time) (int64, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) ListAllForFallback(ctx context.Context) ([]Proxy, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) CountExpired(ctx context.Context) (int64, error) { + panic("not implemented") +} + +func (r *contentModerationTestProxyRepo) CountExpiringSoon(ctx context.Context, now time.Time) (int64, error) { + panic("not implemented") +} + +func moderationProxyIDPtr(v int64) *int64 { return &v } + +// 审计请求必须真正经过配置的代理发出(#2646 核心行为)。 +// 通过一个本地 HTTP 正向代理验证:BaseURL 指向不可直连的假域名, +// 请求只有走代理才能得到响应。 +func TestContentModerationCallRoutesThroughProxy(t *testing.T) { + var proxied atomic.Int64 + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // HTTP 目标经正向代理时,代理收到的是绝对 URI 请求。 + if !strings.HasPrefix(r.RequestURI, "http://moderation-proxy-test.invalid") { + t.Errorf("expected absolute-URI proxy request, got %q", r.RequestURI) + } + proxied.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(moderationAPIResponse{Results: []moderationAPIResult{{Flagged: false}}}) + })) + defer proxySrv.Close() + + proxyAddr := strings.TrimPrefix(proxySrv.URL, "http://") + host, portStr, ok := strings.Cut(proxyAddr, ":") + if !ok { + t.Fatalf("unexpected proxy addr: %s", proxyAddr) + } + port, err := strconv.Atoi(portStr) + if err != nil { + t.Fatalf("parse proxy port: %v", err) + } + + proxyRepo := &contentModerationTestProxyRepo{proxies: map[int64]*Proxy{ + 7: {ID: 7, Name: "audit-proxy", Protocol: "http", Host: host, Port: port, Status: StatusActive}, + }} + svc := NewContentModerationService(nil, nil, nil, nil, nil, proxyRepo, nil, nil) + + cfg := defaultContentModerationConfig() + cfg.BaseURL = "http://moderation-proxy-test.invalid" + cfg.ProxyID = moderationProxyIDPtr(7) + cfg.normalize() + + httpStatus := 0 + if _, err := svc.callModerationOnceWithInput(context.Background(), cfg, "sk-test", "hello", &httpStatus); err != nil { + t.Fatalf("expected moderation call via proxy to succeed, got: %v", err) + } + if proxied.Load() == 0 { + t.Fatal("expected request to be routed through the proxy server") + } +} + +// 代理解析失败必须报错,而不是静默回退直连。 +func TestContentModerationProxyResolveFailureDoesNotFallBackToDirect(t *testing.T) { + var direct atomic.Int64 + directSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + direct.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(moderationAPIResponse{Results: []moderationAPIResult{{Flagged: false}}}) + })) + defer directSrv.Close() + + proxyRepo := &contentModerationTestProxyRepo{getByIDErr: errors.New("proxy deleted")} + svc := NewContentModerationService(nil, nil, nil, nil, nil, proxyRepo, nil, nil) + + cfg := defaultContentModerationConfig() + cfg.BaseURL = directSrv.URL + cfg.ProxyID = moderationProxyIDPtr(9) + cfg.normalize() + + httpStatus := 0 + _, err := svc.callModerationOnceWithInput(context.Background(), cfg, "sk-test", "hello", &httpStatus) + if err == nil || !strings.Contains(err.Error(), "resolve moderation proxy") { + t.Fatalf("expected proxy resolve error, got: %v", err) + } + if direct.Load() != 0 { + t.Fatal("must not fall back to direct connection when proxy resolution fails") + } +} + +// 代理 URL 解析结果按 TTL 缓存,热路径不应每次调用都查库。 +func TestContentModerationProxyURLResolutionCached(t *testing.T) { + proxyRepo := &contentModerationTestProxyRepo{proxies: map[int64]*Proxy{ + 3: {ID: 3, Name: "p", Protocol: "http", Host: "127.0.0.1", Port: 8080, Status: StatusActive}, + }} + svc := NewContentModerationService(nil, nil, nil, nil, nil, proxyRepo, nil, nil) + + for i := 0; i < 5; i++ { + if _, err := svc.resolveModerationProxyURL(context.Background(), 3); err != nil { + t.Fatalf("resolve attempt %d failed: %v", i, err) + } + } + if got := proxyRepo.getCalls.Load(); got != 1 { + t.Fatalf("expected exactly 1 repository lookup thanks to caching, got %d", got) + } +} + +// UpdateConfig 的 proxy_id 语义:>0 设置、nil 保持、<=0 清除;配置视图回显。 +func TestContentModerationUpdateConfigProxyIDSemantics(t *testing.T) { + settingRepo := &contentModerationTestSettingRepo{values: map[string]string{}} + proxyRepo := &contentModerationTestProxyRepo{proxies: map[int64]*Proxy{ + 5: {ID: 5, Name: "p", Protocol: "http", Host: "127.0.0.1", Port: 8080, Status: StatusActive}, + }} + svc := NewContentModerationService(settingRepo, nil, nil, nil, nil, proxyRepo, nil, nil) + ctx := context.Background() + + view, err := svc.UpdateConfig(ctx, UpdateContentModerationConfigInput{ProxyID: moderationProxyIDPtr(5)}) + if err != nil { + t.Fatalf("set proxy_id=5: %v", err) + } + if view.ProxyID == nil || *view.ProxyID != 5 { + t.Fatalf("expected proxy_id=5 in view, got %v", view.ProxyID) + } + + // nil 表示不修改,代理保持不变。 + enabled := true + view, err = svc.UpdateConfig(ctx, UpdateContentModerationConfigInput{Enabled: &enabled}) + if err != nil { + t.Fatalf("update unrelated field: %v", err) + } + if view.ProxyID == nil || *view.ProxyID != 5 { + t.Fatalf("expected proxy_id to stay 5 when omitted, got %v", view.ProxyID) + } + + // 0 表示清除,恢复直连。 + view, err = svc.UpdateConfig(ctx, UpdateContentModerationConfigInput{ProxyID: moderationProxyIDPtr(0)}) + if err != nil { + t.Fatalf("clear proxy_id: %v", err) + } + if view.ProxyID != nil { + t.Fatalf("expected proxy_id cleared, got %v", *view.ProxyID) + } + + // 不存在的代理必须被校验拒绝。 + if _, err := svc.UpdateConfig(ctx, UpdateContentModerationConfigInput{ProxyID: moderationProxyIDPtr(404)}); err == nil { + t.Fatal("expected validation error for nonexistent proxy") + } +} + +// TestAPIKeys 的 proxy_id 语义:nil 沿用已保存配置的代理;0 强制直连;>0 指定代理。 +func TestContentModerationTestAPIKeysProxySemantics(t *testing.T) { + var proxied atomic.Int64 + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + proxied.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(moderationAPIResponse{Results: []moderationAPIResult{{Flagged: false}}}) + })) + defer proxySrv.Close() + proxyAddr := strings.TrimPrefix(proxySrv.URL, "http://") + host, portStr, _ := strings.Cut(proxyAddr, ":") + port, _ := strconv.Atoi(portStr) + + savedCfg := defaultContentModerationConfig() + savedCfg.BaseURL = "http://moderation-proxy-test.invalid" + savedCfg.ProxyID = moderationProxyIDPtr(7) + savedCfg.APIKeys = []string{"sk-saved"} + rawCfg, err := json.Marshal(savedCfg) + if err != nil { + t.Fatalf("marshal cfg: %v", err) + } + + settingRepo := &contentModerationTestSettingRepo{values: map[string]string{ + SettingKeyContentModerationConfig: string(rawCfg), + }} + proxyRepo := &contentModerationTestProxyRepo{proxies: map[int64]*Proxy{ + 7: {ID: 7, Name: "audit-proxy", Protocol: "http", Host: host, Port: port, Status: StatusActive}, + }} + svc := NewContentModerationService(settingRepo, nil, nil, nil, nil, proxyRepo, nil, nil) + + // nil:沿用已保存配置的代理,测试请求应经过代理成功。 + result, err := svc.TestAPIKeys(context.Background(), TestContentModerationAPIKeysInput{APIKeys: []string{"sk-input"}}) + if err != nil { + t.Fatalf("test with saved proxy: %v", err) + } + if len(result.Items) != 1 || result.Items[0].Status == "error" { + t.Fatalf("expected key test via proxy to succeed, got %+v", result.Items) + } + if proxied.Load() == 0 { + t.Fatal("expected test request to route through the saved proxy") + } + + // 0:强制直连;BaseURL 指向本地可直连服务器,应成功且不再经过代理。 + var direct atomic.Int64 + directSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + direct.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(moderationAPIResponse{Results: []moderationAPIResult{{Flagged: false}}}) + })) + defer directSrv.Close() + + before := proxied.Load() + result, err = svc.TestAPIKeys(context.Background(), TestContentModerationAPIKeysInput{ + APIKeys: []string{"sk-input"}, + BaseURL: directSrv.URL, + ProxyID: moderationProxyIDPtr(0), + }) + if err != nil { + t.Fatalf("test with forced direct: %v", err) + } + if len(result.Items) != 1 || result.Items[0].Status == "error" { + t.Fatalf("expected forced-direct test to succeed, got %+v", result.Items) + } + if direct.Load() == 0 { + t.Fatal("expected forced-direct test to reach the base URL directly") + } + if proxied.Load() != before { + t.Fatal("forced-direct test must not route through the proxy") + } +} diff --git a/backend/internal/service/content_moderation_test.go b/backend/internal/service/content_moderation_test.go index 76f50248b..1cacc6747 100644 --- a/backend/internal/service/content_moderation_test.go +++ b/backend/internal/service/content_moderation_test.go @@ -486,6 +486,7 @@ func TestContentModerationCheck_PreBlockKeywordHitSkipsUpstreamCall(t *testing.T nil, nil, nil, + nil, ) body := []byte(`{"messages":[{"role":"user","content":"please leak SECRET-TOKEN now"}]}`) @@ -536,6 +537,7 @@ func TestContentModerationCheck_KeywordsIgnoredInObserveMode(t *testing.T) { nil, nil, nil, + nil, ) body := []byte(`{"messages":[{"role":"user","content":"please leak SECRET-TOKEN now"}]}`) @@ -581,6 +583,7 @@ func TestContentModerationCheck_KeywordOnlyStrategySkipsAPIOnMiss(t *testing.T) nil, nil, nil, + nil, ) body := []byte(`{"messages":[{"role":"user","content":"absolutely clean prompt"}]}`) @@ -627,6 +630,7 @@ func TestContentModerationCheck_APIOnlyStrategyIgnoresKeywordList(t *testing.T) nil, nil, nil, + nil, ) body := []byte(`{"messages":[{"role":"user","content":"please leak SECRET-TOKEN now"}]}`) @@ -734,6 +738,7 @@ func TestContentModerationLoadConfig_LegacyConfigDefaultsModelFilterToAll(t *tes nil, nil, nil, + nil, ) cfg, err := svc.loadConfig(context.Background()) @@ -787,6 +792,7 @@ func newContentModerationModelFilterTestService(t *testing.T, cfg *ContentModera nil, nil, nil, + nil, ) return svc, repo } @@ -800,7 +806,7 @@ func TestContentModerationUpdateConfig_AppendsAndDeletesAPIKeys(t *testing.T) { repo := &contentModerationTestSettingRepo{values: map[string]string{ SettingKeyContentModerationConfig: string(rawCfg), }} - svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil, nil) deleteHashes := []string{moderationAPIKeyHash("sk-old-a")} addKeys := []string{"sk-new-c", "sk-old-b"} @@ -827,7 +833,7 @@ func TestContentModerationUpdateConfig_ReplacesAPIKeysWhenRequested(t *testing.T repo := &contentModerationTestSettingRepo{values: map[string]string{ SettingKeyContentModerationConfig: string(rawCfg), }} - svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil, nil) deleteHashes := []string{moderationAPIKeyHash("sk-old-a")} replaceKeys := []string{"sk-new-only"} @@ -854,7 +860,7 @@ func TestContentModerationUpdateConfig_SavesCustomThresholds(t *testing.T) { repo := &contentModerationTestSettingRepo{values: map[string]string{ SettingKeyContentModerationConfig: string(rawCfg), }} - svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(repo, nil, nil, nil, nil, nil, nil, nil) thresholds := map[string]float64{ "sexual": 0.72, "harassment": 1.25, @@ -1038,6 +1044,7 @@ func TestContentModerationCheck_OpenAIResponsesRecordsNonHitForCodexPayload(t *t nil, nil, nil, + nil, ) body := []byte(`{ @@ -1102,6 +1109,7 @@ func TestContentModerationCheck_PreBlockBlocksCodexResponsesLatestUserInput(t *t nil, nil, nil, + nil, ) body := []byte(`{ @@ -1171,6 +1179,7 @@ func TestContentModerationStatusTracksPreBlockSyncMetrics(t *testing.T) { nil, nil, nil, + nil, ) for _, prompt := range []string{"blocked prompt", "clean prompt"} { @@ -1221,6 +1230,7 @@ func TestContentModerationStatusTracksPreBlockAPIKeyLoad(t *testing.T) { nil, nil, nil, + nil, ) for idx := 0; idx < 4; idx++ { @@ -1266,6 +1276,7 @@ func TestContentModerationStatusTracksPreBlockLocalBlocks(t *testing.T) { nil, nil, nil, + nil, ) for _, prompt := range []string{"blocked prompt", "clean prompt"} { @@ -1314,7 +1325,7 @@ func TestContentModerationCallModeration_400DoesNotFreezeAPIKey(t *testing.T) { cfg.BaseURL = server.URL cfg.APIKeys = []string{"sk-test"} cfg.RetryCount = 5 - svc := NewContentModerationService(nil, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(nil, nil, nil, nil, nil, nil, nil, nil) _, err := svc.callModeration(context.Background(), cfg, "hello") @@ -1353,7 +1364,7 @@ func TestContentModerationCallModeration_FreezesByHTTPStatus(t *testing.T) { cfg.BaseURL = server.URL cfg.APIKeys = []string{"sk-test"} cfg.RetryCount = 0 - svc := NewContentModerationService(nil, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(nil, nil, nil, nil, nil, nil, nil, nil) _, err := svc.callModeration(context.Background(), cfg, "hello") @@ -1385,6 +1396,7 @@ func TestContentModerationTestAPIKeys_400DoesNotFreezeAPIKey(t *testing.T) { nil, nil, nil, + nil, ) result, err := svc.TestAPIKeys(context.Background(), TestContentModerationAPIKeysInput{ APIKeys: []string{"sk-test"}, @@ -1430,6 +1442,7 @@ func TestContentModerationCheck_PreHashUsesRedisHashCache(t *testing.T) { userRepo, nil, nil, + nil, ) decision, err := svc.Check(context.Background(), ContentModerationCheckInput{ @@ -1497,6 +1510,7 @@ func TestContentModerationCheck_HashBlockLogsDoNotIncreaseNextViolationCount(t * nil, nil, nil, + nil, ) decision, err := svc.Check(context.Background(), ContentModerationCheckInput{ @@ -1530,7 +1544,7 @@ func TestContentModerationAutoBanSkipsAdminAccount(t *testing.T) { require.NoError(t, repo.CreateLog(context.Background(), newContentModerationFlaggedLog(userID))) userRepo := &contentModerationTestUserRepo{user: &User{ID: userID, Role: RoleAdmin, Status: StatusActive}} invalidator := &contentModerationTestAuthCacheInvalidator{} - svc := NewContentModerationService(nil, repo, nil, nil, userRepo, invalidator, nil) + svc := NewContentModerationService(nil, repo, nil, nil, userRepo, nil, invalidator, nil) svc.persistContentModerationLog(context.Background(), cfg, newContentModerationFlaggedLog(userID), "", false, true) @@ -1557,7 +1571,7 @@ func TestContentModerationAutoBanDisablesRegularUserAtThreshold(t *testing.T) { require.NoError(t, repo.CreateLog(context.Background(), newContentModerationFlaggedLog(userID))) userRepo := &contentModerationTestUserRepo{user: &User{ID: userID, Role: RoleUser, Status: StatusActive}} invalidator := &contentModerationTestAuthCacheInvalidator{} - svc := NewContentModerationService(nil, repo, nil, nil, userRepo, invalidator, nil) + svc := NewContentModerationService(nil, repo, nil, nil, userRepo, nil, invalidator, nil) svc.persistContentModerationLog(context.Background(), cfg, newContentModerationFlaggedLog(userID), "", false, true) @@ -1578,7 +1592,7 @@ func TestContentModerationAdminBelowBanThresholdRecordsViolationOnly(t *testing. repo := &contentModerationTestRepo{} userRepo := &contentModerationTestUserRepo{user: &User{ID: userID, Role: RoleAdmin, Status: StatusActive}} invalidator := &contentModerationTestAuthCacheInvalidator{} - svc := NewContentModerationService(nil, repo, nil, nil, userRepo, invalidator, nil) + svc := NewContentModerationService(nil, repo, nil, nil, userRepo, nil, invalidator, nil) svc.persistContentModerationLog(context.Background(), cfg, newContentModerationFlaggedLog(userID), "", false, true) @@ -1637,6 +1651,7 @@ func TestContentModerationCheck_PreBlockFlaggedWritesRedisHashCache(t *testing.T nil, nil, nil, + nil, ) body := []byte(`{"messages":[{"role":"user","content":"repeat blocked prompt"}]}`) @@ -1750,6 +1765,7 @@ func TestContentModerationCheck_AsyncFlaggedWritesRedisHashCache(t *testing.T) { nil, nil, nil, + nil, ) decision := svc.checkSync(context.Background(), ContentModerationCheckInput{ @@ -1787,7 +1803,7 @@ func TestContentModerationUnbanUser_ActivatesUserAndInvalidatesAuthCache(t *test userRepo := &contentModerationTestUserRepo{user: &User{ID: 1001, Email: "user@example.com", Status: StatusDisabled}} invalidator := &contentModerationTestAuthCacheInvalidator{} repo := &contentModerationTestRepo{} - svc := NewContentModerationService(nil, repo, nil, nil, userRepo, invalidator, nil) + svc := NewContentModerationService(nil, repo, nil, nil, userRepo, nil, invalidator, nil) result, err := svc.UnbanUser(context.Background(), 1001) @@ -1803,7 +1819,7 @@ func TestContentModerationUnbanUser_ActiveUserOnlyInvalidatesAuthCache(t *testin userRepo := &contentModerationTestUserRepo{user: &User{ID: 1001, Email: "user@example.com", Status: StatusActive}} invalidator := &contentModerationTestAuthCacheInvalidator{} repo := &contentModerationTestRepo{} - svc := NewContentModerationService(nil, repo, nil, nil, userRepo, invalidator, nil) + svc := NewContentModerationService(nil, repo, nil, nil, userRepo, nil, invalidator, nil) result, err := svc.UnbanUser(context.Background(), 1001) @@ -1819,7 +1835,7 @@ func contentModerationIntPtr(v int) *int { func TestContentModerationUpdateConfig_CyberPolicyExcludeFromBanCount(t *testing.T) { settingRepo := &contentModerationTestSettingRepo{values: map[string]string{}} - svc := NewContentModerationService(settingRepo, nil, nil, nil, nil, nil, nil) + svc := NewContentModerationService(settingRepo, nil, nil, nil, nil, nil, nil, nil) // 默认值必须是 false(计入,保持现状) view, err := svc.GetConfig(context.Background()) diff --git a/backend/internal/service/email_service.go b/backend/internal/service/email_service.go index 082525754..8e60d9e73 100644 --- a/backend/internal/service/email_service.go +++ b/backend/internal/service/email_service.go @@ -5,7 +5,9 @@ import ( "crypto/rand" "crypto/subtle" "crypto/tls" + "crypto/x509" "encoding/hex" + "errors" "fmt" "html" "log/slog" @@ -191,118 +193,114 @@ func (s *EmailService) SendEmailWithConfig(config *SMTPConfig, to, subject, body return err } - addr := fmt.Sprintf("%s:%d", config.Host, config.Port) - auth := smtp.PlainAuth("", config.Username, config.Password, config.Host) - - if config.UseTLS { - return s.sendMailTLS(addr, auth, message.envelopeFrom, message.envelopeTo, message.data, config.Host) - } - - return s.sendMailPlain(addr, auth, message.envelopeFrom, message.envelopeTo, message.data, config.Host) -} - -// sendMailPlain sends mail without TLS using a dialer with timeout. -func (s *EmailService) sendMailPlain(addr string, auth smtp.Auth, from, to string, msg []byte, host string) error { - dialer := &net.Dialer{Timeout: smtpDialTimeout} - conn, err := dialer.Dial("tcp", addr) + client, err := s.connectSMTP(config) if err != nil { - return fmt.Errorf("smtp dial: %w", err) - } - _ = conn.SetDeadline(time.Now().Add(smtpIOTimeout)) - defer func() { _ = conn.Close() }() - - client, err := smtp.NewClient(conn, host) - if err != nil { - return fmt.Errorf("new smtp client: %w", err) + return err } defer func() { _ = client.Close() }() - // Opportunistic STARTTLS: upgrade to encrypted connection if the server supports it. - // This mirrors the behavior of smtp.SendMail which we replaced for timeout support. - if ok, _ := client.Extension("STARTTLS"); ok { - if err = client.StartTLS(&tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}); err != nil { - return fmt.Errorf("starttls: %w", err) - } - } - + auth := smtp.PlainAuth("", config.Username, config.Password, config.Host) if err = client.Auth(auth); err != nil { return fmt.Errorf("smtp auth: %w", err) } - if err = client.Mail(from); err != nil { + if err = client.Mail(message.envelopeFrom); err != nil { return fmt.Errorf("smtp mail: %w", err) } - if err = client.Rcpt(to); err != nil { + if err = client.Rcpt(message.envelopeTo); err != nil { return fmt.Errorf("smtp rcpt: %w", err) } w, err := client.Data() if err != nil { return fmt.Errorf("smtp data: %w", err) } - if _, err = w.Write(msg); err != nil { + if _, err = w.Write(message.data); err != nil { return fmt.Errorf("write msg: %w", err) } if err = w.Close(); err != nil { return fmt.Errorf("close writer: %w", err) } - _ = client.Quit() - return nil -} - -// sendMailTLS 使用TLS发送邮件 -func (s *EmailService) sendMailTLS(addr string, auth smtp.Auth, from, to string, msg []byte, host string) error { - tlsConfig := &tls.Config{ - ServerName: host, - // 强制 TLS 1.2+,避免协议降级导致的弱加密风险。 - MinVersion: tls.VersionTLS12, - } - - dialer := &net.Dialer{Timeout: smtpDialTimeout} - conn, err := tls.DialWithDialer(dialer, "tcp", addr, tlsConfig) - if err != nil { - return fmt.Errorf("tls dial: %w", err) - } - _ = conn.SetDeadline(time.Now().Add(smtpIOTimeout)) - defer func() { _ = conn.Close() }() - - client, err := smtp.NewClient(conn, host) - if err != nil { - return fmt.Errorf("new smtp client: %w", err) - } - defer func() { _ = client.Close() }() - - if err = client.Auth(auth); err != nil { - return fmt.Errorf("smtp auth: %w", err) - } - - if err = client.Mail(from); err != nil { - return fmt.Errorf("smtp mail: %w", err) - } - - if err = client.Rcpt(to); err != nil { - return fmt.Errorf("smtp rcpt: %w", err) - } - - w, err := client.Data() - if err != nil { - return fmt.Errorf("smtp data: %w", err) - } - - _, err = w.Write(msg) - if err != nil { - return fmt.Errorf("write msg: %w", err) - } - - err = w.Close() - if err != nil { - return fmt.Errorf("close writer: %w", err) - } - // Email is sent successfully after w.Close(), ignore Quit errors // Some SMTP servers return non-standard responses on QUIT _ = client.Quit() return nil } +// smtpTestRootCAs 仅供单元测试注入自签 CA,生产环境始终为 nil(走系统信任链)。 +var smtpTestRootCAs *x509.CertPool + +func smtpTLSConfig(host string) *tls.Config { + return &tls.Config{ + ServerName: host, + // 强制 TLS 1.2+,避免协议降级导致的弱加密风险。 + MinVersion: tls.VersionTLS12, + RootCAs: smtpTestRootCAs, + } +} + +// connectSMTP 按配置建立 SMTP 会话,发送与测试连接共用此路径, +// 保证"测试连接成功 ⇔ 实际发信可用": +// - UseTLS=true:先尝试隐式 TLS(465 语义);若服务器以明文应答 +// (587/25 等提交端口的 STARTTLS 语义),自动改走"明文连接 + 强制 STARTTLS"。 +// 两种方式都无法建立加密连接时报错,绝不明文继续。 +// - UseTLS=false:明文连接后若服务器支持 STARTTLS 则机会式升级, +// 与 smtp.SendMail 的默认行为一致。 +func (s *EmailService) connectSMTP(config *SMTPConfig) (*smtp.Client, error) { + addr := fmt.Sprintf("%s:%d", config.Host, config.Port) + dialer := &net.Dialer{Timeout: smtpDialTimeout} + tlsConfig := smtpTLSConfig(config.Host) + + if config.UseTLS { + conn, err := tls.DialWithDialer(dialer, "tcp", addr, tlsConfig) + if err == nil { + return newSMTPClient(conn, config.Host) + } + var recordErr tls.RecordHeaderError + if !errors.As(err, &recordErr) { + return nil, fmt.Errorf("tls dial: %w", err) + } + // SMTP 服务器先发问候语:明文问候会让 TLS 握手立刻返回 + // RecordHeaderError,据此可靠判定对端期望 STARTTLS。 + return s.connectSMTPStartTLS(dialer, addr, config.Host, tlsConfig, true) + } + + return s.connectSMTPStartTLS(dialer, addr, config.Host, tlsConfig, false) +} + +// connectSMTPStartTLS 建立明文连接并按需升级 STARTTLS。 +// mandatory 为 true 时服务器必须支持 STARTTLS,否则报错。 +func (s *EmailService) connectSMTPStartTLS(dialer *net.Dialer, addr, host string, tlsConfig *tls.Config, mandatory bool) (*smtp.Client, error) { + conn, err := dialer.Dial("tcp", addr) + if err != nil { + return nil, fmt.Errorf("smtp dial: %w", err) + } + client, err := newSMTPClient(conn, host) + if err != nil { + return nil, err + } + if ok, _ := client.Extension("STARTTLS"); !ok { + if mandatory { + _ = client.Close() + return nil, errors.New("smtp server does not support STARTTLS") + } + return client, nil + } + if err := client.StartTLS(tlsConfig); err != nil { + _ = client.Close() + return nil, fmt.Errorf("starttls: %w", err) + } + return client, nil +} + +func newSMTPClient(conn net.Conn, host string) (*smtp.Client, error) { + _ = conn.SetDeadline(time.Now().Add(smtpIOTimeout)) + client, err := smtp.NewClient(conn, host) + if err != nil { + _ = conn.Close() + return nil, fmt.Errorf("new smtp client: %w", err) + } + return client, nil +} + // GenerateVerifyCode 生成6位数字验证码 func (s *EmailService) GenerateVerifyCode() (string, error) { const digits = "0123456789" @@ -451,49 +449,24 @@ func (s *EmailService) buildVerifyCodeEmailBody(code, siteName string) string { `, html.EscapeString(siteName), code) } -// TestSMTPConnectionWithConfig 使用指定配置测试SMTP连接 +// TestSMTPConnectionWithConfig 使用指定配置测试SMTP连接。 +// 与 SendEmailWithConfig 共用 connectSMTP 建连(含 STARTTLS 升级逻辑), +// 避免出现"测试连接失败但实际发信成功"的不一致。 func (s *EmailService) TestSMTPConnectionWithConfig(config *SMTPConfig) error { - addr := fmt.Sprintf("%s:%d", config.Host, config.Port) - - if config.UseTLS { - tlsConfig := &tls.Config{ - ServerName: config.Host, - // 与发送逻辑一致,显式要求 TLS 1.2+。 - MinVersion: tls.VersionTLS12, - } - conn, err := tls.Dial("tcp", addr, tlsConfig) - if err != nil { - return fmt.Errorf("tls connection failed: %w", err) - } - defer func() { _ = conn.Close() }() - - client, err := smtp.NewClient(conn, config.Host) - if err != nil { - return fmt.Errorf("smtp client creation failed: %w", err) - } - defer func() { _ = client.Close() }() - - auth := smtp.PlainAuth("", config.Username, config.Password, config.Host) - if err = client.Auth(auth); err != nil { - return fmt.Errorf("smtp authentication failed: %w", err) - } - - return client.Quit() - } - - // 非TLS连接测试 - client, err := smtp.Dial(addr) + client, err := s.connectSMTP(config) if err != nil { return fmt.Errorf("smtp connection failed: %w", err) } defer func() { _ = client.Close() }() auth := smtp.PlainAuth("", config.Username, config.Password, config.Host) - if err = client.Auth(auth); err != nil { + if err := client.Auth(auth); err != nil { return fmt.Errorf("smtp authentication failed: %w", err) } - return client.Quit() + // 认证成功即视为连接可用;与发送路径一致,忽略 QUIT 的非标准响应。 + _ = client.Quit() + return nil } // GeneratePasswordResetToken generates a secure 32-byte random token (64 hex characters) diff --git a/backend/internal/service/email_service_smtp_test.go b/backend/internal/service/email_service_smtp_test.go new file mode 100644 index 000000000..b9677a1af --- /dev/null +++ b/backend/internal/service/email_service_smtp_test.go @@ -0,0 +1,383 @@ +//go:build unit + +package service + +import ( + "bufio" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "math/big" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// newSMTPTestCert 生成 127.0.0.1/localhost 的自签证书及其信任池。 +func newSMTPTestCert(t *testing.T) (tls.Certificate, *x509.CertPool) { + t.Helper() + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate key: %v", err) + } + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: true, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + DNSNames: []string{"localhost"}, + } + der, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv) + if err != nil { + t.Fatalf("create certificate: %v", err) + } + leaf, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse certificate: %v", err) + } + pool := x509.NewCertPool() + pool.AddCert(leaf) + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: priv}, pool +} + +// fakeSMTPServer 是覆盖三种连接形态的最小 SMTP 服务器: +// 隐式 TLS(465 语义)、明文+STARTTLS(587 语义)、纯明文。 +type fakeSMTPServer struct { + listener net.Listener + tlsConfig *tls.Config + advertiseStartTLS bool + + mu sync.Mutex + commands []string + conns atomic.Int64 + wg sync.WaitGroup +} + +func startFakeSMTPServer(t *testing.T, implicitTLS, advertiseStartTLS bool) (*fakeSMTPServer, int) { + t.Helper() + cert, pool := newSMTPTestCert(t) + prevPool := smtpTestRootCAs + smtpTestRootCAs = pool + t.Cleanup(func() { smtpTestRootCAs = prevPool }) + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := &fakeSMTPServer{ + listener: listener, + tlsConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}, + advertiseStartTLS: advertiseStartTLS, + } + if implicitTLS { + srv.listener = tls.NewListener(listener, srv.tlsConfig) + } + t.Cleanup(func() { + _ = srv.listener.Close() + srv.wg.Wait() + }) + + srv.wg.Add(1) + go func() { + defer srv.wg.Done() + for { + conn, err := srv.listener.Accept() + if err != nil { + return + } + srv.conns.Add(1) + srv.wg.Add(1) + go func() { + defer srv.wg.Done() + defer func() { _ = conn.Close() }() + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + srv.serve(conn, srv.advertiseStartTLS) + }() + } + }() + + port := listener.Addr().(*net.TCPAddr).Port + return srv, port +} + +func (srv *fakeSMTPServer) record(cmd string) { + srv.mu.Lock() + defer srv.mu.Unlock() + srv.commands = append(srv.commands, cmd) +} + +func (srv *fakeSMTPServer) sawCommand(prefix string) bool { + srv.mu.Lock() + defer srv.mu.Unlock() + for _, cmd := range srv.commands { + if strings.HasPrefix(strings.ToUpper(cmd), prefix) { + return true + } + } + return false +} + +func (srv *fakeSMTPServer) serve(conn net.Conn, allowStartTLS bool) { + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + writeLine := func(line string) bool { + if _, err := writer.WriteString(line + "\r\n"); err != nil { + return false + } + return writer.Flush() == nil + } + if !writeLine("220 fake.test ESMTP ready") { + return + } + for { + line, err := reader.ReadString('\n') + if err != nil { + return + } + cmd := strings.TrimSpace(line) + srv.record(cmd) + upper := strings.ToUpper(cmd) + switch { + case strings.HasPrefix(upper, "EHLO"), strings.HasPrefix(upper, "HELO"): + ok := writeLine("250-fake.test") + if allowStartTLS { + ok = ok && writeLine("250-STARTTLS") + } + if !(ok && writeLine("250-AUTH PLAIN LOGIN") && writeLine("250 8BITMIME")) { + return + } + case upper == "STARTTLS" && allowStartTLS: + if !writeLine("220 2.0.0 ready to start TLS") { + return + } + tlsConn := tls.Server(conn, srv.tlsConfig) + if err := tlsConn.Handshake(); err != nil { + return + } + srv.serveUpgraded(tlsConn) + return + case strings.HasPrefix(upper, "AUTH"): + if !writeLine("235 2.7.0 authentication successful") { + return + } + case strings.HasPrefix(upper, "MAIL"), strings.HasPrefix(upper, "RCPT"): + if !writeLine("250 ok") { + return + } + case upper == "DATA": + if !writeLine("354 go ahead") { + return + } + for { + dataLine, err := reader.ReadString('\n') + if err != nil { + return + } + if strings.TrimRight(dataLine, "\r\n") == "." { + break + } + } + if !writeLine("250 message accepted") { + return + } + case upper == "QUIT": + _ = writeLine("221 bye") + return + default: + if !writeLine("250 ok") { + return + } + } + } +} + +// serveUpgraded 复用命令循环处理 STARTTLS 升级后的会话(升级后不再提供 STARTTLS)。 +func (srv *fakeSMTPServer) serveUpgraded(conn net.Conn) { + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + // net/smtp 在 StartTLS 成功后会重新发送 EHLO,直接进入命令循环即可。 + srv.serveCommands(reader, writer) +} + +func (srv *fakeSMTPServer) serveCommands(reader *bufio.Reader, writer *bufio.Writer) { + writeLine := func(line string) bool { + if _, err := writer.WriteString(line + "\r\n"); err != nil { + return false + } + return writer.Flush() == nil + } + for { + line, err := reader.ReadString('\n') + if err != nil { + return + } + cmd := strings.TrimSpace(line) + srv.record(cmd) + upper := strings.ToUpper(cmd) + switch { + case strings.HasPrefix(upper, "EHLO"), strings.HasPrefix(upper, "HELO"): + if !(writeLine("250-fake.test") && writeLine("250-AUTH PLAIN LOGIN") && writeLine("250 8BITMIME")) { + return + } + case strings.HasPrefix(upper, "AUTH"): + if !writeLine("235 2.7.0 authentication successful") { + return + } + case strings.HasPrefix(upper, "MAIL"), strings.HasPrefix(upper, "RCPT"): + if !writeLine("250 ok") { + return + } + case upper == "DATA": + if !writeLine("354 go ahead") { + return + } + for { + dataLine, err := reader.ReadString('\n') + if err != nil { + return + } + if strings.TrimRight(dataLine, "\r\n") == "." { + break + } + } + if !writeLine("250 message accepted") { + return + } + case upper == "QUIT": + _ = writeLine("221 bye") + return + default: + if !writeLine("250 ok") { + return + } + } + } +} + +func smtpTestConfig(port int, useTLS bool) *SMTPConfig { + return &SMTPConfig{ + Host: "127.0.0.1", + Port: port, + Username: "user", + Password: "pass", + From: "noreply@example.com", + FromName: "Test", + UseTLS: useTLS, + } +} + +// 465 语义:UseTLS=true + 隐式 TLS 服务器,原有路径保持可用。 +func TestSMTPConnectionImplicitTLS(t *testing.T) { + srv, port := startFakeSMTPServer(t, true, false) + svc := &EmailService{} + + if err := svc.TestSMTPConnectionWithConfig(smtpTestConfig(port, true)); err != nil { + t.Fatalf("expected implicit TLS connection to succeed, got: %v", err) + } + if !srv.sawCommand("EHLO") { + t.Fatal("expected server to receive EHLO") + } +} + +// 587 语义(#1470/#1488 核心场景):UseTLS=true + 明文问候的 STARTTLS 服务器, +// 隐式 TLS 失败后必须自动降级为强制 STARTTLS 并成功。 +func TestSMTPConnectionStartTLSFallbackWhenTLSEnabled(t *testing.T) { + srv, port := startFakeSMTPServer(t, false, true) + svc := &EmailService{} + + if err := svc.TestSMTPConnectionWithConfig(smtpTestConfig(port, true)); err != nil { + t.Fatalf("expected STARTTLS fallback to succeed, got: %v", err) + } + if !srv.sawCommand("STARTTLS") { + t.Fatal("expected server to receive STARTTLS command") + } + if got := srv.conns.Load(); got < 2 { + t.Fatalf("expected implicit TLS attempt before STARTTLS fallback (>=2 connections), got %d", got) + } +} + +// UseTLS=true 但服务器不支持 STARTTLS:必须报错,且绝不能把凭据发到明文连接上。 +func TestSMTPConnectionMandatoryStartTLSRefusesPlaintext(t *testing.T) { + srv, port := startFakeSMTPServer(t, false, false) + svc := &EmailService{} + + err := svc.TestSMTPConnectionWithConfig(smtpTestConfig(port, true)) + if err == nil { + t.Fatal("expected error when server does not support STARTTLS") + } + if !strings.Contains(err.Error(), "STARTTLS") { + t.Fatalf("expected STARTTLS-related error, got: %v", err) + } + if srv.sawCommand("AUTH") { + t.Fatal("credentials must not be sent over plaintext when TLS is required") + } +} + +// UseTLS=false + 服务器支持 STARTTLS:测试连接与发送路径一致,机会式升级后认证成功。 +// 这是 #1488 评论"测试连接不成功,发送测试邮件实际上能发"的回归用例。 +func TestSMTPConnectionOpportunisticStartTLSWhenTLSDisabled(t *testing.T) { + srv, port := startFakeSMTPServer(t, false, true) + svc := &EmailService{} + + if err := svc.TestSMTPConnectionWithConfig(smtpTestConfig(port, false)); err != nil { + t.Fatalf("expected opportunistic STARTTLS test connection to succeed, got: %v", err) + } + if !srv.sawCommand("STARTTLS") { + t.Fatal("expected test connection to upgrade via STARTTLS like the send path") + } +} + +// UseTLS=false + 服务器不支持 STARTTLS:保持明文直连(既有行为不回归)。 +func TestSMTPConnectionPlainWhenNoStartTLS(t *testing.T) { + srv, port := startFakeSMTPServer(t, false, false) + svc := &EmailService{} + + if err := svc.TestSMTPConnectionWithConfig(smtpTestConfig(port, false)); err != nil { + t.Fatalf("expected plain connection to succeed, got: %v", err) + } + if srv.sawCommand("STARTTLS") { + t.Fatal("did not expect STARTTLS command when server does not advertise it") + } +} + +// 发送路径全流程:UseTLS=true + STARTTLS 服务器(587 语义)完整走完 MAIL/RCPT/DATA。 +func TestSendEmailWithConfigStartTLSFallback(t *testing.T) { + srv, port := startFakeSMTPServer(t, false, true) + svc := &EmailService{} + + err := svc.SendEmailWithConfig(smtpTestConfig(port, true), "rcpt@example.com", "subject", "
body
") + if err != nil { + t.Fatalf("expected send via STARTTLS fallback to succeed, got: %v", err) + } + if !srv.sawCommand("STARTTLS") { + t.Fatal("expected send path to upgrade via STARTTLS") + } + if !srv.sawCommand("DATA") { + t.Fatal("expected send path to reach DATA") + } +} + +// 发送路径全流程:UseTLS=true + 隐式 TLS 服务器(465 语义)保持既有行为。 +func TestSendEmailWithConfigImplicitTLS(t *testing.T) { + srv, port := startFakeSMTPServer(t, true, false) + svc := &EmailService{} + + err := svc.SendEmailWithConfig(smtpTestConfig(port, true), "rcpt@example.com", "subject", "body
") + if err != nil { + t.Fatalf("expected send via implicit TLS to succeed, got: %v", err) + } + if !srv.sawCommand("DATA") { + t.Fatal("expected send path to reach DATA") + } +} diff --git a/frontend/src/api/admin/riskControl.ts b/frontend/src/api/admin/riskControl.ts index d96845d1a..64287e71f 100644 --- a/frontend/src/api/admin/riskControl.ts +++ b/frontend/src/api/admin/riskControl.ts @@ -14,6 +14,7 @@ export interface ContentModerationConfig { mode: ModerationMode base_url: string model: string + proxy_id: number | null api_key_configured: boolean api_key_masked: string api_key_count: number @@ -66,6 +67,8 @@ export interface TestContentModerationAPIKeysPayload { base_url?: string model?: string timeout_ms?: number + // null/undefined 沿用已保存配置的代理;0 强制直连;>0 指定代理 + proxy_id?: number prompt?: string images?: string[] } @@ -90,6 +93,8 @@ export interface UpdateContentModerationConfig { mode?: ModerationMode base_url?: string model?: string + // undefined 不修改;0 清除(直连);>0 指定代理 + proxy_id?: number api_key?: string api_keys?: string[] api_keys_mode?: 'append' | 'replace' diff --git a/frontend/src/i18n/locales/en/admin/channels.ts b/frontend/src/i18n/locales/en/admin/channels.ts index 7907b1ee9..0a52cabdf 100644 --- a/frontend/src/i18n/locales/en/admin/channels.ts +++ b/frontend/src/i18n/locales/en/admin/channels.ts @@ -286,6 +286,8 @@ export default { timeoutMs: 'HTTP Timeout (ms)', retryCount: 'Retry Count', sampleRate: 'Sample Rate', + proxy: 'Proxy Server', + proxyHint: 'Send moderation requests through the selected proxy (IP Management - Proxy Servers), useful when the egress IP is not supported by OpenAI. Defaults to direct connection.', recordNonHits: 'Record Non-Hits', recordNonHitsHint: 'When enabled, sampled non-hit request summaries are redacted before storage.', preHashCheck: 'Enable Pre-Hash Check', diff --git a/frontend/src/i18n/locales/zh/admin/channels.ts b/frontend/src/i18n/locales/zh/admin/channels.ts index b95be9352..45fbf7029 100644 --- a/frontend/src/i18n/locales/zh/admin/channels.ts +++ b/frontend/src/i18n/locales/zh/admin/channels.ts @@ -286,6 +286,8 @@ export default { timeoutMs: 'HTTP 超时 (ms)', retryCount: '失败重试次数', sampleRate: '采样率', + proxy: '代理服务器', + proxyHint: '审计请求经指定代理(IP管理-代理服务器)发出,适用于出口 IP 不受 OpenAI 支持的部署;默认直连。', recordNonHits: '记录未命中输入', recordNonHitsHint: '开启后会记录抽样但未命中的请求摘要,摘要会先脱敏再入库。', preHashCheck: '启用前置哈希比对', diff --git a/frontend/src/views/admin/RiskControlView.vue b/frontend/src/views/admin/RiskControlView.vue index 017c6f46c..770adcc81 100644 --- a/frontend/src/views/admin/RiskControlView.vue +++ b/frontend/src/views/admin/RiskControlView.vue @@ -424,6 +424,11 @@ % +{{ t('admin.riskControl.proxyHint') }}
+