fix(openai): HTTP 转发的执行作用域改从原始请求计算

This commit is contained in:
xlplbo
2026-09-11 11:36:29 +08:00
parent d0ca057ca4
commit 613722eee4
4 changed files with 106 additions and 24 deletions
@@ -42,6 +42,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
restrictionResult := s.detectCodexClientRestriction(c, account, body)
apiKeyID := getAPIKeyIDFromContext(c)
// 执行作用域必须取自客户端原始身份:后面的账号 namespace 改写与指纹收敛会改掉
// 请求体里的 client_metadata / prompt_cache_key,用改写后的值取键会让不同会话
// 落到同一个键,也会与 WS 接入路径按原始报文算出的键对不上。
wsExecutionScope, _ := resolveOpenAIWSExecutionScope(c, body, apiKeyID)
logCodexCLIOnlyDetection(ctx, c, account, apiKeyID, restrictionResult, body)
if restrictionResult.Enabled && !restrictionResult.Matched {
MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalPolicyDenied)
@@ -874,6 +878,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
account,
wsReqBody,
clientPromptCacheKey,
wsExecutionScope,
token,
wsDecision,
isCodexCLI,
@@ -40,28 +40,6 @@ func resolveOpenAIWSClientThreadID(c *gin.Context, body []byte) string {
return codexTurnMetadataThreadID(gjson.GetBytes(body, "client_metadata."+openAIWSTurnMetadataHeader).String())
}
// openAIWSExecutionScopeBodyFromRequest 只序列化执行作用域会读取的字段,
// 供已解析成 map 的 HTTP 请求体取键,避免为此重新编码整个请求体。
func openAIWSExecutionScopeBodyFromRequest(reqBody map[string]any) []byte {
if len(reqBody) == 0 {
return nil
}
subset := make(map[string]any, 3)
for _, key := range []string{"client_metadata", "prompt_cache_key", "previous_response_id"} {
if value, ok := reqBody[key]; ok {
subset[key] = value
}
}
if len(subset) == 0 {
return nil
}
raw, err := json.Marshal(subset)
if err != nil {
return nil
}
return raw
}
func codexTurnMetadataThreadID(raw string) string {
raw = strings.TrimSpace(raw)
if raw == "" {
@@ -23,6 +23,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
account *Account,
reqBody map[string]any,
clientPromptCacheKey string,
executionScope string,
token string,
decision OpenAIWSProtocolDecision,
isCodexCLI bool,
@@ -126,8 +127,9 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
}
// 与 WS 接入路径共用执行作用域:codex 多智能体共用 session-id,turn state 与
// store=false 的连接绑定必须按线程隔离,同一线程在两条路径之间也才能共享状态。
if scope, _ := resolveOpenAIWSExecutionScope(c, openAIWSExecutionScopeBodyFromRequest(reqBody), getAPIKeyIDFromContext(c)); scope != "" {
sessionHash = scope
// 作用域由 Forward 从改写前的原始请求算出后传入,reqBody 此时已带账号 namespace。
if executionScope = strings.TrimSpace(executionScope); executionScope != "" {
sessionHash = executionScope
}
if turnState == "" && stateStore != nil && sessionHash != "" {
if savedTurnState, ok := stateStore.GetSessionTurnState(groupID, sessionHash); ok {
@@ -2,6 +2,7 @@ package service
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"testing"
@@ -102,3 +103,99 @@ func TestOpenAIGatewayService_Forward_WSv2_TurnStateBoundToExecutionScope(t *tes
require.True(t, turnStateOnScope, "turn state 应绑定到执行作用域")
require.Equal(t, "turn-state-from-upstream", turnState)
}
// Forward 在转发前会按账号 namespace 改写请求体里的 client_metadata,指纹 full 模式还会给每个
// 请求注入同一个 thread_id。执行作用域必须取自客户端原始请求:否则同一 API key 下不同
// session_id 的会话会落到同一个键共用 turn state,客户端自带线程标识时也会与 WS 接入路径
// 按原始报文算出的键对不上。
func TestOpenAIGatewayService_Forward_WSv2_ExecutionScopeUsesOriginalIdentity(t *testing.T) {
gin.SetMode(gin.TestMode)
cfg := &config.Config{}
cfg.Security.URLAllowlist.Enabled = false
cfg.Security.URLAllowlist.AllowInsecureHTTP = true
cfg.Gateway.OpenAIWS.Enabled = true
cfg.Gateway.OpenAIWS.OAuthEnabled = true
cfg.Gateway.OpenAIWS.APIKeyEnabled = true
cfg.Gateway.OpenAIWS.ResponsesWebsocketsV2 = true
cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1
cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0
cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1
completed := func(id string) []byte {
return []byte(`{"type":"response.completed","response":{"id":"` + id + `","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`)
}
captureConn := &openAIWSCaptureConn{events: [][]byte{completed("resp_a"), completed("resp_b"), completed("resp_c")}}
handshake := http.Header{}
handshake.Set(openAIWSTurnStateHeader, "turn-state-from-upstream")
pool := newOpenAIWSConnPool(cfg)
pool.setClientDialerForTest(&openAIWSCaptureDialer{conn: captureConn, handshake: handshake})
defer pool.Close()
stateStore := NewOpenAIWSStateStore(nil)
svc := &OpenAIGatewayService{
cfg: cfg,
httpUpstream: &httpUpstreamRecorder{},
cache: &stubGatewayCache{},
openaiWSResolver: NewOpenAIWSProtocolResolver(cfg),
toolCorrector: NewCodexToolCorrector(),
openaiWSPool: pool,
openaiWSStateStore: stateStore,
}
groupID := int64(9)
const apiKeyID = int64(21)
account := &Account{
ID: 457,
Name: "openai-oauth-fingerprint-full",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Credentials: map[string]any{"access_token": "oauth-token-1"},
Extra: map[string]any{
"responses_websockets_v2_enabled": true,
codexFingerprintModeExtraKey: "full",
codexFingerprintSeedExtraKey: "11111111-1111-4111-8111-aaaaaaaaaaaa",
},
}
forward := func(sessionID, body string) (*gin.Context, []byte) {
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", nil)
c.Request.Header.Set("session_id", sessionID)
c.Set("api_key", &APIKey{ID: apiKeyID, GroupID: &groupID})
raw := []byte(body)
result, err := svc.Forward(context.Background(), c, account, raw)
require.NoError(t, err)
require.NotNil(t, result)
return c, raw
}
plainBody := `{"model":"gpt-5.1","stream":false,"input":[{"type":"input_text","text":"hello"}]}`
cA, rawA := forward("session-a", plainBody)
cB, rawB := forward("session-b", plainBody)
scopeA, _ := resolveOpenAIWSExecutionScope(cA, rawA, apiKeyID)
scopeB, _ := resolveOpenAIWSExecutionScope(cB, rawB, apiKeyID)
require.NotEmpty(t, scopeA)
require.NotEqual(t, scopeA, scopeB, "不同 session_id 的会话必须落在不同的作用域")
_, boundA := stateStore.GetSessionTurnState(groupID, scopeA)
require.True(t, boundA, "会话 A 的 turn state 应落在按原始 session_id 算出的作用域")
_, boundB := stateStore.GetSessionTurnState(groupID, scopeB)
require.True(t, boundB, "会话 B 的 turn state 应落在按原始 session_id 算出的作用域")
injected := resolveCodexFingerprintIDs(account, "", codexFingerprintFull)
require.NotNil(t, injected)
injectedScope, _ := deriveOpenAISessionHashes(fmt.Sprintf("openai_ws_exec:%d|thread=%s", apiKeyID, injected.threadID))
_, boundToInjected := stateStore.GetSessionTurnState(groupID, injectedScope)
require.False(t, boundToInjected, "指纹收敛注入的固定 thread_id 不得成为状态键")
threadBody := `{"model":"gpt-5.1","stream":false,"client_metadata":{"thread_id":"child-thread"},"input":[{"type":"input_text","text":"hello"}]}`
cC, rawC := forward("session-c", threadBody)
scopeC, threadC := resolveOpenAIWSExecutionScope(cC, rawC, apiKeyID)
require.Equal(t, "child-thread", threadC)
_, boundC := stateStore.GetSessionTurnState(groupID, scopeC)
require.True(t, boundC, "客户端自带线程标识时,键必须与按原始报文算出的一致,不受账号 namespace 改写影响")
}