fix: address remaining audit findings (rate limit, wildcard order, SSE errors)

Severe fixes:
- SendNotifyEmailCode: add user-level rate limit (5 codes per 10min)
  via Redis counter, new EmailCache methods IncrNotifyCodeUserRate/
  GetNotifyCodeUserRate
- findPricingForModel: change wildcard matching from longest-prefix-first
  to configuration-order-first (first match wins), matching channel
  cache behavior
- channel_service.go: fix misleading comments about prefix length sorting

Medium fixes:
- SSE flushSSEJSON: check fmt.Fprintf write errors, chain error handling
  in writeWebSearchStreamResponse with early termination
- validateWebSearchConfig: validate APIKey non-empty after merge for
  enabled configurations
- user_service.go: log.Printf replaced with slog.Error
- buildNotifyVerifyEmailBody: extract HTML template to const
- SendNotifyEmailCode: split into checkNotifyCodeRateLimit,
  saveNotifyVerifyCode, sendNotifyVerifyEmail (each ≤30 lines)
- VerifyAndAddNotifyEmail: split into verifyNotifyCode,
  addOrVerifyNotifyEmail (each ≤30 lines)
- Document known TOCTOU race in addOrVerifyNotifyEmail (small window,
  harmless worst case)
This commit is contained in:
erio
2026-04-13 12:41:37 +08:00
parent 68f546e3ec
commit 22a4c1f711
8 changed files with 178 additions and 85 deletions
+34 -4
View File
@@ -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
}
@@ -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 判断平台是否匹配(空平台视为不限平台)。
@@ -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{
+2 -2
View File
@@ -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
@@ -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
@@ -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()
}
+71 -32
View File
@@ -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(`
<!DOCTYPE html>
// notifyVerifyEmailTemplate is the HTML template for notify email verification.
// Format args: siteName, code.
const notifyVerifyEmailTemplate = `<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
@@ -439,6 +475,9 @@ func buildNotifyVerifyEmailBody(code, siteName string) string {
</div>
</div>
</body>
</html>
`, siteName, code)
</html>`
// buildNotifyVerifyEmailBody builds the HTML email body for notify email verification.
func buildNotifyVerifyEmailBody(code, siteName string) string {
return fmt.Sprintf(notifyVerifyEmailTemplate, siteName, code)
}
@@ -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)