mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-07 14:08:14 +08:00
Merge pull request #5147 from Wei-Shaw/feat/moderation-proxy-and-smtp-starttls
feat(moderation): proxy support for content audit; fix(email): SMTP STARTTLS test/send parity
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
|
||||
@@ -1266,6 +1266,7 @@ func TestOpenAIResponsesWebSocket_ContentModerationBlocksFirstFrame(t *testing.T
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
decision, err := moderationSvc.Check(context.Background(), service.ContentModerationCheckInput{
|
||||
UserID: 1,
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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())
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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", "<p>body</p>")
|
||||
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", "<p>body</p>")
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -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'
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -286,6 +286,8 @@ export default {
|
||||
timeoutMs: 'HTTP 超时 (ms)',
|
||||
retryCount: '失败重试次数',
|
||||
sampleRate: '采样率',
|
||||
proxy: '代理服务器',
|
||||
proxyHint: '审计请求经指定代理(IP管理-代理服务器)发出,适用于出口 IP 不受 OpenAI 支持的部署;默认直连。',
|
||||
recordNonHits: '记录未命中输入',
|
||||
recordNonHitsHint: '开启后会记录抽样但未命中的请求摘要,摘要会先脱敏再入库。',
|
||||
preHashCheck: '启用前置哈希比对',
|
||||
|
||||
@@ -424,6 +424,11 @@
|
||||
<span class="pointer-events-none absolute right-3 top-1/2 -translate-y-1/2 text-gray-400">%</span>
|
||||
</div>
|
||||
</div>
|
||||
<div>
|
||||
<label class="input-label">{{ t('admin.riskControl.proxy') }}</label>
|
||||
<ProxySelector v-model="configForm.proxy_id" :proxies="proxies" />
|
||||
<p class="mt-2 text-xs leading-5 text-gray-500 dark:text-gray-400">{{ t('admin.riskControl.proxyHint') }}</p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="overflow-hidden rounded-xl border border-gray-100 bg-white shadow-sm dark:border-dark-700 dark:bg-dark-800">
|
||||
@@ -1123,6 +1128,7 @@ import Select from '@/components/common/Select.vue'
|
||||
import Toggle from '@/components/common/Toggle.vue'
|
||||
import Pagination from '@/components/common/Pagination.vue'
|
||||
import ModelWhitelistSelector from '@/components/account/ModelWhitelistSelector.vue'
|
||||
import ProxySelector from '@/components/common/ProxySelector.vue'
|
||||
import { adminAPI } from '@/api/admin'
|
||||
import type {
|
||||
ContentModerationAPIKeyLoad,
|
||||
@@ -1137,7 +1143,7 @@ import type {
|
||||
ModerationMode,
|
||||
UpdateContentModerationConfig,
|
||||
} from '@/api/admin/riskControl'
|
||||
import type { AdminGroup, SelectOption } from '@/types'
|
||||
import type { AdminGroup, Proxy, SelectOption } from '@/types'
|
||||
import { useAppStore } from '@/stores/app'
|
||||
import { extractApiErrorMessage } from '@/utils/apiError'
|
||||
import { formatDateTime as formatDateTimeValue } from '@/utils/format'
|
||||
@@ -1205,6 +1211,7 @@ const activeSettingsTab = ref<SettingsTab>('basic')
|
||||
const groupSearch = ref('')
|
||||
const flaggedHashInput = ref('')
|
||||
const groups = ref<AdminGroup[]>([])
|
||||
const proxies = ref<Proxy[]>([])
|
||||
const logs = ref<ContentModerationLog[]>([])
|
||||
const status = ref<ContentModerationRuntimeStatus | null>(null)
|
||||
const testedApiKeyStatuses = ref<ContentModerationAPIKeyStatus[]>([])
|
||||
@@ -1221,6 +1228,7 @@ const configForm = reactive({
|
||||
mode: 'pre_block' as ModerationMode,
|
||||
base_url: 'https://api.openai.com',
|
||||
model: 'omni-moderation-latest',
|
||||
proxy_id: null as number | null,
|
||||
api_keys_text: '',
|
||||
api_key_configured: false,
|
||||
api_key_masked: '',
|
||||
@@ -1695,6 +1703,7 @@ function applyConfig(config: ContentModerationConfig) {
|
||||
configForm.mode = config.mode
|
||||
configForm.base_url = config.base_url || 'https://api.openai.com'
|
||||
configForm.model = config.model || 'omni-moderation-latest'
|
||||
configForm.proxy_id = config.proxy_id || null
|
||||
configForm.api_keys_text = ''
|
||||
configForm.api_key_configured = config.api_key_configured
|
||||
configForm.api_key_masked = config.api_key_masked || ''
|
||||
@@ -1735,14 +1744,17 @@ function applyConfig(config: ContentModerationConfig) {
|
||||
async function loadAll() {
|
||||
loading.value = true
|
||||
try {
|
||||
const [config, groupItems, runtimeStatus] = await Promise.all([
|
||||
const [config, groupItems, runtimeStatus, proxyItems] = await Promise.all([
|
||||
adminAPI.riskControl.getConfig(),
|
||||
adminAPI.groups.getAll(),
|
||||
adminAPI.riskControl.getStatus(),
|
||||
// 代理列表加载失败不阻塞风控页面(仅影响下拉可选项)
|
||||
adminAPI.proxies.getAll().catch(() => [] as Proxy[]),
|
||||
])
|
||||
applyConfig(config)
|
||||
groups.value = groupItems
|
||||
status.value = runtimeStatus
|
||||
proxies.value = proxyItems
|
||||
if (Array.isArray(runtimeStatus.api_key_statuses)) {
|
||||
configForm.api_key_statuses = [...runtimeStatus.api_key_statuses]
|
||||
prunePendingDeleteAPIKeyHashes()
|
||||
@@ -1786,6 +1798,8 @@ async function saveConfig() {
|
||||
mode: configForm.mode,
|
||||
base_url: configForm.base_url,
|
||||
model: configForm.model,
|
||||
// 后端语义:0 清除代理(直连),>0 指定代理
|
||||
proxy_id: configForm.proxy_id ?? 0,
|
||||
timeout_ms: Number(configForm.timeout_ms) || 3000,
|
||||
retry_count: Number(configForm.retry_count) || 0,
|
||||
sample_rate: Number(configForm.sample_rate) || 0,
|
||||
@@ -1984,6 +1998,8 @@ async function testApiKeys(useInputKeys: boolean) {
|
||||
base_url: configForm.base_url,
|
||||
model: configForm.model,
|
||||
timeout_ms: Number(configForm.timeout_ms) || 3000,
|
||||
// 与保存语义一致:0 强制直连,>0 指定代理,确保测试与实际审计走同一条链路
|
||||
proxy_id: configForm.proxy_id ?? 0,
|
||||
prompt: moderationTestPrompt.value,
|
||||
images: moderationTestImages.value,
|
||||
})
|
||||
|
||||
@@ -12,6 +12,7 @@ const {
|
||||
getStatus,
|
||||
listLogs,
|
||||
getGroups,
|
||||
getProxies,
|
||||
showError,
|
||||
showSuccess,
|
||||
} = vi.hoisted(() => ({
|
||||
@@ -20,6 +21,7 @@ const {
|
||||
getStatus: vi.fn(),
|
||||
listLogs: vi.fn(),
|
||||
getGroups: vi.fn(),
|
||||
getProxies: vi.fn(),
|
||||
showError: vi.fn(),
|
||||
showSuccess: vi.fn(),
|
||||
}))
|
||||
@@ -39,6 +41,9 @@ vi.mock('@/api/admin', () => ({
|
||||
groups: {
|
||||
getAll: getGroups,
|
||||
},
|
||||
proxies: {
|
||||
getAll: getProxies,
|
||||
},
|
||||
},
|
||||
}))
|
||||
|
||||
@@ -73,6 +78,7 @@ const baseConfig = (): ContentModerationConfig => ({
|
||||
mode: 'pre_block',
|
||||
base_url: 'https://api.openai.com',
|
||||
model: 'omni-moderation-latest',
|
||||
proxy_id: null,
|
||||
api_key_configured: false,
|
||||
api_key_masked: '',
|
||||
api_key_count: 0,
|
||||
@@ -198,6 +204,7 @@ describe('admin RiskControlView', () => {
|
||||
getStatus.mockResolvedValue(runtimeStatus())
|
||||
listLogs.mockResolvedValue({ items: [], total: 0, page: 1, page_size: 20, pages: 1 })
|
||||
getGroups.mockResolvedValue([])
|
||||
getProxies.mockResolvedValue([])
|
||||
updateConfig.mockImplementation(async (payload: UpdateContentModerationConfig) => ({
|
||||
...baseConfig(),
|
||||
...payload,
|
||||
@@ -221,6 +228,7 @@ describe('admin RiskControlView', () => {
|
||||
Toggle: true,
|
||||
Pagination: true,
|
||||
ModelWhitelistSelector: ModelWhitelistSelectorStub,
|
||||
ProxySelector: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -254,6 +262,7 @@ describe('admin RiskControlView', () => {
|
||||
Toggle: true,
|
||||
Pagination: true,
|
||||
ModelWhitelistSelector: ModelWhitelistSelectorStub,
|
||||
ProxySelector: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -294,6 +303,7 @@ describe('admin RiskControlView', () => {
|
||||
Toggle: true,
|
||||
Pagination: true,
|
||||
ModelWhitelistSelector: ModelWhitelistSelectorStub,
|
||||
ProxySelector: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -361,6 +371,7 @@ describe('admin RiskControlView', () => {
|
||||
Toggle: true,
|
||||
Pagination: true,
|
||||
ModelWhitelistSelector: ModelWhitelistSelectorStub,
|
||||
ProxySelector: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user