From 4b667ccd45747162ac58545cdbb6f6d88737bf04 Mon Sep 17 00:00:00 2001 From: Randark Date: Mon, 17 Aug 2026 04:07:06 +0000 Subject: [PATCH] =?UTF-8?q?fix(review):=20=E5=A4=84=E7=90=86=20PR=20#5666?= =?UTF-8?q?=20=E5=9B=9B=E4=B8=AA=E9=98=BB=E6=96=AD=E9=A1=B9=EF=BC=88B1-B4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - B1: 移除根目录 docker-compose.yml（个人镜像测试产物混入） - B2: 新增迁移 224 放宽 user_platform_quotas CHECK 至 8 平台，补 BulkInsertInitial 国产平台集成测试与迁移内容断言测试 - B3: CC/Responses×anthropic 四个上游读循环接入间隔超时泵 （gateway.stream_data_interval_timeout，默认 180s）：上游挂住 SSE 不发数据也不断连时结束排水/组装、关闭 resp.Body 归还连接池位， 排水超时仍按已累计 usage 返回（与 messages 主路径同语义） - B4: 配额/余额探测发起前过 cnValidateProbeURL（与网关转发同一套 security.url_allowlist 策略），被拒时零出站、API key 不离开本机 --- ...er_platform_quota_repo_integration_test.go | 29 ++ .../service/cn_provider_balance_service.go | 7 + .../internal/service/cn_provider_probe_url.go | 48 +++ .../service/cn_provider_probe_url_test.go | 129 ++++++++ .../service/cn_provider_quota_service.go | 8 + .../openai_gateway_anthropic_native_pump.go | 126 ++++++++ ...enai_gateway_anthropic_native_pump_test.go | 244 ++++++++++++++ ...teway_chat_completions_anthropic_native.go | 111 +++++-- ...enai_gateway_responses_anthropic_native.go | 106 ++++-- ..._user_platform_quotas_add_cn_providers.sql | 17 + ...tform_quota_cn_providers_migration_test.go | 22 ++ docker-compose.yml | 302 ------------------ 12 files changed, 795 insertions(+), 354 deletions(-) create mode 100644 backend/internal/service/cn_provider_probe_url.go create mode 100644 backend/internal/service/cn_provider_probe_url_test.go create mode 100644 backend/internal/service/openai_gateway_anthropic_native_pump.go create mode 100644 backend/internal/service/openai_gateway_anthropic_native_pump_test.go create mode 100644 backend/migrations/224_user_platform_quotas_add_cn_providers.sql create mode 100644 backend/migrations/user_platform_quota_cn_providers_migration_test.go delete mode 100644 docker-compose.yml diff --git a/backend/internal/repository/user_platform_quota_repo_integration_test.go b/backend/internal/repository/user_platform_quota_repo_integration_test.go index 4a8d1b5896..1bd4b08786 100644 --- a/backend/internal/repository/user_platform_quota_repo_integration_test.go +++ b/backend/internal/repository/user_platform_quota_repo_integration_test.go @@ -98,6 +98,35 @@ func TestUserPlatformQuotaRepository_BulkInsertInitial_GrokAllowed(t *testing.T) require.InDelta(t, 9.0, *rec.DailyLimitUSD, 1e-9) } +// TestUserPlatformQuotaRepository_BulkInsertInitial_CNProvidersAllowed 回归迁移 224： +// kimi/zhipu/deepseek 平台必须能写入 user_platform_quotas（CHECK 约束已含国产供应商）。 +// 历史 bug：三个平台不在约束内 → 注册预填充 8 平台默认配额时整条多行 INSERT 中止 → +// fail-open 吞错 → 新用户拿到零条配额记录（缺失配额行 = 无限额）。 +func TestUserPlatformQuotaRepository_BulkInsertInitial_CNProvidersAllowed(t *testing.T) { + ctx := context.Background() + tx := testEntTx(t) + txCtx := dbent.NewTxContext(ctx, tx) + client := tx.Client() + + userID := mustCreateUserForQuota(t, client) + repo := NewUserPlatformQuotaRepository(client) + + daily := 12.0 + records := []UserPlatformQuotaRecord{ + {UserID: userID, Platform: "kimi", DailyLimitUSD: &daily}, + {UserID: userID, Platform: "zhipu"}, + {UserID: userID, Platform: "deepseek"}, + } + require.NoError(t, repo.BulkInsertInitial(txCtx, records), + "kimi/zhipu/deepseek 平台应可写入（迁移 224 后 CHECK 约束已含国产供应商）") + + for _, platform := range []string{"kimi", "zhipu", "deepseek"} { + rec, err := repo.GetByUserPlatform(txCtx, userID, platform) + require.NoError(t, err) + require.NotNil(t, rec, "%s 配额行应已写入", platform) + } +} + func TestUserPlatformQuotaRepository_GetByUserPlatform(t *testing.T) { ctx := context.Background() tx := testEntTx(t) diff --git a/backend/internal/service/cn_provider_balance_service.go b/backend/internal/service/cn_provider_balance_service.go index 2f306e8b62..d01c9b6d08 100644 --- a/backend/internal/service/cn_provider_balance_service.go +++ b/backend/internal/service/cn_provider_balance_service.go @@ -125,6 +125,13 @@ func (s *CNProviderBalanceService) queryBalance(ctx context.Context, accountID i } targetURL := cnBalanceURL(account) + // 探测发起前过出站 URL 安全策略（与网关转发/Grok 探测同一套校验）： + // DeepSeek 端点由账号 base_url 衍生，不得把 API key 发往策略外主机。 + validatedURL, err := cnValidateProbeURL(s.cfg, targetURL) + if err != nil { + return nil, infraerrors.New(http.StatusForbidden, "CN_BALANCE_URL_REJECTED", err.Error()) + } + targetURL = validatedURL proxyURL := s.resolveProxyURL(ctx, account) callCtx, cancel := context.WithTimeout(ctx, cnBalanceUpstreamTimeout) defer cancel() diff --git a/backend/internal/service/cn_provider_probe_url.go b/backend/internal/service/cn_provider_probe_url.go new file mode 100644 index 0000000000..95b0ff783f --- /dev/null +++ b/backend/internal/service/cn_provider_probe_url.go @@ -0,0 +1,48 @@ +package service + +// CN 供应商探测端点的出站 URL 安全策略校验（配额/余额探测共用）。 +// +// 背景（review B4）：这两条探测路径会把账号 API key 发往 base_url 衍生端点， +// 此前完全绕过 security.url_allowlist——在加固部署里构成任意外发与内网探测 +// 面（本项目此前发生过账号测试 SSRF 生产事件）。与网关转发 +// （validateUpstreamBaseURL）、Grok 探测（grokOperatorPolicyValidator）一致， +// 探测发起前必须过同一套运营者策略。 + +import ( + "errors" + "fmt" + "strings" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/util/urlvalidator" +) + +// cnValidateProbeURL 按全局出站 URL 安全策略校验探测端点，返回规范化 URL。 +// 白名单开启时强制 UpstreamHosts（阻断私网与未列名主机）；关闭时仅做格式 +// 校验（HTTP 允许与否跟随配置）；cfg 为 nil 时退化为纯格式校验。 +func cnValidateProbeURL(cfg *config.Config, raw string) (string, error) { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return "", errors.New("probe url is required") + } + if cfg != nil && cfg.Security.URLAllowlist.Enabled { + normalized, err := urlvalidator.ValidateHTTPSURL(trimmed, urlvalidator.ValidationOptions{ + AllowedHosts: cfg.Security.URLAllowlist.UpstreamHosts, + RequireAllowlist: true, + AllowPrivate: cfg.Security.URLAllowlist.AllowPrivateHosts, + }) + if err != nil { + return "", fmt.Errorf("probe target rejected by URL security policy: %w", err) + } + return normalized, nil + } + var allowInsecureHTTP bool + if cfg != nil { + allowInsecureHTTP = cfg.Security.URLAllowlist.AllowInsecureHTTP + } + normalized, err := urlvalidator.ValidateURLFormat(trimmed, allowInsecureHTTP) + if err != nil { + return "", fmt.Errorf("probe target rejected by URL security policy: %w", err) + } + return normalized, nil +} diff --git a/backend/internal/service/cn_provider_probe_url_test.go b/backend/internal/service/cn_provider_probe_url_test.go new file mode 100644 index 0000000000..f285088400 --- /dev/null +++ b/backend/internal/service/cn_provider_probe_url_test.go @@ -0,0 +1,129 @@ +package service + +// CN 供应商探测端点 URL 安全策略回归测试（review B4）： +// 配额/余额探测不得绕过 security.url_allowlist——base_url 衍生的探测端点 +// 必须先过运营者策略，被拒绝时不得发起任何上游请求（API key 不出站）。 + +import ( + "context" + "net/http" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" +) + +func cnProbeAllowlistConfig(hosts ...string) *config.Config { + return &config.Config{ + Security: config.SecurityConfig{ + URLAllowlist: config.URLAllowlistConfig{ + Enabled: true, + UpstreamHosts: hosts, + }, + }, + } +} + +func TestCNValidateProbeURL_AllowlistPolicy(t *testing.T) { + cfg := cnProbeAllowlistConfig("api.moonshot.cn", "api.deepseek.com") + + // 白名单内主机放行（保留完整路径）。 + ok, err := cnValidateProbeURL(cfg, "https://api.moonshot.cn/v1/users/me/balance") + require.NoError(t, err) + require.Equal(t, "https://api.moonshot.cn/v1/users/me/balance", ok) + + // 白名单外主机拒绝。 + _, err = cnValidateProbeURL(cfg, "https://relay.attacker.example/v1/usages") + require.Error(t, err) + require.Contains(t, err.Error(), "rejected by URL security policy") + + // 私网主机拒绝（内网探测面）。 + _, err = cnValidateProbeURL(cfg, "http://169.254.169.254/latest/meta-data") + require.Error(t, err) + + // 白名单关闭：仅格式校验，任意 https 主机放行。 + formatOnly, err := cnValidateProbeURL(&config.Config{ + Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{Enabled: false}}, + }, "https://relay.attacker.example/v1/usages") + require.NoError(t, err) + require.Equal(t, "https://relay.attacker.example/v1/usages", formatOnly) +} + +// recordingHTTPUpstream 断言探测被策略拒绝时没有任何上游请求发出。 +type recordingHTTPUpstream struct{ calls int } + +func (u *recordingHTTPUpstream) Do(req *http.Request, proxyURL string, accountID int64, accountConcurrency int) (*http.Response, error) { + u.calls++ + return nil, context.DeadlineExceeded +} + +func (u *recordingHTTPUpstream) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, profile *tlsfingerprint.Profile) (*http.Response, error) { + u.calls++ + return nil, context.DeadlineExceeded +} + +type fakeCNProbeAccountRepo struct { + AccountRepository + account *Account +} + +func (r *fakeCNProbeAccountRepo) GetByID(ctx context.Context, id int64) (*Account, error) { + return r.account, nil +} + +// kimi coding 账号的 base_url 指向中转（含 api.kimi.com/coding 路径段即可被识别 +// 为 kimi coding plan）→ 衍生额度端点落在中转主机上，白名单未列名必须拒绝。 +func TestCNProviderQuotaService_RejectsURLBlockedByPolicy(t *testing.T) { + repo := &fakeCNProbeAccountRepo{account: &Account{ + ID: 1, Platform: PlatformKimi, Type: AccountTypeAPIKey, Status: StatusActive, + Credentials: map[string]any{ + "account_mode": "coding", + "api_key": "sk-test", + "base_url": "https://relay.attacker.example/api.kimi.com/coding", + }, + }} + upstream := &recordingHTTPUpstream{} + svc := NewCNProviderQuotaService(repo, nil, upstream, cnProbeAllowlistConfig("api.kimi.com")) + + _, err := svc.QueryUsage(context.Background(), 1) + require.Error(t, err) + require.Contains(t, err.Error(), "CN_QUOTA_URL_REJECTED") + require.Zero(t, upstream.calls, "probe must not issue any upstream request when URL policy rejects the target") +} + +// deepseek payg 账号自定义 base_url → 余额端点落在中转主机上，必须先过策略。 +func TestCNProviderBalanceService_RejectsURLBlockedByPolicy(t *testing.T) { + repo := &fakeCNProbeAccountRepo{account: &Account{ + ID: 2, Platform: PlatformDeepseek, Type: AccountTypeAPIKey, Status: StatusActive, + Credentials: map[string]any{ + "account_mode": "payg", + "api_key": "sk-test", + "base_url": "https://relay.attacker.example", + }, + }} + upstream := &recordingHTTPUpstream{} + svc := NewCNProviderBalanceService(repo, nil, upstream, cnProbeAllowlistConfig("api.deepseek.com")) + + _, err := svc.QueryBalance(context.Background(), 2) + require.Error(t, err) + require.Contains(t, err.Error(), "CN_BALANCE_URL_REJECTED") + require.Zero(t, upstream.calls, "probe must not issue any upstream request when URL policy rejects the target") +} + +// 白名单包含官方主机的正常路径：URL 校验通过后才发出上游请求（此处允许到达 +// httpUpstream 层即视为通过校验，不发真实网络）。 +func TestCNProviderBalanceService_OfficialHostPassesValidation(t *testing.T) { + repo := &fakeCNProbeAccountRepo{account: &Account{ + ID: 3, Platform: PlatformDeepseek, Type: AccountTypeAPIKey, Status: StatusActive, + Credentials: map[string]any{ + "account_mode": "payg", + "api_key": "sk-test", + }, + }} + upstream := &recordingHTTPUpstream{} + svc := NewCNProviderBalanceService(repo, nil, upstream, cnProbeAllowlistConfig("api.deepseek.com")) + + _, _ = svc.QueryBalance(context.Background(), 3) + require.Equal(t, 1, upstream.calls, "official host must pass URL policy and reach the upstream layer") +} diff --git a/backend/internal/service/cn_provider_quota_service.go b/backend/internal/service/cn_provider_quota_service.go index e8fdba9c56..f4cda8d6d1 100644 --- a/backend/internal/service/cn_provider_quota_service.go +++ b/backend/internal/service/cn_provider_quota_service.go @@ -144,6 +144,14 @@ func (s *CNProviderQuotaService) queryUsage(ctx context.Context, accountID int64 authHeader = apiKey // 智谱额度端点鉴权不加 Bearer 前缀 } + // 探测发起前过出站 URL 安全策略（与网关转发/Grok 探测同一套校验）： + // 端点多由账号 base_url 衍生，不得把 API key 发往策略外主机。 + validatedURL, err := cnValidateProbeURL(s.cfg, targetURL) + if err != nil { + return nil, infraerrors.New(http.StatusForbidden, "CN_QUOTA_URL_REJECTED", err.Error()) + } + targetURL = validatedURL + proxyURL := s.resolveProxyURL(ctx, account) callCtx, cancel := context.WithTimeout(ctx, cnQuotaUpstreamTimeout) defer cancel() diff --git a/backend/internal/service/openai_gateway_anthropic_native_pump.go b/backend/internal/service/openai_gateway_anthropic_native_pump.go new file mode 100644 index 0000000000..8a19677705 --- /dev/null +++ b/backend/internal/service/openai_gateway_anthropic_native_pump.go @@ -0,0 +1,126 @@ +package service + +// 国产供应商 Anthropic 协议转换路径的上游 SSE 行泵。 +// +// 这两条转换链（CC×anthropic / Responses×anthropic）的上游 ctx 是 +// WithoutCancel（detachStreamUpstreamContext）、http.Client 无整体 Timeout， +// 客户端断开后的排水阶段若上游挂住 SSE（不发数据也不断连）， +// scanner.Scan() 将永久阻塞：goroutine 钉死、resp.Body 不归还、连接池位 +// 被占用、usage 永不落库。 +// +// 与 handleAnthropicStreamingResponse / readOpenAICompatBufferedTerminal 的 +// 同类排水一致，本泵用 gateway.stream_data_interval_timeout（默认 180s）作为 +// 逐行读间隔上限，超时即向调用方返回 errAnthropicNativeStreamIdle，由调用方 +// 关闭 resp.Body 解除阻塞的读并结束排水。 + +import ( + "bufio" + "errors" + "io" + "time" +) + +// errAnthropicNativeStreamIdle 表示上游流读间隔超时（见上方文件注释）。 +var errAnthropicNativeStreamIdle = errors.New("stream data interval timeout") + +// anthropicNativeLineEvent 是行泵交付的单次读取结果：line 为一行 SSE 文本， +// err 为 scanner 读错误（流自然结束时 next 返回 io.EOF，不经过本字段）。 +type anthropicNativeLineEvent struct { + line string + err error +} + +// anthropicNativeLinePump 以独立 goroutine 泵送 scanner 的行，并对逐行到达 +// 间隔施加 interval 上限（<=0 表示禁用，保持无界读的旧行为）。 +type anthropicNativeLinePump struct { + events chan anthropicNativeLineEvent + done chan struct{} + timer *time.Timer + interval time.Duration +} + +// newAnthropicNativeLinePump 启动泵 goroutine；调用方 defer pump.stop()。 +func newAnthropicNativeLinePump(scanner *bufio.Scanner, interval time.Duration) *anthropicNativeLinePump { + p := &anthropicNativeLinePump{ + events: make(chan anthropicNativeLineEvent, 16), + done: make(chan struct{}), + interval: interval, + } + if interval > 0 { + p.timer = time.NewTimer(interval) + } + go func() { + defer close(p.events) + for scanner.Scan() { + select { + case p.events <- anthropicNativeLineEvent{line: scanner.Text()}: + case <-p.done: + return + } + } + if err := scanner.Err(); err != nil { + select { + case p.events <- anthropicNativeLineEvent{err: err}: + case <-p.done: + } + } + }() + return p +} + +// next 阻塞返回下一行。返回 io.EOF 表示上游正常收流；errAnthropicNativeStreamIdle +// 表示 interval 内无任何数据到达（计时从收到上一行时起算，事件处理耗时不算入， +// 与 readOpenAICompatBufferedTerminal 的 resetTimeout 语义一致）。 +func (p *anthropicNativeLinePump) next() (string, error) { + var timeoutCh <-chan time.Time + if p.timer != nil { + timeoutCh = p.timer.C + } + select { + case ev, ok := <-p.events: + if !ok { + return "", io.EOF + } + p.resetTimer() + return ev.line, ev.err + case <-timeoutCh: + return "", errAnthropicNativeStreamIdle + } +} + +// resetTimer 在收到一行后重启间隔计时器。 +func (p *anthropicNativeLinePump) resetTimer() { + if p.timer == nil { + return + } + if !p.timer.Stop() { + select { + case <-p.timer.C: + default: + } + } + p.timer.Reset(p.interval) +} + +// stop 终止泵 goroutine。注意：goroutine 若正阻塞在 scanner.Read 上，需由 +// 调用方关闭 resp.Body（间隔超时分支已做）才能真正退出。 +func (p *anthropicNativeLinePump) stop() { + close(p.done) + if p.timer != nil { + if !p.timer.Stop() { + select { + case <-p.timer.C: + default: + } + } + } +} + +// anthropicNativeStreamInterval 返回本组转换路径适用的读间隔上限； +// gateway.stream_data_interval_timeout <= 0 时视为禁用。 +func (s *OpenAIGatewayService) anthropicNativeStreamInterval() time.Duration { + if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 { + return time.Duration(s.cfg.Gateway.StreamDataIntervalTimeout) * time.Second + } + return 0 +} diff --git a/backend/internal/service/openai_gateway_anthropic_native_pump_test.go b/backend/internal/service/openai_gateway_anthropic_native_pump_test.go new file mode 100644 index 0000000000..1e9e1b39a6 --- /dev/null +++ b/backend/internal/service/openai_gateway_anthropic_native_pump_test.go @@ -0,0 +1,244 @@ +package service + +// 国产供应商 Anthropic 协议转换路径的上游读间隔超时回归测试（B3）： +// 上游挂住 SSE（不发数据也不断连）时，CC×anthropic / Responses×anthropic +// 的读循环必须按 gateway.stream_data_interval_timeout 结束，而不是永久阻塞。 + +import ( + "bufio" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" + "github.com/gin-gonic/gin" +) + +func newNativeAnthropicHangTestService(intervalSec int) *OpenAIGatewayService { + return &OpenAIGatewayService{ + cfg: &config.Config{ + Gateway: config.GatewayConfig{ + StreamDataIntervalTimeout: intervalSec, + MaxLineSize: defaultMaxLineSize, + }, + }, + } +} + +func newHangingUpstreamResponse() (*http.Response, *io.PipeReader, *io.PipeWriter) { + pr, pw := io.Pipe() + return &http.Response{StatusCode: http.StatusOK, Body: pr, Header: http.Header{}}, pr, pw +} + +// miniAnthropicSSEStream 是一段最小可转换的 Anthropic 事件流。 +func miniAnthropicSSEStream() string { + return strings.Join([]string{ + "event: message_start", + `data: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"glm-4.7","usage":{"input_tokens":10,"output_tokens":1}}}`, + "", + "event: content_block_start", + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + "", + "event: content_block_delta", + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}`, + "", + "event: content_block_stop", + `data: {"type":"content_block_stop","index":0}`, + "", + "event: message_delta", + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`, + "", + "event: message_stop", + `data: {"type":"message_stop"}`, + "", + "", + }, "\n") +} + +func TestAnthropicNativeLinePump_TimesOutWithoutData(t *testing.T) { + pr, _ := io.Pipe() + scanner := bufio.NewScanner(pr) + defer func() { _ = pr.Close() }() + + pump := newAnthropicNativeLinePump(scanner, 50*time.Millisecond) + defer pump.stop() + + start := time.Now() + _, err := pump.next() + if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") { + t.Fatalf("expected interval timeout, got %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("timeout not respected: %v", elapsed) + } +} + +func TestAnthropicNativeLinePump_DataResetsTimer(t *testing.T) { + pr, pw := io.Pipe() + scanner := bufio.NewScanner(pr) + pump := newAnthropicNativeLinePump(scanner, 1*time.Second) + defer pump.stop() + + go func() { + _, _ = pw.Write([]byte("event: ping\n")) + // 保持流打开且不再发数据：第二次 next 必须超时。 + time.Sleep(3 * time.Second) + _ = pw.Close() + }() + defer func() { _ = pr.Close() }() + + line, err := pump.next() + if err != nil || line != "event: ping" { + t.Fatalf("expected first line, got %q err=%v", line, err) + } + + start := time.Now() + _, err = pump.next() + if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") { + t.Fatalf("expected interval timeout after data stops, got %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("timeout not respected: %v", elapsed) + } +} + +func TestCCStreamingFromNativeAnthropic_HangTimesOut(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newNativeAnthropicHangTestService(1) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp, pr, pw := newHangingUpstreamResponse() + start := time.Now() + res, err := svc.handleCCStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start, true) + _ = pw.Close() + _ = pr.Close() + + if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") { + t.Fatalf("expected stream timeout error, got %v", err) + } + if res == nil { + t.Fatalf("expected result carrying accumulated usage") + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("handler did not respect interval bound: %v", elapsed) + } +} + +func TestCCBufferedFromNativeAnthropic_HangTimesOut(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newNativeAnthropicHangTestService(1) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp, pr, pw := newHangingUpstreamResponse() + start := time.Now() + _, err := svc.handleCCBufferedFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start) + _ = pw.Close() + _ = pr.Close() + + if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") { + t.Fatalf("expected stream timeout error, got %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("handler did not respect interval bound: %v", elapsed) + } + if !strings.Contains(rec.Body.String(), "Upstream stream data interval timeout") { + t.Fatalf("expected 502 error body, got %q", rec.Body.String()) + } +} + +func TestResponsesStreamingFromNativeAnthropic_HangTimesOut(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newNativeAnthropicHangTestService(1) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp, pr, pw := newHangingUpstreamResponse() + start := time.Now() + res, err := svc.handleResponsesStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start, apicompat.ResponsesClientToolMapping{}) + _ = pw.Close() + _ = pr.Close() + + if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") { + t.Fatalf("expected stream timeout error, got %v", err) + } + if res == nil { + t.Fatalf("expected result carrying accumulated usage") + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("handler did not respect interval bound: %v", elapsed) + } +} + +func TestCCStreamingFromNativeAnthropic_HappyPathStillConverts(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newNativeAnthropicHangTestService(5) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp, pr, pw := newHangingUpstreamResponse() + go func() { + _, _ = pw.Write([]byte(miniAnthropicSSEStream())) + _ = pw.Close() + }() + defer func() { _ = pr.Close() }() + + res, err := svc.handleCCStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, time.Now(), true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if res == nil { + t.Fatalf("expected result") + } + body := rec.Body.String() + if !strings.Contains(body, "Hello") { + t.Fatalf("expected converted text chunk, got %q", body) + } + if !strings.Contains(body, "data: [DONE]") { + t.Fatalf("expected [DONE] terminator, got %q", body) + } +} + +func TestCCBufferedFromNativeAnthropic_HappyPathStillConverts(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newNativeAnthropicHangTestService(5) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp, pr, pw := newHangingUpstreamResponse() + go func() { + _, _ = pw.Write([]byte(miniAnthropicSSEStream())) + _ = pw.Close() + }() + defer func() { _ = pr.Close() }() + + res, err := svc.handleCCBufferedFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, time.Now()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if res == nil { + t.Fatalf("expected result") + } + body := rec.Body.String() + if !strings.Contains(body, "Hello") { + t.Fatalf("expected converted text in buffered response, got %q", body) + } + if res.Usage.InputTokens != 10 || res.Usage.OutputTokens != 5 { + t.Fatalf("expected usage 10/5, got %+v", res.Usage) + } +} diff --git a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go index 0ef89a7ed0..b41bc644c6 100644 --- a/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go +++ b/backend/internal/service/openai_gateway_chat_completions_anthropic_native.go @@ -14,6 +14,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net/http" "strings" "time" @@ -163,18 +164,53 @@ func (s *OpenAIGatewayService) handleCCBufferedFromNativeAnthropic( var finalResp *apicompat.AnthropicResponse var usage ClaudeUsage - for scanner.Scan() { - line := scanner.Text() + // 读间隔上限：上游挂住 SSE 时中止组装（缓冲路径尚未提交响应头，可回 502）。 + streamInterval := s.anthropicNativeStreamInterval() + pump := newAnthropicNativeLinePump(scanner, streamInterval) + defer pump.stop() + + logReadErr := func(err error) { + if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + logger.L().Warn("openai cc via native anthropic buffered: read error", + zap.Error(err), + zap.String("request_id", requestID), + ) + } + } + onIdle := func() (*OpenAIForwardResult, error) { + _ = resp.Body.Close() + logger.L().Warn("openai cc via native anthropic buffered: data interval timeout", + zap.String("request_id", requestID), + zap.Duration("interval", streamInterval), + ) + writeChatCompletionsError(c, http.StatusBadGateway, "server_error", "Upstream stream data interval timeout") + return nil, fmt.Errorf("stream data interval timeout") + } + + for { + line, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) + break + } // SSE 规范允许 `event:xxx`（冒号后无空格）：Kimi 等 Anthropic 兼容上游 // 返回紧凑格式，严格匹配 "event: " 会丢弃全部事件（#4653 同根因）。 if _, ok := extractOpenAISSEEventLine(line); !ok { continue } - if !scanner.Scan() { + dataLine, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) break } - payload, ok := extractOpenAISSEDataLine(scanner.Text()) + payload, ok := extractOpenAISSEDataLine(dataLine) if !ok { continue } @@ -214,15 +250,6 @@ func (s *OpenAIGatewayService) handleCCBufferedFromNativeAnthropic( } } - if err := scanner.Err(); err != nil { - if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai cc via native anthropic buffered: read error", - zap.Error(err), - zap.String("request_id", requestID), - ) - } - } - if finalResp == nil { writeChatCompletionsError(c, http.StatusBadGateway, "server_error", "Upstream stream ended without a response") return nil, fmt.Errorf("upstream stream ended without response") @@ -322,6 +349,34 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic( } } + // 读间隔上限：上游挂住 SSE（不发数据也不断连）时结束排水。上游 ctx 为 + // WithoutCancel 且 http.Client 无整体 Timeout，无此界限则客户端断开后 + // scanner.Scan() 永久阻塞（见 anthropic native pump 文件注释）。 + streamInterval := s.anthropicNativeStreamInterval() + pump := newAnthropicNativeLinePump(scanner, streamInterval) + defer pump.stop() + + logReadErr := func(err error) { + if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + logger.L().Warn("openai cc via native anthropic stream: read error", + zap.Error(err), + zap.String("request_id", requestID), + ) + } + } + // onIdle 关闭上游连接（解除阻塞的读、归还连接池位），并按已累计 usage + // 返回——与 messages 主路径 "stream usage incomplete after timeout" 同语义。 + onIdle := func() (*OpenAIForwardResult, error) { + _ = resp.Body.Close() + if !clientDisconnected { + logger.L().Warn("openai cc via native anthropic stream: data interval timeout", + zap.String("request_id", requestID), + zap.Duration("interval", streamInterval), + ) + } + return resultWithUsage(), fmt.Errorf("stream data interval timeout") + } + writeChunk := func(chunk apicompat.ChatCompletionsChunk) bool { if clientDisconnected { return false // 已断开：不再写客户端，只排水上游累计 usage @@ -372,16 +427,29 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic( return false } - for scanner.Scan() { - line := scanner.Text() + for { + line, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) + break + } if _, ok := extractOpenAISSEEventLine(line); !ok { continue } - if !scanner.Scan() { + dataLine, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + // EOF / 读错误：事件行后流终止，进入 finalize。 + logReadErr(rerr) break } - payload, ok := extractOpenAISSEDataLine(scanner.Text()) + payload, ok := extractOpenAISSEDataLine(dataLine) if !ok { continue } @@ -396,15 +464,6 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic( } } - if err := scanner.Err(); err != nil { - if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai cc via native anthropic stream: read error", - zap.Error(err), - zap.String("request_id", requestID), - ) - } - } - // Finalize both state machines（客户端已断开时仍执行，保证 usage 汇总完整）。 finalResEvents := apicompat.FinalizeAnthropicResponsesStream(anthState) for _, resEvt := range finalResEvents { diff --git a/backend/internal/service/openai_gateway_responses_anthropic_native.go b/backend/internal/service/openai_gateway_responses_anthropic_native.go index 7cb4a13d2d..b72067fc4d 100644 --- a/backend/internal/service/openai_gateway_responses_anthropic_native.go +++ b/backend/internal/service/openai_gateway_responses_anthropic_native.go @@ -14,6 +14,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net/http" "strings" "time" @@ -166,17 +167,52 @@ func (s *OpenAIGatewayService) handleResponsesBufferedFromNativeAnthropic( var finalResp *apicompat.AnthropicResponse var usage ClaudeUsage - for scanner.Scan() { - line := scanner.Text() + // 读间隔上限：上游挂住 SSE 时中止组装（缓冲路径尚未提交响应头，可回 502）。 + streamInterval := s.anthropicNativeStreamInterval() + pump := newAnthropicNativeLinePump(scanner, streamInterval) + defer pump.stop() + + logReadErr := func(err error) { + if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + logger.L().Warn("openai responses via native anthropic buffered: read error", + zap.Error(err), + zap.String("request_id", requestID), + ) + } + } + onIdle := func() (*OpenAIForwardResult, error) { + _ = resp.Body.Close() + logger.L().Warn("openai responses via native anthropic buffered: data interval timeout", + zap.String("request_id", requestID), + zap.Duration("interval", streamInterval), + ) + writeResponsesError(c, http.StatusBadGateway, "server_error", "Upstream stream data interval timeout") + return nil, fmt.Errorf("stream data interval timeout") + } + + for { + line, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) + break + } // SSE 规范允许 `event:xxx`（冒号后无空格）：Kimi 等上游返回紧凑格式。 if _, ok := extractOpenAISSEEventLine(line); !ok { continue } - if !scanner.Scan() { + dataLine, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) break } - payload, ok := extractOpenAISSEDataLine(scanner.Text()) + payload, ok := extractOpenAISSEDataLine(dataLine) if !ok { continue } @@ -216,15 +252,6 @@ func (s *OpenAIGatewayService) handleResponsesBufferedFromNativeAnthropic( } } - if err := scanner.Err(); err != nil { - if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai responses via native anthropic buffered: read error", - zap.Error(err), - zap.String("request_id", requestID), - ) - } - } - if finalResp == nil { writeResponsesError(c, http.StatusBadGateway, "server_error", "Upstream stream ended without a response") return nil, fmt.Errorf("upstream stream ended without response") @@ -326,6 +353,30 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic( } } + // 读间隔上限：上游挂住 SSE（不发数据也不断连）时结束转换循环。上游 ctx 为 + // WithoutCancel 且 http.Client 无整体 Timeout，无此界限则 scanner.Scan() + // 永久阻塞（见 anthropic native pump 文件注释）。 + streamInterval := s.anthropicNativeStreamInterval() + pump := newAnthropicNativeLinePump(scanner, streamInterval) + defer pump.stop() + + logReadErr := func(err error) { + if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { + logger.L().Warn("openai responses via native anthropic stream: read error", + zap.Error(err), + zap.String("request_id", requestID), + ) + } + } + onIdle := func() (*OpenAIForwardResult, error) { + _ = resp.Body.Close() + logger.L().Warn("openai responses via native anthropic stream: data interval timeout", + zap.String("request_id", requestID), + zap.Duration("interval", streamInterval), + ) + return resultWithUsage(), fmt.Errorf("stream data interval timeout") + } + processAnthropicEvent := func(event *apicompat.AnthropicStreamEvent) bool { if firstChunk { firstChunk = false @@ -365,16 +416,28 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic( return false } - for scanner.Scan() { - line := scanner.Text() + for { + line, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) + break + } if _, ok := extractOpenAISSEEventLine(line); !ok { continue } - if !scanner.Scan() { + dataLine, rerr := pump.next() + if rerr != nil { + if errors.Is(rerr, errAnthropicNativeStreamIdle) { + return onIdle() + } + logReadErr(rerr) break } - payload, ok := extractOpenAISSEDataLine(scanner.Text()) + payload, ok := extractOpenAISSEDataLine(dataLine) if !ok { continue } @@ -389,15 +452,6 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic( } } - if err := scanner.Err(); err != nil { - if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai responses via native anthropic stream: read error", - zap.Error(err), - zap.String("request_id", requestID), - ) - } - } - // Finalize state machine（客户端已断开时仍执行，保证 usage 汇总完整）。 if finalEvents := apicompat.FinalizeAnthropicResponsesStream(state); len(finalEvents) > 0 { for _, evt := range finalEvents { diff --git a/backend/migrations/224_user_platform_quotas_add_cn_providers.sql b/backend/migrations/224_user_platform_quotas_add_cn_providers.sql new file mode 100644 index 0000000000..011a7762af --- /dev/null +++ b/backend/migrations/224_user_platform_quotas_add_cn_providers.sql @@ -0,0 +1,17 @@ +-- 把 kimi/zhipu/deepseek 平台加入 user_platform_quotas.platform 的 CHECK 约束。 +-- +-- 背景：国产供应商进入 AllowedQuotaPlatforms（internal/service/domain_constants.go）， +-- 注册时 GetDefaultPlatformQuotas 会为全部 8 平台预填充默认配额行，但 157 号迁移的 +-- CHECK 仍只允许 5 平台。BulkInsertInitial 是单条多行 INSERT，任一违约行会中止整条 +-- 语句 → 注册路径 fail-open 吞错 → 新用户拿到零条配额记录（含原有 5 平台，缺失配额 +-- 行 = 无限额）。与 157 头注释记载的 grok 同型事故一致。 +-- +-- 修复：把约束与代码平台列表（PlatformKimi/PlatformZhipu/PlatformDeepseek）对齐。 +-- DROP ... IF EXISTS 保证可重入；新约束是旧约束的超集，存量行（仅 5 平台）瞬时校验通过。 +ALTER TABLE user_platform_quotas + DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check; + +ALTER TABLE user_platform_quotas + ADD CONSTRAINT user_platform_quotas_platform_check + CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', + 'kimi', 'zhipu', 'deepseek')); diff --git a/backend/migrations/user_platform_quota_cn_providers_migration_test.go b/backend/migrations/user_platform_quota_cn_providers_migration_test.go new file mode 100644 index 0000000000..cce672aaaf --- /dev/null +++ b/backend/migrations/user_platform_quota_cn_providers_migration_test.go @@ -0,0 +1,22 @@ +package migrations + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestUserPlatformQuotasCNProvidersMigration 校验 224 号迁移把 kimi/zhipu/deepseek +// 加入 user_platform_quotas.platform 的 CHECK 约束（对照 157 号 grok 迁移）。 +// 约束未放宽时，注册预填充 8 平台默认配额会整条 INSERT 中止 → 新用户零配额行 +//（缺失配额行 = 无限额），管理端设置国产平台配额直接 500。 +func TestUserPlatformQuotasCNProvidersMigration(t *testing.T) { + content, err := FS.ReadFile("224_user_platform_quotas_add_cn_providers.sql") + require.NoError(t, err) + + sql := strings.Join(strings.Fields(string(content)), " ") + require.Contains(t, sql, "DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check") + require.Contains(t, sql, + "CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', 'kimi', 'zhipu', 'deepseek'))") +} diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100644 index 67edd44e22..0000000000 --- a/docker-compose.yml +++ /dev/null @@ -1,302 +0,0 @@ -# ============================================================================= -# Sub2API Docker Compose Configuration -# ============================================================================= -# Quick Start: -# 1. Copy .env.example to .env and configure -# 2. docker-compose up -d -# 3. Check logs: docker-compose logs -f sub2api -# 4. Access: http://localhost:8080 -# -# All configuration is done via environment variables. -# No Setup Wizard needed - the system auto-initializes on first run. -# ============================================================================= - -services: - # =========================================================================== - # Sub2API Application - # =========================================================================== - sub2api: - image: randark/sub2api:latest - container_name: sub2api - restart: unless-stopped - security_opt: - - no-new-privileges:true - ulimits: - nofile: - soft: 100000 - hard: 100000 - ports: - - "${BIND_HOST:-0.0.0.0}:${SERVER_PORT:-8080}:8080" - volumes: - # Data persistence (config.yaml will be auto-generated here) - - sub2api_data:/app/data - # Optional: Mount custom config.yaml (uncomment and create the file first) - # Copy config.example.yaml to config.yaml, modify it, then uncomment: - # - ./config.yaml:/app/data/config.yaml - # Optional: Mount a custom Codex instructions template file, then point - # gateway.forced_codex_instructions_template_file at /app/data/codex-instructions.md.tmpl - # in config.yaml. - # - ./codex-instructions.md.tmpl:/app/data/codex-instructions.md.tmpl:ro - environment: - # ======================================================================= - # Auto Setup (REQUIRED for Docker deployment) - # ======================================================================= - - AUTO_SETUP=true - - # ======================================================================= - # Server Configuration - # ======================================================================= - - SERVER_HOST=0.0.0.0 - - SERVER_PORT=8080 - - SERVER_MODE=${SERVER_MODE:-release} - - ENABLE_SERVER_TIMING=${ENABLE_SERVER_TIMING:-false} - - RUN_MODE=${RUN_MODE:-standard} - - UPDATE_GITHUB_TOKEN=${UPDATE_GITHUB_TOKEN:-} - - ALIPAY_MOBILE_PRECREATE_DEEP_LINK=${ALIPAY_MOBILE_PRECREATE_DEEP_LINK:-} - - # ======================================================================= - # Database Configuration (PostgreSQL) - # ======================================================================= - - DATABASE_HOST=postgres - - DATABASE_PORT=5432 - - DATABASE_USER=${POSTGRES_USER:-sub2api} - - DATABASE_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required} - - DATABASE_DBNAME=${POSTGRES_DB:-sub2api} - - DATABASE_SSLMODE=disable - - DATABASE_MAX_OPEN_CONNS=${DATABASE_MAX_OPEN_CONNS:-50} - - DATABASE_MAX_IDLE_CONNS=${DATABASE_MAX_IDLE_CONNS:-10} - - DATABASE_CONN_MAX_LIFETIME_MINUTES=${DATABASE_CONN_MAX_LIFETIME_MINUTES:-30} - - DATABASE_CONN_MAX_IDLE_TIME_MINUTES=${DATABASE_CONN_MAX_IDLE_TIME_MINUTES:-5} - - # ======================================================================= - # Redis Configuration - # ======================================================================= - - REDIS_HOST=redis - - REDIS_PORT=6379 - - REDIS_USERNAME=${REDIS_USERNAME:-} - - REDIS_PASSWORD=${REDIS_PASSWORD:-} - - REDIS_DB=${REDIS_DB:-0} - - REDIS_POOL_SIZE=${REDIS_POOL_SIZE:-1024} - - REDIS_MIN_IDLE_CONNS=${REDIS_MIN_IDLE_CONNS:-10} - - REDIS_ENABLE_TLS=${REDIS_ENABLE_TLS:-false} - - # ======================================================================= - # Admin Account (auto-created on first run) - # ======================================================================= - - ADMIN_EMAIL=${ADMIN_EMAIL:-admin@sub2api.local} - - ADMIN_PASSWORD=${ADMIN_PASSWORD:-} - - # ======================================================================= - # JWT Configuration - # ======================================================================= - # IMPORTANT: Set a fixed JWT_SECRET to prevent login sessions from being - # invalidated after container restarts. If left empty, a random secret - # will be generated on each startup. - # Generate a secure secret: openssl rand -hex 32 - - JWT_SECRET=${JWT_SECRET:-} - - JWT_EXPIRE_HOUR=${JWT_EXPIRE_HOUR:-24} - - # ======================================================================= - # Setup Configuration - # ======================================================================= - - SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0} - - # ======================================================================= - # TOTP (2FA) Configuration - # ======================================================================= - # IMPORTANT: Set a fixed encryption key for TOTP secrets. If left empty, - # a random key will be generated on each startup, causing all existing - # TOTP configurations to become invalid (users won't be able to login - # with 2FA). - # Generate a secure key: openssl rand -hex 32 - - TOTP_ENCRYPTION_KEY=${TOTP_ENCRYPTION_KEY:-} - - # ======================================================================= - # Timezone Configuration - # This affects ALL time operations in the application: - # - Database timestamps - # - Usage statistics "today" boundary - # - Subscription expiry times - # - Log timestamps - # Common values: Asia/Shanghai, America/New_York, Europe/London, UTC - # ======================================================================= - - TZ=${TZ:-Asia/Shanghai} - - # ======================================================================= - # Gemini OAuth Configuration (for Gemini accounts) - # ======================================================================= - - GEMINI_OAUTH_CLIENT_ID=${GEMINI_OAUTH_CLIENT_ID:-} - - GEMINI_OAUTH_CLIENT_SECRET=${GEMINI_OAUTH_CLIENT_SECRET:-} - - GEMINI_OAUTH_SCOPES=${GEMINI_OAUTH_SCOPES:-} - - GEMINI_QUOTA_POLICY=${GEMINI_QUOTA_POLICY:-} - - # Built-in OAuth client secrets (optional) - # SECURITY: This repo does not embed third-party client_secret. - - GEMINI_CLI_OAUTH_CLIENT_SECRET=${GEMINI_CLI_OAUTH_CLIENT_SECRET:-} - - ANTIGRAVITY_OAUTH_CLIENT_SECRET=${ANTIGRAVITY_OAUTH_CLIENT_SECRET:-} - - ANTIGRAVITY_USER_AGENT_VERSION=${ANTIGRAVITY_USER_AGENT_VERSION:-} - - # ======================================================================= - # Security Configuration (URL Allowlist) - # ======================================================================= - # Enable URL allowlist validation (false to skip allowlist checks) - - SECURITY_URL_ALLOWLIST_ENABLED=${SECURITY_URL_ALLOWLIST_ENABLED:-false} - # Allow insecure HTTP URLs when allowlist is disabled (default: true; set to false to require https) - - SECURITY_URL_ALLOWLIST_ALLOW_INSECURE_HTTP=${SECURITY_URL_ALLOWLIST_ALLOW_INSECURE_HTTP:-true} - # Allow private IP addresses for upstream/pricing/CRS (default: true; set to false to block private hosts) - - SECURITY_URL_ALLOWLIST_ALLOW_PRIVATE_HOSTS=${SECURITY_URL_ALLOWLIST_ALLOW_PRIVATE_HOSTS:-true} - # Upstream hosts whitelist (comma-separated, only used when enabled=true) - - SECURITY_URL_ALLOWLIST_UPSTREAM_HOSTS=${SECURITY_URL_ALLOWLIST_UPSTREAM_HOSTS:-} - - # ======================================================================= - # Update Configuration (在线更新配置) - # ======================================================================= - # Proxy for accessing GitHub (online updates + pricing data) - # Examples: http://host:port, socks5://host:port - - UPDATE_PROXY_URL=${UPDATE_PROXY_URL:-} - - # ======================================================================= - # Image Generation Stream & Concurrency - # ======================================================================= - # OpenAI HTTP upstream protocol/timeout - - GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT=${GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT:-0} - - GATEWAY_OPENAI_HTTP2_ENABLED=${GATEWAY_OPENAI_HTTP2_ENABLED:-true} - - GATEWAY_OPENAI_HTTP2_ALLOW_PROXY_FALLBACK_TO_HTTP1=${GATEWAY_OPENAI_HTTP2_ALLOW_PROXY_FALLBACK_TO_HTTP1:-true} - - GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2} - - GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60} - - GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600} - - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2} - - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60} - - GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600} - - GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900} - - GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10} - - GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false} - - GATEWAY_IMAGE_CONCURRENCY_MAX_CONCURRENT_REQUESTS=${GATEWAY_IMAGE_CONCURRENCY_MAX_CONCURRENT_REQUESTS:-0} - - GATEWAY_IMAGE_CONCURRENCY_OVERFLOW_MODE=${GATEWAY_IMAGE_CONCURRENCY_OVERFLOW_MODE:-reject} - - GATEWAY_IMAGE_CONCURRENCY_WAIT_TIMEOUT_SECONDS=${GATEWAY_IMAGE_CONCURRENCY_WAIT_TIMEOUT_SECONDS:-30} - - GATEWAY_IMAGE_CONCURRENCY_MAX_WAITING_REQUESTS=${GATEWAY_IMAGE_CONCURRENCY_MAX_WAITING_REQUESTS:-100} - depends_on: - postgres: - condition: service_healthy - redis: - condition: service_healthy - networks: - - sub2api-network - healthcheck: - test: - [ - "CMD", - "wget", - "-q", - "-T", - "5", - "-O", - "/dev/null", - "http://localhost:8080/health", - ] - interval: 30s - timeout: 10s - retries: 3 - start_period: 30s - - # =========================================================================== - # PostgreSQL Database - # =========================================================================== - postgres: - image: postgres:18-alpine - container_name: sub2api-postgres - restart: unless-stopped - ulimits: - nofile: - soft: 100000 - hard: 100000 - # Pass the tuning values documented in .env.example to postgres. The - # fallbacks are the postgres:18 stock defaults, so nothing changes when - # the variables are not set. - command: > - postgres - -c max_connections=${POSTGRES_MAX_CONNECTIONS:-100} - -c shared_buffers=${POSTGRES_SHARED_BUFFERS:-128MB} - -c effective_cache_size=${POSTGRES_EFFECTIVE_CACHE_SIZE:-4GB} - -c maintenance_work_mem=${POSTGRES_MAINTENANCE_WORK_MEM:-64MB} - volumes: - - postgres_data:/var/lib/postgresql/data - environment: - # postgres:18-alpine 默认 PGDATA=/var/lib/postgresql/18/docker（位于镜像声明的匿名卷 /var/lib/postgresql 内）。 - # 若不显式设置 PGDATA，则即使挂载了 postgres_data 到 /var/lib/postgresql/data，数据也不会落盘到该命名卷， - # docker compose down/up 后会触发 initdb 重新初始化，导致用户/密码等数据丢失。 - - PGDATA=/var/lib/postgresql/data - - POSTGRES_USER=${POSTGRES_USER:-sub2api} - - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required} - - POSTGRES_DB=${POSTGRES_DB:-sub2api} - - TZ=${TZ:-Asia/Shanghai} - networks: - - sub2api-network - healthcheck: - test: - [ - "CMD-SHELL", - "pg_isready -U ${POSTGRES_USER:-sub2api} -d ${POSTGRES_DB:-sub2api}", - ] - interval: 10s - timeout: 5s - retries: 5 - start_period: 10s - # 注意：不暴露端口到宿主机，应用通过内部网络连接 - # 如需调试，可临时添加：ports: ["127.0.0.1:5433:5432"] - - # =========================================================================== - # Redis Cache - # =========================================================================== - redis: - image: redis:8-alpine - container_name: sub2api-redis - restart: unless-stopped - ulimits: - nofile: - soft: 100000 - hard: 100000 - volumes: - - redis_data:/data - # The command is one quoted script for the inner `sh -c`. Compose keeps - # the newlines inside the quoted string, so every line needs a trailing - # `\` — without it, `redis-server` on the first line runs with no flags - # at all, and the --save/--appendonly/--appendfsync lines are never read. - command: > - sh -c ' - redis-server \ - --save 60 1 \ - --appendonly yes \ - --appendfsync everysec \ - ${REDIS_PASSWORD:+--requirepass "$REDIS_PASSWORD"}' - environment: - - TZ=${TZ:-Asia/Shanghai} - # REDISCLI_AUTH is used by redis-cli for authentication (safer than -a flag) - - REDISCLI_AUTH=${REDIS_PASSWORD:-} - networks: - - sub2api-network - healthcheck: - test: ["CMD", "redis-cli", "ping"] - interval: 10s - timeout: 5s - retries: 5 - start_period: 5s - # 注意：不暴露端口到宿主机，应用通过内部网络连接 - # 如需调试，可临时添加：ports: ["127.0.0.1:6379:6379"] -# ============================================================================= -# Volumes -# ============================================================================= -volumes: - sub2api_data: - driver: local - postgres_data: - driver: local - redis_data: - driver: local - -# ============================================================================= -# Networks -# ============================================================================= -networks: - sub2api-network: - driver: bridge