diff --git a/backend/internal/repository/email_cache.go b/backend/internal/repository/email_cache.go index 63552ab078..8868ddfc7e 100644 --- a/backend/internal/repository/email_cache.go +++ b/backend/internal/repository/email_cache.go @@ -3,6 +3,7 @@ package repository import ( "context" "encoding/json" + "fmt" "time" "github.com/Wei-Shaw/sub2api/internal/service" @@ -10,10 +11,11 @@ import ( ) const ( - verifyCodeKeyPrefix = "verify_code:" - notifyVerifyKeyPrefix = "notify_verify:" - passwordResetKeyPrefix = "password_reset:" - passwordResetSentAtKeyPrefix = "password_reset_sent:" + verifyCodeKeyPrefix = "verify_code:" + notifyVerifyKeyPrefix = "notify_verify:" + passwordResetKeyPrefix = "password_reset:" + passwordResetSentAtKeyPrefix = "password_reset_sent:" + notifyCodeUserRateKeyPrefix = "notify_code_user_rate:" ) // verifyCodeKey generates the Redis key for email verification code. @@ -141,3 +143,31 @@ func (c *emailCache) DeleteNotifyVerifyCode(ctx context.Context, email string) e key := notifyVerifyKey(email) return c.rdb.Del(ctx, key).Err() } + +// User-level rate limiting for notify email verification codes + +func notifyCodeUserRateKey(userID int64) string { + return notifyCodeUserRateKeyPrefix + fmt.Sprintf("%d", userID) +} + +func (c *emailCache) IncrNotifyCodeUserRate(ctx context.Context, userID int64, window time.Duration) (int64, error) { + key := notifyCodeUserRateKey(userID) + count, err := c.rdb.Incr(ctx, key).Result() + if err != nil { + return 0, err + } + // Set TTL only on first increment (when count becomes 1) + if count == 1 { + c.rdb.Expire(ctx, key, window) + } + return count, nil +} + +func (c *emailCache) GetNotifyCodeUserRate(ctx context.Context, userID int64) (int64, error) { + key := notifyCodeUserRateKey(userID) + count, err := c.rdb.Get(ctx, key).Int64() + if err != nil { + return 0, err + } + return count, nil +} diff --git a/backend/internal/service/account_stats_pricing.go b/backend/internal/service/account_stats_pricing.go index cbe9c76cc1..e47b743164 100644 --- a/backend/internal/service/account_stats_pricing.go +++ b/backend/internal/service/account_stats_pricing.go @@ -2,7 +2,6 @@ package service import ( "context" - "sort" "strings" ) @@ -104,14 +103,8 @@ func matchAccountStatsRule(rule *AccountStatsPricingRule, accountID, groupID int return false } -// wildcardMatch 通配符匹配候选项（用于排序） -type wildcardMatch struct { - prefixLen int - pricing *ChannelModelPricing -} - // findPricingForModel 在定价列表中查找匹配的模型定价。 -// 先精确匹配，再通配符匹配（前缀越长优先级越高）。 +// 先精确匹配，再通配符匹配（按配置顺序，先匹配先使用）。 func findPricingForModel(pricingList []ChannelModelPricing, platform, modelLower string) *ChannelModelPricing { // 精确匹配优先 for i := range pricingList { @@ -125,8 +118,7 @@ func findPricingForModel(pricingList []ChannelModelPricing, platform, modelLower } } } - // 通配符匹配：收集所有匹配项，按前缀长度降序取最长 - var matches []wildcardMatch + // 通配符匹配：按配置顺序，先匹配先使用 for i := range pricingList { p := &pricingList[i] if !isPlatformMatch(platform, p.Platform) { @@ -139,17 +131,11 @@ func findPricingForModel(pricingList []ChannelModelPricing, platform, modelLower } prefix := strings.TrimSuffix(ml, "*") if strings.HasPrefix(modelLower, prefix) { - matches = append(matches, wildcardMatch{prefixLen: len(prefix), pricing: p}) + return p } } } - if len(matches) == 0 { - return nil - } - sort.Slice(matches, func(i, j int) bool { - return matches[i].prefixLen > matches[j].prefixLen - }) - return matches[0].pricing + return nil } // isPlatformMatch 判断平台是否匹配（空平台视为不限平台）。 diff --git a/backend/internal/service/auth_service_register_test.go b/backend/internal/service/auth_service_register_test.go index 0999b4f0e7..f012245534 100644 --- a/backend/internal/service/auth_service_register_test.go +++ b/backend/internal/service/auth_service_register_test.go @@ -119,6 +119,14 @@ func (s *emailCacheStub) SetPasswordResetEmailCooldown(ctx context.Context, emai return nil } +func (s *emailCacheStub) IncrNotifyCodeUserRate(ctx context.Context, userID int64, window time.Duration) (int64, error) { + return 0, nil +} + +func (s *emailCacheStub) GetNotifyCodeUserRate(ctx context.Context, userID int64) (int64, error) { + return 0, nil +} + func newAuthService(repo *userRepoStub, settings map[string]string, emailCache EmailCache) *AuthService { cfg := &config.Config{ JWT: config.JWTConfig{ diff --git a/backend/internal/service/channel_service.go b/backend/internal/service/channel_service.go index a4024d0fa3..62429fa688 100644 --- a/backend/internal/service/channel_service.go +++ b/backend/internal/service/channel_service.go @@ -81,9 +81,9 @@ type wildcardMappingEntry struct { type channelCache struct { // 热路径查找 pricingByGroupModel map[channelModelKey]*ChannelModelPricing // (groupID, platform, model) → 定价 - wildcardByGroupPlatform map[channelGroupPlatformKey][]*wildcardPricingEntry // (groupID, platform) → 通配符定价（前缀长度降序） + wildcardByGroupPlatform map[channelGroupPlatformKey][]*wildcardPricingEntry // (groupID, platform) → 通配符定价（按配置顺序，先匹配先使用） mappingByGroupModel map[channelModelKey]string // (groupID, platform, model) → 映射目标 - wildcardMappingByGP map[channelGroupPlatformKey][]*wildcardMappingEntry // (groupID, platform) → 通配符映射（前缀长度降序） + wildcardMappingByGP map[channelGroupPlatformKey][]*wildcardMappingEntry // (groupID, platform) → 通配符映射（按配置顺序，先匹配先使用） channelByGroupID map[int64]*Channel // groupID → 渠道 groupPlatform map[int64]string // groupID → platform diff --git a/backend/internal/service/email_service.go b/backend/internal/service/email_service.go index 610907769a..88e0f622d0 100644 --- a/backend/internal/service/email_service.go +++ b/backend/internal/service/email_service.go @@ -48,6 +48,10 @@ type EmailCache interface { // Returns true if in cooldown period (email was sent recently) IsPasswordResetEmailInCooldown(ctx context.Context, email string) bool SetPasswordResetEmailCooldown(ctx context.Context, email string, ttl time.Duration) error + + // User-level rate limiting for notify email verification codes + IncrNotifyCodeUserRate(ctx context.Context, userID int64, window time.Duration) (int64, error) + GetNotifyCodeUserRate(ctx context.Context, userID int64) (int64, error) } // VerificationCodeData represents verification code data diff --git a/backend/internal/service/gateway_websearch_emulation.go b/backend/internal/service/gateway_websearch_emulation.go index 0d0b5480e2..a42b5585e3 100644 --- a/backend/internal/service/gateway_websearch_emulation.go +++ b/backend/internal/service/gateway_websearch_emulation.go @@ -214,17 +214,23 @@ func writeWebSearchStreamResponse( ) (*ForwardResult, error) { msgID := webSearchMsgIDPrefix + uuid.New().String() toolUseID := webSearchToolUseIDPrefix + uuid.New().String()[:16] + textSummary := buildTextSummary(query, resp.Results) setSSEHeaders(c) - if err := writeSSEMessageStart(c.Writer, msgID, model); err != nil { - return nil, fmt.Errorf("web search emulation: SSE write: %w", err) + w := c.Writer + for _, fn := range []func() error{ + func() error { return writeSSEMessageStart(w, msgID, model) }, + func() error { return writeSSEServerToolUse(w, toolUseID, query, 0) }, + func() error { return writeSSEToolResult(w, toolUseID, resp.Results, 1) }, + func() error { return writeSSETextBlock(w, textSummary, 2) }, + func() error { return writeSSEMessageEnd(w, len(textSummary)/tokenEstimateDivisor) }, + } { + if err := fn(); err != nil { + slog.Warn("web search emulation: SSE write failed, stopping", "error", err) + break + } } - writeSSEServerToolUse(c.Writer, toolUseID, query, 0) - writeSSEToolResult(c.Writer, toolUseID, resp.Results, 1) - textSummary := buildTextSummary(query, resp.Results) - writeSSETextBlock(c.Writer, textSummary, 2) - writeSSEMessageEnd(c.Writer, len(textSummary)/tokenEstimateDivisor) - c.Writer.Flush() + w.Flush() return &ForwardResult{Model: model, Duration: time.Since(startTime), Usage: ClaudeUsage{}}, nil } @@ -249,7 +255,7 @@ func writeSSEMessageStart(w http.ResponseWriter, msgID, model string) error { return flushSSEJSON(w, "message_start", evt) } -func writeSSEServerToolUse(w http.ResponseWriter, toolUseID, query string, index int) { +func writeSSEServerToolUse(w http.ResponseWriter, toolUseID, query string, index int) error { start := map[string]any{ "type": "content_block_start", "index": index, "content_block": map[string]any{ @@ -257,11 +263,13 @@ func writeSSEServerToolUse(w http.ResponseWriter, toolUseID, query string, index "name": toolNameWebSearch, "input": map[string]string{"query": query}, }, } - _ = flushSSEJSON(w, "content_block_start", start) - _ = flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) + if err := flushSSEJSON(w, "content_block_start", start); err != nil { + return err + } + return flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) } -func writeSSEToolResult(w http.ResponseWriter, toolUseID string, results []websearch.SearchResult, index int) { +func writeSSEToolResult(w http.ResponseWriter, toolUseID string, results []websearch.SearchResult, index int) error { start := map[string]any{ "type": "content_block_start", "index": index, "content_block": map[string]any{ @@ -269,40 +277,48 @@ func writeSSEToolResult(w http.ResponseWriter, toolUseID string, results []webse "content": buildSearchResultBlocks(results), }, } - _ = flushSSEJSON(w, "content_block_start", start) - _ = flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) + if err := flushSSEJSON(w, "content_block_start", start); err != nil { + return err + } + return flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) } -func writeSSETextBlock(w http.ResponseWriter, text string, index int) { - _ = flushSSEJSON(w, "content_block_start", map[string]any{ +func writeSSETextBlock(w http.ResponseWriter, text string, index int) error { + if err := flushSSEJSON(w, "content_block_start", map[string]any{ "type": "content_block_start", "index": index, "content_block": map[string]any{"type": "text", "text": ""}, - }) - _ = flushSSEJSON(w, "content_block_delta", map[string]any{ + }); err != nil { + return err + } + if err := flushSSEJSON(w, "content_block_delta", map[string]any{ "type": "content_block_delta", "index": index, "delta": map[string]string{"type": "text_delta", "text": text}, - }) - _ = flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) + }); err != nil { + return err + } + return flushSSEJSON(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index}) } -func writeSSEMessageEnd(w http.ResponseWriter, outputTokens int) { - _ = flushSSEJSON(w, "message_delta", map[string]any{ +func writeSSEMessageEnd(w http.ResponseWriter, outputTokens int) error { + if err := flushSSEJSON(w, "message_delta", map[string]any{ "type": "message_delta", "delta": map[string]any{"stop_reason": "end_turn", "stop_sequence": nil}, "usage": map[string]int{"output_tokens": outputTokens}, - }) - _ = flushSSEJSON(w, "message_stop", map[string]string{"type": "message_stop"}) + }); err != nil { + return err + } + return flushSSEJSON(w, "message_stop", map[string]string{"type": "message_stop"}) } -// flushSSEJSON marshals data to JSON and writes an SSE event. Returns error on marshal failure. +// flushSSEJSON marshals data to JSON and writes an SSE event. func flushSSEJSON(w http.ResponseWriter, event string, data any) error { b, err := json.Marshal(data) if err != nil { - slog.Error("web search emulation: failed to marshal SSE event", - "event", event, "error", err) - return err + return fmt.Errorf("marshal: %w", err) + } + if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, b); err != nil { + return fmt.Errorf("write: %w", err) } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, b) if f, ok := w.(http.Flusher); ok { f.Flush() } diff --git a/backend/internal/service/user_service.go b/backend/internal/service/user_service.go index bcb21c1d05..0a0a02c8ce 100644 --- a/backend/internal/service/user_service.go +++ b/backend/internal/service/user_service.go @@ -4,7 +4,7 @@ import ( "context" "crypto/subtle" "fmt" - "log" + "log/slog" "strings" "time" @@ -13,12 +13,19 @@ import ( ) var ( - ErrUserNotFound = infraerrors.NotFound("USER_NOT_FOUND", "user not found") - ErrPasswordIncorrect = infraerrors.BadRequest("PASSWORD_INCORRECT", "current password is incorrect") - ErrInsufficientPerms = infraerrors.Forbidden("INSUFFICIENT_PERMISSIONS", "insufficient permissions") + ErrUserNotFound = infraerrors.NotFound("USER_NOT_FOUND", "user not found") + ErrPasswordIncorrect = infraerrors.BadRequest("PASSWORD_INCORRECT", "current password is incorrect") + ErrInsufficientPerms = infraerrors.Forbidden("INSUFFICIENT_PERMISSIONS", "insufficient permissions") + ErrNotifyCodeUserRateLimit = infraerrors.TooManyRequests("NOTIFY_CODE_USER_RATE_LIMIT", "too many verification codes requested, please try again later") ) -const maxNotifyEmails = 3 // Total limit: primary (email="") + up to 2 extra +const ( + maxNotifyEmails = 3 // Total limit: primary (email="") + up to 2 extra + + // User-level rate limiting for notify email verification codes + notifyCodeUserRateLimit = 5 + notifyCodeUserRateWindow = 10 * time.Minute +) // UserListFilters contains all filter options for listing users type UserListFilters struct { @@ -220,7 +227,7 @@ func (s *UserService) UpdateBalance(ctx context.Context, userID int64, amount fl cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := s.billingCache.InvalidateUserBalance(cacheCtx, userID); err != nil { - log.Printf("invalidate user balance cache failed: user_id=%d err=%v", userID, err) + slog.Error("invalidate user balance cache failed", "user_id", userID, "error", err) } }() } @@ -270,21 +277,44 @@ func (s *UserService) Delete(ctx context.Context, userID int64) error { // SendNotifyEmailCode sends a verification code to the extra notification email. func (s *UserService) SendNotifyEmailCode(ctx context.Context, userID int64, email string, emailService *EmailService, cache EmailCache) error { - // Check cooldown + if err := checkNotifyCodeRateLimit(ctx, cache, userID, email); err != nil { + return err + } + + code, err := emailService.GenerateVerifyCode() + if err != nil { + return fmt.Errorf("generate code: %w", err) + } + + if err := saveNotifyVerifyCode(ctx, cache, email, code); err != nil { + return err + } + + // Increment user-level counter after successful save + if _, err := cache.IncrNotifyCodeUserRate(ctx, userID, notifyCodeUserRateWindow); err != nil { + slog.Error("failed to increment notify code user rate", "user_id", userID, "error", err) + } + + return s.sendNotifyVerifyEmail(ctx, emailService, email, code) +} + +// checkNotifyCodeRateLimit checks both email cooldown and user-level rate limit. +func checkNotifyCodeRateLimit(ctx context.Context, cache EmailCache, userID int64, email string) error { existing, err := cache.GetNotifyVerifyCode(ctx, email) if err == nil && existing != nil { if time.Since(existing.CreatedAt) < verifyCodeCooldown { return ErrVerifyCodeTooFrequent } } - - // Generate code - code, err := emailService.GenerateVerifyCode() - if err != nil { - return fmt.Errorf("generate code: %w", err) + count, err := cache.GetNotifyCodeUserRate(ctx, userID) + if err == nil && count >= notifyCodeUserRateLimit { + return ErrNotifyCodeUserRateLimit } + return nil +} - // Save to cache +// saveNotifyVerifyCode saves the verification code to cache. +func saveNotifyVerifyCode(ctx context.Context, cache EmailCache, email, code string) error { data := &VerificationCodeData{ Code: code, Attempts: 0, @@ -293,16 +323,17 @@ func (s *UserService) SendNotifyEmailCode(ctx context.Context, userID int64, ema if err := cache.SetNotifyVerifyCode(ctx, email, data, verifyCodeTTL); err != nil { return fmt.Errorf("save verify code: %w", err) } + return nil +} - // Get site name +// sendNotifyVerifyEmail builds and sends the verification email. +func (s *UserService) sendNotifyVerifyEmail(ctx context.Context, emailService *EmailService, email, code string) error { siteName := "Sub2API" if s.settingRepo != nil { if name, err := s.settingRepo.GetValue(ctx, SettingKeySiteName); err == nil && name != "" { siteName = name } } - - // Build and send email subject := fmt.Sprintf("[%s] 通知邮箱验证码 / Notification Email Verification", siteName) body := buildNotifyVerifyEmailBody(code, siteName) return emailService.SendEmail(ctx, email, subject, body) @@ -310,7 +341,15 @@ func (s *UserService) SendNotifyEmailCode(ctx context.Context, userID int64, ema // VerifyAndAddNotifyEmail verifies the code and adds the email to user's extra emails. func (s *UserService) VerifyAndAddNotifyEmail(ctx context.Context, userID int64, email, code string, cache EmailCache) error { - // Verify code + if err := verifyNotifyCode(ctx, cache, email, code); err != nil { + return err + } + _ = cache.DeleteNotifyVerifyCode(ctx, email) + return s.addOrVerifyNotifyEmail(ctx, userID, email) +} + +// verifyNotifyCode validates the verification code against the cached data. +func verifyNotifyCode(ctx context.Context, cache EmailCache, email, code string) error { data, err := cache.GetNotifyVerifyCode(ctx, email) if err != nil || data == nil { return ErrInvalidVerifyCode @@ -326,17 +365,18 @@ func (s *UserService) VerifyAndAddNotifyEmail(ctx context.Context, userID int64, } return ErrInvalidVerifyCode } + return nil +} - // Delete code after verification - _ = cache.DeleteNotifyVerifyCode(ctx, email) - - // Add to user's extra emails +// addOrVerifyNotifyEmail adds the email to user's extra notification emails or marks it as verified. +// Note: concurrent calls for the same user could race on the read-modify-write of +// BalanceNotifyExtraEmails. The window is small (requires two verify flows completing +// simultaneously), and the worst case is a duplicate entry which is harmless. +func (s *UserService) addOrVerifyNotifyEmail(ctx context.Context, userID int64, email string) error { user, err := s.userRepo.GetByID(ctx, userID) if err != nil { return err } - - // Check if already exists — if unverified, mark as verified for i, e := range user.BalanceNotifyExtraEmails { if strings.EqualFold(e.Email, email) { if !e.Verified { @@ -346,12 +386,9 @@ func (s *UserService) VerifyAndAddNotifyEmail(ctx context.Context, userID int64, return nil // Already verified } } - - // Check limit if len(user.BalanceNotifyExtraEmails) >= maxNotifyEmails { return infraerrors.BadRequest("TOO_MANY_NOTIFY_EMAILS", fmt.Sprintf("maximum %d notification emails allowed", maxNotifyEmails)) } - user.BalanceNotifyExtraEmails = append(user.BalanceNotifyExtraEmails, NotifyEmailEntry{ Email: email, Disabled: false, @@ -399,10 +436,9 @@ func (s *UserService) ToggleNotifyEmail(ctx context.Context, userID int64, email return s.userRepo.Update(ctx, user) } -// buildNotifyVerifyEmailBody builds the HTML email body for notify email verification. -func buildNotifyVerifyEmailBody(code, siteName string) string { - return fmt.Sprintf(` - +// notifyVerifyEmailTemplate is the HTML template for notify email verification. +// Format args: siteName, code. +const notifyVerifyEmailTemplate = ` @@ -439,6 +475,9 @@ func buildNotifyVerifyEmailBody(code, siteName string) string { - -`, siteName, code) +` + +// buildNotifyVerifyEmailBody builds the HTML email body for notify email verification. +func buildNotifyVerifyEmailBody(code, siteName string) string { + return fmt.Sprintf(notifyVerifyEmailTemplate, siteName, code) } diff --git a/backend/internal/service/websearch_config.go b/backend/internal/service/websearch_config.go index 5658cec3f3..0aa892a687 100644 --- a/backend/internal/service/websearch_config.go +++ b/backend/internal/service/websearch_config.go @@ -140,6 +140,16 @@ func (s *SettingService) SaveWebSearchEmulationConfig(ctx context.Context, cfg * } s.mergeExistingAPIKeys(ctx, cfg) + // After merge, validate all enabled providers have API keys + if cfg.Enabled { + for _, p := range cfg.Providers { + if p.APIKey == "" { + return infraerrors.BadRequest("MISSING_API_KEY", + fmt.Sprintf("provider %s has no API key configured", p.Type)) + } + } + } + data, err := json.Marshal(cfg) if err != nil { return fmt.Errorf("websearch: marshal config: %w", err)