feat(proxy,sora): 增强代理质量检测与Sora稳定性并修复审查问题

This commit is contained in:
yangjianbo
2026-02-19 21:18:35 +08:00
parent 36a1a7998b
commit 46d9aee6dd
23 changed files with 1408 additions and 45 deletions

View File

@@ -4,11 +4,15 @@ import (
"context"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
"github.com/Wei-Shaw/sub2api/internal/util/soraerror"
)
// AdminService interface defines admin management operations
@@ -65,6 +69,7 @@ type AdminService interface {
GetProxyAccounts(ctx context.Context, proxyID int64) ([]ProxyAccountSummary, error)
CheckProxyExists(ctx context.Context, host string, port int, username, password string) (bool, error)
TestProxy(ctx context.Context, id int64) (*ProxyTestResult, error)
CheckProxyQuality(ctx context.Context, id int64) (*ProxyQualityCheckResult, error)
// Redeem code management
ListRedeemCodes(ctx context.Context, page, pageSize int, codeType, status, search string) ([]RedeemCode, int64, error)
@@ -288,6 +293,32 @@ type ProxyTestResult struct {
CountryCode string `json:"country_code,omitempty"`
}
type ProxyQualityCheckResult struct {
ProxyID int64 `json:"proxy_id"`
Score int `json:"score"`
Grade string `json:"grade"`
Summary string `json:"summary"`
ExitIP string `json:"exit_ip,omitempty"`
Country string `json:"country,omitempty"`
CountryCode string `json:"country_code,omitempty"`
BaseLatencyMs int64 `json:"base_latency_ms,omitempty"`
PassedCount int `json:"passed_count"`
WarnCount int `json:"warn_count"`
FailedCount int `json:"failed_count"`
ChallengeCount int `json:"challenge_count"`
CheckedAt int64 `json:"checked_at"`
Items []ProxyQualityCheckItem `json:"items"`
}
type ProxyQualityCheckItem struct {
Target string `json:"target"`
Status string `json:"status"` // pass/warn/fail/challenge
HTTPStatus int `json:"http_status,omitempty"`
LatencyMs int64 `json:"latency_ms,omitempty"`
Message string `json:"message,omitempty"`
CFRay string `json:"cf_ray,omitempty"`
}
// ProxyExitInfo represents proxy exit information from ip-api.com
type ProxyExitInfo struct {
IP string
@@ -302,6 +333,58 @@ type ProxyExitInfoProber interface {
ProbeProxy(ctx context.Context, proxyURL string) (*ProxyExitInfo, int64, error)
}
type proxyQualityTarget struct {
Target string
URL string
Method string
AllowedStatuses map[int]struct{}
}
var proxyQualityTargets = []proxyQualityTarget{
{
Target: "openai",
URL: "https://api.openai.com/v1/models",
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusUnauthorized: {},
},
},
{
Target: "anthropic",
URL: "https://api.anthropic.com/v1/messages",
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusUnauthorized: {},
http.StatusMethodNotAllowed: {},
http.StatusNotFound: {},
http.StatusBadRequest: {},
},
},
{
Target: "gemini",
URL: "https://generativelanguage.googleapis.com/$discovery/rest?version=v1beta",
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusOK: {},
},
},
{
Target: "sora",
URL: "https://sora.chatgpt.com/backend/me",
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusUnauthorized: {},
},
},
}
const (
proxyQualityRequestTimeout = 15 * time.Second
proxyQualityResponseHeaderTimeout = 10 * time.Second
proxyQualityMaxBodyBytes = int64(8 * 1024)
proxyQualityClientUserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
)
// adminServiceImpl implements AdminService
type adminServiceImpl struct {
userRepo UserRepository
@@ -1690,6 +1773,187 @@ func (s *adminServiceImpl) TestProxy(ctx context.Context, id int64) (*ProxyTestR
}, nil
}
func (s *adminServiceImpl) CheckProxyQuality(ctx context.Context, id int64) (*ProxyQualityCheckResult, error) {
proxy, err := s.proxyRepo.GetByID(ctx, id)
if err != nil {
return nil, err
}
result := &ProxyQualityCheckResult{
ProxyID: id,
Score: 100,
Grade: "A",
CheckedAt: time.Now().Unix(),
Items: make([]ProxyQualityCheckItem, 0, len(proxyQualityTargets)+1),
}
proxyURL := proxy.URL()
if s.proxyProber == nil {
result.Items = append(result.Items, ProxyQualityCheckItem{
Target: "base_connectivity",
Status: "fail",
Message: "代理探测服务未配置",
})
result.FailedCount++
finalizeProxyQualityResult(result)
return result, nil
}
exitInfo, latencyMs, err := s.proxyProber.ProbeProxy(ctx, proxyURL)
if err != nil {
result.Items = append(result.Items, ProxyQualityCheckItem{
Target: "base_connectivity",
Status: "fail",
LatencyMs: latencyMs,
Message: err.Error(),
})
result.FailedCount++
finalizeProxyQualityResult(result)
return result, nil
}
result.ExitIP = exitInfo.IP
result.Country = exitInfo.Country
result.CountryCode = exitInfo.CountryCode
result.BaseLatencyMs = latencyMs
result.Items = append(result.Items, ProxyQualityCheckItem{
Target: "base_connectivity",
Status: "pass",
LatencyMs: latencyMs,
Message: "代理出口连通正常",
})
result.PassedCount++
client, err := httpclient.GetClient(httpclient.Options{
ProxyURL: proxyURL,
Timeout: proxyQualityRequestTimeout,
ResponseHeaderTimeout: proxyQualityResponseHeaderTimeout,
ProxyStrict: true,
})
if err != nil {
result.Items = append(result.Items, ProxyQualityCheckItem{
Target: "http_client",
Status: "fail",
Message: fmt.Sprintf("创建检测客户端失败: %v", err),
})
result.FailedCount++
finalizeProxyQualityResult(result)
return result, nil
}
for _, target := range proxyQualityTargets {
item := runProxyQualityTarget(ctx, client, target)
result.Items = append(result.Items, item)
switch item.Status {
case "pass":
result.PassedCount++
case "warn":
result.WarnCount++
case "challenge":
result.ChallengeCount++
default:
result.FailedCount++
}
}
finalizeProxyQualityResult(result)
return result, nil
}
func runProxyQualityTarget(ctx context.Context, client *http.Client, target proxyQualityTarget) ProxyQualityCheckItem {
item := ProxyQualityCheckItem{
Target: target.Target,
}
req, err := http.NewRequestWithContext(ctx, target.Method, target.URL, nil)
if err != nil {
item.Status = "fail"
item.Message = fmt.Sprintf("构建请求失败: %v", err)
return item
}
req.Header.Set("Accept", "application/json,text/html,*/*")
req.Header.Set("User-Agent", proxyQualityClientUserAgent)
start := time.Now()
resp, err := client.Do(req)
if err != nil {
item.Status = "fail"
item.LatencyMs = time.Since(start).Milliseconds()
item.Message = fmt.Sprintf("请求失败: %v", err)
return item
}
defer func() { _ = resp.Body.Close() }()
item.LatencyMs = time.Since(start).Milliseconds()
item.HTTPStatus = resp.StatusCode
body, readErr := io.ReadAll(io.LimitReader(resp.Body, proxyQualityMaxBodyBytes+1))
if readErr != nil {
item.Status = "fail"
item.Message = fmt.Sprintf("读取响应失败: %v", readErr)
return item
}
if int64(len(body)) > proxyQualityMaxBodyBytes {
body = body[:proxyQualityMaxBodyBytes]
}
if target.Target == "sora" && soraerror.IsCloudflareChallengeResponse(resp.StatusCode, resp.Header, body) {
item.Status = "challenge"
item.CFRay = soraerror.ExtractCloudflareRayID(resp.Header, body)
item.Message = "Sora 命中 Cloudflare challenge"
return item
}
if _, ok := target.AllowedStatuses[resp.StatusCode]; ok {
item.Status = "pass"
item.Message = fmt.Sprintf("HTTP %d", resp.StatusCode)
return item
}
if resp.StatusCode == http.StatusTooManyRequests {
item.Status = "warn"
item.Message = "目标返回 429可能存在频控"
return item
}
item.Status = "fail"
item.Message = fmt.Sprintf("非预期状态码: %d", resp.StatusCode)
return item
}
func finalizeProxyQualityResult(result *ProxyQualityCheckResult) {
if result == nil {
return
}
score := 100 - result.WarnCount*10 - result.FailedCount*22 - result.ChallengeCount*30
if score < 0 {
score = 0
}
result.Score = score
result.Grade = proxyQualityGrade(score)
result.Summary = fmt.Sprintf(
"通过 %d 项,告警 %d 项,失败 %d 项,挑战 %d 项",
result.PassedCount,
result.WarnCount,
result.FailedCount,
result.ChallengeCount,
)
}
func proxyQualityGrade(score int) string {
switch {
case score >= 90:
return "A"
case score >= 75:
return "B"
case score >= 60:
return "C"
case score >= 40:
return "D"
default:
return "F"
}
}
func (s *adminServiceImpl) probeProxyLatency(ctx context.Context, proxy *Proxy) {
if s.proxyProber == nil || proxy == nil {
return

View File

@@ -0,0 +1,73 @@
package service
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/require"
)
func TestFinalizeProxyQualityResult_ScoreAndGrade(t *testing.T) {
result := &ProxyQualityCheckResult{
PassedCount: 2,
WarnCount: 1,
FailedCount: 1,
ChallengeCount: 1,
}
finalizeProxyQualityResult(result)
require.Equal(t, 38, result.Score)
require.Equal(t, "F", result.Grade)
require.Contains(t, result.Summary, "通过 2 项")
require.Contains(t, result.Summary, "告警 1 项")
require.Contains(t, result.Summary, "失败 1 项")
require.Contains(t, result.Summary, "挑战 1 项")
}
func TestRunProxyQualityTarget_SoraChallenge(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/html")
w.Header().Set("cf-ray", "test-ray-123")
w.WriteHeader(http.StatusForbidden)
_, _ = w.Write([]byte("<!DOCTYPE html><title>Just a moment...</title><script>window._cf_chl_opt={};</script>"))
}))
defer server.Close()
target := proxyQualityTarget{
Target: "sora",
URL: server.URL,
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusUnauthorized: {},
},
}
item := runProxyQualityTarget(context.Background(), server.Client(), target)
require.Equal(t, "challenge", item.Status)
require.Equal(t, http.StatusForbidden, item.HTTPStatus)
require.Equal(t, "test-ray-123", item.CFRay)
}
func TestRunProxyQualityTarget_AllowedStatusPass(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"error":"unauthorized"}`))
}))
defer server.Close()
target := proxyQualityTarget{
Target: "openai",
URL: server.URL,
Method: http.MethodGet,
AllowedStatuses: map[int]struct{}{
http.StatusUnauthorized: {},
},
}
item := runProxyQualityTarget(context.Background(), server.Client(), target)
require.Equal(t, "pass", item.Status)
require.Equal(t, http.StatusUnauthorized, item.HTTPStatus)
}

View File

@@ -375,10 +375,10 @@ type ForwardResult struct {
// UpstreamFailoverError indicates an upstream error that should trigger account failover.
type UpstreamFailoverError struct {
StatusCode int
ResponseBody []byte // 上游响应体,用于错误透传规则匹配
ResponseHeaders http.Header
ForceCacheBilling bool // Antigravity 粘性会话切换时设为 true
RetryableOnSameAccount bool // 临时性错误(如 Google 间歇性 400、空响应应在同一账号上重试 N 次再切换
ResponseBody []byte // 上游响应体,用于错误透传规则匹配
ResponseHeaders http.Header // 上游响应头,用于透传 cf-ray/cf-mitigated/content-type 等诊断信息
ForceCacheBilling bool // Antigravity 粘性会话切换时设为 true
RetryableOnSameAccount bool // 临时性错误(如 Google 间歇性 400、空响应应在同一账号上重试 N 次再切换
}
func (e *UpstreamFailoverError) Error() string {

View File

@@ -27,6 +27,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config"
openaioauth "github.com/Wei-Shaw/sub2api/internal/pkg/openai"
"github.com/Wei-Shaw/sub2api/internal/util/logredact"
"github.com/Wei-Shaw/sub2api/internal/util/soraerror"
"github.com/google/uuid"
"github.com/tidwall/gjson"
"golang.org/x/crypto/sha3"
@@ -221,12 +222,16 @@ func (e *SoraUpstreamError) Error() string {
// SoraDirectClient 直连 Sora 实现
type SoraDirectClient struct {
cfg *config.Config
httpUpstream HTTPUpstream
tokenProvider *OpenAITokenProvider
accountRepo AccountRepository
soraAccountRepo SoraAccountRepository
baseURL string
cfg *config.Config
httpUpstream HTTPUpstream
tokenProvider *OpenAITokenProvider
accountRepo AccountRepository
soraAccountRepo SoraAccountRepository
baseURL string
challengeCooldownMu sync.RWMutex
challengeCooldowns map[string]soraChallengeCooldownEntry
sidecarSessionMu sync.RWMutex
sidecarSessions map[string]soraSidecarSessionEntry
}
// NewSoraDirectClient 创建 Sora 直连客户端
@@ -240,10 +245,12 @@ func NewSoraDirectClient(cfg *config.Config, httpUpstream HTTPUpstream, tokenPro
}
}
return &SoraDirectClient{
cfg: cfg,
httpUpstream: httpUpstream,
tokenProvider: tokenProvider,
baseURL: baseURL,
cfg: cfg,
httpUpstream: httpUpstream,
tokenProvider: tokenProvider,
baseURL: baseURL,
challengeCooldowns: make(map[string]soraChallengeCooldownEntry),
sidecarSessions: make(map[string]soraSidecarSessionEntry),
}
}
@@ -1461,6 +1468,9 @@ func (c *SoraDirectClient) doRequestWithProxy(
if proxyURL == "" {
proxyURL = c.resolveProxyURL(account)
}
if cooldownErr := c.checkCloudflareChallengeCooldown(account, proxyURL); cooldownErr != nil {
return nil, nil, cooldownErr
}
timeout := 0
if c != nil && c.cfg != nil {
timeout = c.cfg.Sora.Client.TimeoutSeconds
@@ -1561,7 +1571,11 @@ func (c *SoraDirectClient) doRequestWithProxy(
}
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
if !authRecovered && shouldAttemptSoraTokenRecover(resp.StatusCode, urlStr) && account != nil {
isCFChallenge := soraerror.IsCloudflareChallengeResponse(resp.StatusCode, resp.Header, respBody)
if isCFChallenge {
c.recordCloudflareChallengeCooldown(account, proxyURL, resp.StatusCode, resp.Header, respBody)
}
if !isCFChallenge && !authRecovered && shouldAttemptSoraTokenRecover(resp.StatusCode, urlStr) && account != nil {
if recovered, recoverErr := c.recoverAccessToken(ctx, account, fmt.Sprintf("upstream_status_%d", resp.StatusCode)); recoverErr == nil && strings.TrimSpace(recovered) != "" {
headers.Set("Authorization", "Bearer "+recovered)
authRecovered = true
@@ -1590,6 +1604,9 @@ func (c *SoraDirectClient) doRequestWithProxy(
}
upstreamErr := c.buildUpstreamError(resp.StatusCode, resp.Header, respBody, urlStr)
lastErr = upstreamErr
if isCFChallenge {
return nil, resp.Header, upstreamErr
}
if allowRetry && attempt < attempts && (resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500) {
if c.debugEnabled() {
c.debugLogf("request_retry_scheduled method=%s url=%s reason=status_%d next_attempt=%d/%d", method, sanitizeSoraLogURL(urlStr), resp.StatusCode, attempt+1, attempts)
@@ -1631,7 +1648,7 @@ func shouldAttemptSoraTokenRecover(statusCode int, rawURL string) bool {
func (c *SoraDirectClient) doHTTP(req *http.Request, proxyURL string, account *Account) (*http.Response, error) {
if c != nil && c.cfg != nil && c.cfg.Sora.Client.CurlCFFISidecar.Enabled {
resp, err := c.doHTTPViaCurlCFFISidecar(req, proxyURL)
resp, err := c.doHTTPViaCurlCFFISidecar(req, proxyURL, account)
if err != nil {
return nil, err
}

View File

@@ -693,10 +693,11 @@ func TestSoraDirectClient_DoHTTP_UsesCurlCFFISidecarWhenEnabled(t *testing.T) {
Client: config.SoraClientConfig{
BaseURL: "https://sora.chatgpt.com/backend",
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
BaseURL: sidecar.URL,
Impersonate: "chrome131",
TimeoutSeconds: 15,
Enabled: true,
BaseURL: sidecar.URL,
Impersonate: "chrome131",
TimeoutSeconds: 15,
SessionReuseEnabled: true,
},
},
},
@@ -715,6 +716,7 @@ func TestSoraDirectClient_DoHTTP_UsesCurlCFFISidecarWhenEnabled(t *testing.T) {
require.JSONEq(t, `{"ok":true}`, string(body))
require.Equal(t, int32(0), atomic.LoadInt32(&upstream.doWithTLSCalls))
require.Equal(t, "http://127.0.0.1:18080", captured.ProxyURL)
require.NotEmpty(t, captured.SessionKey)
require.Equal(t, "chrome131", captured.Impersonate)
require.Equal(t, "https://sora.chatgpt.com/backend/me", captured.URL)
decodedReqBody, err := base64.StdEncoding.DecodeString(captured.BodyBase64)
@@ -781,3 +783,188 @@ func TestConvertSidecarHeaderValue_NilAndSlice(t *testing.T) {
require.Nil(t, convertSidecarHeaderValue(nil))
require.Equal(t, []string{"a", "b"}, convertSidecarHeaderValue([]any{"a", " ", "b"}))
}
func TestSoraDirectClient_DoHTTP_SidecarSessionKeyStableForSameAccountProxy(t *testing.T) {
var captured []soraCurlCFFISidecarRequest
sidecar := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
raw, err := io.ReadAll(r.Body)
require.NoError(t, err)
var reqPayload soraCurlCFFISidecarRequest
require.NoError(t, json.Unmarshal(raw, &reqPayload))
captured = append(captured, reqPayload)
_ = json.NewEncoder(w).Encode(map[string]any{
"status_code": http.StatusOK,
"headers": map[string]any{
"Content-Type": "application/json",
},
"body": `{"ok":true}`,
})
}))
defer sidecar.Close()
cfg := &config.Config{
Sora: config.SoraConfig{
Client: config.SoraClientConfig{
BaseURL: "https://sora.chatgpt.com/backend",
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
BaseURL: sidecar.URL,
SessionReuseEnabled: true,
SessionTTLSeconds: 3600,
},
},
},
}
client := NewSoraDirectClient(cfg, nil, nil)
account := &Account{ID: 1001}
req1, err := http.NewRequest(http.MethodGet, "https://sora.chatgpt.com/backend/me", nil)
require.NoError(t, err)
_, err = client.doHTTP(req1, "http://127.0.0.1:18080", account)
require.NoError(t, err)
req2, err := http.NewRequest(http.MethodGet, "https://sora.chatgpt.com/backend/me", nil)
require.NoError(t, err)
_, err = client.doHTTP(req2, "http://127.0.0.1:18080", account)
require.NoError(t, err)
require.Len(t, captured, 2)
require.NotEmpty(t, captured[0].SessionKey)
require.Equal(t, captured[0].SessionKey, captured[1].SessionKey)
}
func TestSoraDirectClient_DoRequestWithProxy_CloudflareChallengeSetsCooldownAndSkipsRetry(t *testing.T) {
var sidecarCalls int32
sidecar := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&sidecarCalls, 1)
_ = json.NewEncoder(w).Encode(map[string]any{
"status_code": http.StatusForbidden,
"headers": map[string]any{
"cf-ray": "9d05d73dec4d8c8e-GRU",
"content-type": "text/html",
},
"body": `<!DOCTYPE html><html><head><title>Just a moment...</title></head><body><script>window._cf_chl_opt={};</script></body></html>`,
})
}))
defer sidecar.Close()
cfg := &config.Config{
Sora: config.SoraConfig{
Client: config.SoraClientConfig{
BaseURL: "https://sora.chatgpt.com/backend",
MaxRetries: 3,
CloudflareChallengeCooldownSeconds: 60,
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
BaseURL: sidecar.URL,
Impersonate: "chrome131",
},
},
},
}
client := NewSoraDirectClient(cfg, nil, nil)
headers := http.Header{}
_, _, err := client.doRequestWithProxy(
context.Background(),
&Account{ID: 99},
"http://127.0.0.1:18080",
http.MethodGet,
"https://sora.chatgpt.com/backend/me",
headers,
nil,
true,
)
require.Error(t, err)
var upstreamErr *SoraUpstreamError
require.ErrorAs(t, err, &upstreamErr)
require.Equal(t, http.StatusForbidden, upstreamErr.StatusCode)
require.Equal(t, int32(1), atomic.LoadInt32(&sidecarCalls), "challenge should not trigger retry loop")
_, _, err = client.doRequestWithProxy(
context.Background(),
&Account{ID: 99},
"http://127.0.0.1:18080",
http.MethodGet,
"https://sora.chatgpt.com/backend/me",
headers,
nil,
true,
)
require.Error(t, err)
require.ErrorAs(t, err, &upstreamErr)
require.Equal(t, http.StatusTooManyRequests, upstreamErr.StatusCode)
require.Contains(t, upstreamErr.Message, "cooling down")
require.Contains(t, upstreamErr.Message, "cf-ray")
require.Equal(t, int32(1), atomic.LoadInt32(&sidecarCalls), "cooldown should block outbound request")
}
func TestSoraDirectClient_SidecarSessionKey_SkipsWhenAccountMissing(t *testing.T) {
cfg := &config.Config{
Sora: config.SoraConfig{
Client: config.SoraClientConfig{
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
SessionReuseEnabled: true,
SessionTTLSeconds: 3600,
},
},
},
}
client := NewSoraDirectClient(cfg, nil, nil)
require.Equal(t, "", client.sidecarSessionKey(nil, "http://127.0.0.1:18080"))
require.Empty(t, client.sidecarSessions)
}
func TestSoraDirectClient_SidecarSessionKey_PrunesExpiredAndRecreates(t *testing.T) {
cfg := &config.Config{
Sora: config.SoraConfig{
Client: config.SoraClientConfig{
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
SessionReuseEnabled: true,
SessionTTLSeconds: 3600,
},
},
},
}
client := NewSoraDirectClient(cfg, nil, nil)
account := &Account{ID: 123}
key := soraAccountProxyKey(account, "http://127.0.0.1:18080")
client.sidecarSessions[key] = soraSidecarSessionEntry{
SessionKey: "sora-expired",
ExpiresAt: time.Now().Add(-time.Minute),
LastUsedAt: time.Now().Add(-2 * time.Minute),
}
sessionKey := client.sidecarSessionKey(account, "http://127.0.0.1:18080")
require.NotEmpty(t, sessionKey)
require.NotEqual(t, "sora-expired", sessionKey)
require.Len(t, client.sidecarSessions, 1)
}
func TestSoraDirectClient_SidecarSessionKey_TTLZeroKeepsLongLivedSession(t *testing.T) {
cfg := &config.Config{
Sora: config.SoraConfig{
Client: config.SoraClientConfig{
CurlCFFISidecar: config.SoraCurlCFFISidecarConfig{
Enabled: true,
SessionReuseEnabled: true,
SessionTTLSeconds: 0,
},
},
},
}
client := NewSoraDirectClient(cfg, nil, nil)
account := &Account{ID: 456}
first := client.sidecarSessionKey(account, "http://127.0.0.1:18080")
second := client.sidecarSessionKey(account, "http://127.0.0.1:18080")
require.NotEmpty(t, first)
require.Equal(t, first, second)
key := soraAccountProxyKey(account, "http://127.0.0.1:18080")
entry, ok := client.sidecarSessions[key]
require.True(t, ok)
require.True(t, entry.ExpiresAt.After(time.Now().Add(300*24*time.Hour)))
}

View File

@@ -23,6 +23,7 @@ type soraCurlCFFISidecarRequest struct {
Headers map[string][]string `json:"headers,omitempty"`
BodyBase64 string `json:"body_base64,omitempty"`
ProxyURL string `json:"proxy_url,omitempty"`
SessionKey string `json:"session_key,omitempty"`
Impersonate string `json:"impersonate,omitempty"`
TimeoutSeconds int `json:"timeout_seconds,omitempty"`
}
@@ -36,7 +37,7 @@ type soraCurlCFFISidecarResponse struct {
Error string `json:"error"`
}
func (c *SoraDirectClient) doHTTPViaCurlCFFISidecar(req *http.Request, proxyURL string) (*http.Response, error) {
func (c *SoraDirectClient) doHTTPViaCurlCFFISidecar(req *http.Request, proxyURL string, account *Account) (*http.Response, error) {
if req == nil || req.URL == nil {
return nil, errors.New("request url is nil")
}
@@ -73,6 +74,7 @@ func (c *SoraDirectClient) doHTTPViaCurlCFFISidecar(req *http.Request, proxyURL
URL: req.URL.String(),
Headers: headers,
ProxyURL: strings.TrimSpace(proxyURL),
SessionKey: c.sidecarSessionKey(account, proxyURL),
Impersonate: c.curlCFFIImpersonate(),
TimeoutSeconds: c.curlCFFISidecarTimeoutSeconds(),
}
@@ -97,7 +99,9 @@ func (c *SoraDirectClient) doHTTPViaCurlCFFISidecar(req *http.Request, proxyURL
if err != nil {
return nil, fmt.Errorf("sora curl_cffi sidecar request failed: %w", err)
}
defer sidecarResp.Body.Close()
defer func() {
_ = sidecarResp.Body.Close()
}()
sidecarRespBody, err := io.ReadAll(io.LimitReader(sidecarResp.Body, 8<<20))
if err != nil {
@@ -202,6 +206,24 @@ func (c *SoraDirectClient) curlCFFIImpersonate() string {
return impersonate
}
func (c *SoraDirectClient) sidecarSessionReuseEnabled() bool {
if c == nil || c.cfg == nil {
return true
}
return c.cfg.Sora.Client.CurlCFFISidecar.SessionReuseEnabled
}
func (c *SoraDirectClient) sidecarSessionTTLSeconds() int {
if c == nil || c.cfg == nil {
return 3600
}
ttl := c.cfg.Sora.Client.CurlCFFISidecar.SessionTTLSeconds
if ttl < 0 {
return 3600
}
return ttl
}
func convertSidecarHeaderValue(raw any) []string {
switch val := raw.(type) {
case nil:

View File

@@ -906,10 +906,14 @@ func (s *SoraGatewayService) handleSoraRequestError(ctx context.Context, account
s.rateLimitService.HandleUpstreamError(ctx, account, upstreamErr.StatusCode, upstreamErr.Headers, upstreamErr.Body)
}
if s.shouldFailoverUpstreamError(upstreamErr.StatusCode) {
var responseHeaders http.Header
if upstreamErr.Headers != nil {
responseHeaders = upstreamErr.Headers.Clone()
}
return &UpstreamFailoverError{
StatusCode: upstreamErr.StatusCode,
ResponseBody: upstreamErr.Body,
ResponseHeaders: upstreamErr.Headers,
ResponseHeaders: responseHeaders,
}
}
msg := upstreamErr.Message

View File

@@ -397,6 +397,34 @@ func TestSoraGatewayService_WriteSoraError_StreamEscapesJSON(t *testing.T) {
require.Equal(t, "invalid \"prompt\"\nline2", errObj["message"])
}
func TestSoraGatewayService_HandleSoraRequestError_FailoverHeadersCloned(t *testing.T) {
svc := NewSoraGatewayService(nil, nil, nil, &config.Config{})
sourceHeaders := http.Header{}
sourceHeaders.Set("cf-ray", "9d01b0e9ecc35829-SEA")
err := svc.handleSoraRequestError(
context.Background(),
&Account{ID: 1, Platform: PlatformSora},
&SoraUpstreamError{
StatusCode: http.StatusForbidden,
Message: "forbidden",
Headers: sourceHeaders,
Body: []byte(`<!DOCTYPE html><title>Just a moment...</title>`),
},
"sora2-landscape-10s",
nil,
false,
)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.NotNil(t, failoverErr.ResponseHeaders)
require.Equal(t, "9d01b0e9ecc35829-SEA", failoverErr.ResponseHeaders.Get("cf-ray"))
sourceHeaders.Set("cf-ray", "mutated-after-return")
require.Equal(t, "9d01b0e9ecc35829-SEA", failoverErr.ResponseHeaders.Get("cf-ray"))
}
func TestShouldFailoverUpstreamError(t *testing.T) {
svc := NewSoraGatewayService(nil, nil, nil, &config.Config{})
require.True(t, svc.shouldFailoverUpstreamError(401))

View File

@@ -0,0 +1,213 @@
package service
import (
"fmt"
"math"
"net/http"
"net/url"
"strings"
"time"
"github.com/Wei-Shaw/sub2api/internal/util/soraerror"
"github.com/google/uuid"
)
type soraChallengeCooldownEntry struct {
Until time.Time
StatusCode int
CFRay string
}
type soraSidecarSessionEntry struct {
SessionKey string
ExpiresAt time.Time
LastUsedAt time.Time
}
func (c *SoraDirectClient) cloudflareChallengeCooldownSeconds() int {
if c == nil || c.cfg == nil {
return 900
}
cooldown := c.cfg.Sora.Client.CloudflareChallengeCooldownSeconds
if cooldown <= 0 {
return 0
}
return cooldown
}
func (c *SoraDirectClient) checkCloudflareChallengeCooldown(account *Account, proxyURL string) error {
if c == nil {
return nil
}
if account == nil || account.ID <= 0 {
return nil
}
cooldownSeconds := c.cloudflareChallengeCooldownSeconds()
if cooldownSeconds <= 0 {
return nil
}
key := soraAccountProxyKey(account, proxyURL)
now := time.Now()
c.challengeCooldownMu.RLock()
entry, ok := c.challengeCooldowns[key]
c.challengeCooldownMu.RUnlock()
if !ok {
return nil
}
if !entry.Until.After(now) {
c.challengeCooldownMu.Lock()
delete(c.challengeCooldowns, key)
c.challengeCooldownMu.Unlock()
return nil
}
remaining := int(math.Ceil(entry.Until.Sub(now).Seconds()))
if remaining < 1 {
remaining = 1
}
message := fmt.Sprintf("Sora request cooling down due to recent Cloudflare challenge. Retry in %d seconds.", remaining)
if entry.CFRay != "" {
message = fmt.Sprintf("%s (last cf-ray: %s)", message, entry.CFRay)
}
return &SoraUpstreamError{
StatusCode: http.StatusTooManyRequests,
Message: message,
Headers: make(http.Header),
}
}
func (c *SoraDirectClient) recordCloudflareChallengeCooldown(account *Account, proxyURL string, statusCode int, headers http.Header, body []byte) {
if c == nil {
return
}
if account == nil || account.ID <= 0 {
return
}
cooldownSeconds := c.cloudflareChallengeCooldownSeconds()
if cooldownSeconds <= 0 {
return
}
key := soraAccountProxyKey(account, proxyURL)
now := time.Now()
until := now.Add(time.Duration(cooldownSeconds) * time.Second)
cfRay := soraerror.ExtractCloudflareRayID(headers, body)
c.challengeCooldownMu.Lock()
c.cleanupExpiredChallengeCooldownsLocked(now)
existing, ok := c.challengeCooldowns[key]
if ok && existing.Until.After(until) {
until = existing.Until
if cfRay == "" {
cfRay = existing.CFRay
}
}
c.challengeCooldowns[key] = soraChallengeCooldownEntry{
Until: until,
StatusCode: statusCode,
CFRay: cfRay,
}
c.challengeCooldownMu.Unlock()
if c.debugEnabled() {
remain := int(math.Ceil(until.Sub(now).Seconds()))
if remain < 0 {
remain = 0
}
c.debugLogf("cloudflare_challenge_cooldown_set key=%s status=%d remain_s=%d cf_ray=%s", key, statusCode, remain, cfRay)
}
}
func (c *SoraDirectClient) sidecarSessionKey(account *Account, proxyURL string) string {
if c == nil || !c.sidecarSessionReuseEnabled() {
return ""
}
if account == nil || account.ID <= 0 {
return ""
}
key := soraAccountProxyKey(account, proxyURL)
now := time.Now()
ttlSeconds := c.sidecarSessionTTLSeconds()
c.sidecarSessionMu.Lock()
defer c.sidecarSessionMu.Unlock()
c.cleanupExpiredSidecarSessionsLocked(now)
if existing, exists := c.sidecarSessions[key]; exists {
existing.LastUsedAt = now
c.sidecarSessions[key] = existing
return existing.SessionKey
}
expiresAt := now.Add(time.Duration(ttlSeconds) * time.Second)
if ttlSeconds <= 0 {
expiresAt = now.Add(365 * 24 * time.Hour)
}
newEntry := soraSidecarSessionEntry{
SessionKey: "sora-" + uuid.NewString(),
ExpiresAt: expiresAt,
LastUsedAt: now,
}
c.sidecarSessions[key] = newEntry
if c.debugEnabled() {
c.debugLogf("sidecar_session_created key=%s ttl_s=%d", key, ttlSeconds)
}
return newEntry.SessionKey
}
func (c *SoraDirectClient) cleanupExpiredChallengeCooldownsLocked(now time.Time) {
if c == nil || len(c.challengeCooldowns) == 0 {
return
}
for key, entry := range c.challengeCooldowns {
if !entry.Until.After(now) {
delete(c.challengeCooldowns, key)
}
}
}
func (c *SoraDirectClient) cleanupExpiredSidecarSessionsLocked(now time.Time) {
if c == nil || len(c.sidecarSessions) == 0 {
return
}
for key, entry := range c.sidecarSessions {
if !entry.ExpiresAt.After(now) {
delete(c.sidecarSessions, key)
}
}
}
func soraAccountProxyKey(account *Account, proxyURL string) string {
accountID := int64(0)
if account != nil {
accountID = account.ID
}
return fmt.Sprintf("account:%d|proxy:%s", accountID, normalizeSoraProxyKey(proxyURL))
}
func normalizeSoraProxyKey(proxyURL string) string {
raw := strings.TrimSpace(proxyURL)
if raw == "" {
return "direct"
}
parsed, err := url.Parse(raw)
if err != nil {
return strings.ToLower(raw)
}
scheme := strings.ToLower(strings.TrimSpace(parsed.Scheme))
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
port := strings.TrimSpace(parsed.Port())
if host == "" {
return strings.ToLower(raw)
}
if (scheme == "http" && port == "80") || (scheme == "https" && port == "443") {
port = ""
}
if port != "" {
host = host + ":" + port
}
if scheme == "" {
scheme = "proxy"
}
return scheme + "://" + host
}