From 1dab126944374f49ffd2014e9087c7276a49e9eb Mon Sep 17 00:00:00 2001 From: cat Date: Tue, 14 Jul 2026 16:46:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(openai):=20=E6=94=AF=E6=8C=81=20Agent=20Id?= =?UTF-8?q?entity=20=E8=AE=A4=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...ccount_codex_agent_identity_import_test.go | 47 ++ .../handler/admin/account_codex_import.go | 118 ++++- .../handler/dto/credentials_redact_test.go | 4 + .../service/account_credentials_redact.go | 2 +- .../internal/service/account_test_service.go | 89 +++- .../internal/service/account_usage_service.go | 22 +- .../internal/service/openai_agent_identity.go | 431 ++++++++++++++++++ .../openai_agent_identity_compat_test.go | 247 ++++++++++ .../service/openai_agent_identity_test.go | 177 +++++++ .../service/openai_codex_models_service.go | 12 +- .../service/openai_gateway_count_tokens.go | 10 +- .../service/openai_gateway_forward.go | 30 +- .../service/openai_gateway_passthrough.go | 12 +- .../service/openai_gateway_service.go | 4 + .../service/openai_gateway_upstream_errors.go | 1 + backend/internal/service/openai_images.go | 10 +- .../internal/service/openai_quota_service.go | 76 ++- .../service/openai_quota_spark_window_test.go | 43 ++ .../service/openai_ws_forwarder_ingress.go | 15 +- .../service/openai_ws_forwarder_payload.go | 4 +- .../service/openai_ws_forwarder_v2.go | 18 +- backend/internal/service/openai_ws_pool.go | 65 ++- .../openai_ws_v2_passthrough_adapter.go | 26 +- 23 files changed, 1399 insertions(+), 64 deletions(-) create mode 100644 backend/internal/handler/admin/account_codex_agent_identity_import_test.go create mode 100644 backend/internal/service/openai_agent_identity.go create mode 100644 backend/internal/service/openai_agent_identity_compat_test.go create mode 100644 backend/internal/service/openai_agent_identity_test.go diff --git a/backend/internal/handler/admin/account_codex_agent_identity_import_test.go b/backend/internal/handler/admin/account_codex_agent_identity_import_test.go new file mode 100644 index 0000000000..db55182b6b --- /dev/null +++ b/backend/internal/handler/admin/account_codex_agent_identity_import_test.go @@ -0,0 +1,47 @@ +package admin + +import ( + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "encoding/base64" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestNormalizeCodexImportEntryAcceptsAgentIdentityAuthJSON(t *testing.T) { + _, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + der, err := x509.MarshalPKCS8PrivateKey(privateKey) + require.NoError(t, err) + privateKeyBase64 := base64.StdEncoding.EncodeToString(der) + + item, err := normalizeCodexImportEntry(codexImportEntry{ + Index: 1, + Value: map[string]any{ + "auth_mode": "agentIdentity", + "agent_identity": map[string]any{ + "agent_runtime_id": "runtime-import", + "agent_private_key": privateKeyBase64, + "account_id": "account-import", + "chatgpt_user_id": "user-import", + "email": "agent@example.invalid", + "plan_type": "pro", + "chatgpt_account_is_fedramp": false, + }, + }, + }) + require.NoError(t, err) + require.NotNil(t, item) + require.True(t, item.IsAgentIdentity) + require.Equal(t, service.OpenAIAuthModeAgentIdentity, item.Credentials["auth_mode"]) + require.Equal(t, "runtime-import", item.Credentials["agent_runtime_id"]) + require.Equal(t, privateKeyBase64, item.Credentials["agent_private_key"]) + require.Equal(t, "account-import", item.Credentials["chatgpt_account_id"]) + require.Equal(t, "user-import", item.Credentials["chatgpt_user_id"]) + require.NotContains(t, item.Credentials, "access_token") + require.NotContains(t, item.Credentials, "refresh_token") + require.NotEmpty(t, item.WarningTexts) +} diff --git a/backend/internal/handler/admin/account_codex_import.go b/backend/internal/handler/admin/account_codex_import.go index 01a5fbfa1c..8abd269f23 100644 --- a/backend/internal/handler/admin/account_codex_import.go +++ b/backend/internal/handler/admin/account_codex_import.go @@ -72,20 +72,25 @@ type codexImportEntry struct { } type codexImportAccount struct { - Name string - AccessToken string - RefreshToken string - IDToken string - Email string - AccountID string - UserID string - PlanType string - Organization string - Credentials map[string]any - Extra map[string]any - TokenExpiresAt *time.Time - IdentityKeys []string - WarningTexts []string + Name string + AccessToken string + RefreshToken string + IDToken string + Email string + AccountID string + UserID string + PlanType string + Organization string + AgentRuntimeID string + AgentPrivateKey string + AgentTaskID string + AgentFedRAMP bool + IsAgentIdentity bool + Credentials map[string]any + Extra map[string]any + TokenExpiresAt *time.Time + IdentityKeys []string + WarningTexts []string } type codexJWTClaims struct { @@ -492,6 +497,41 @@ func normalizeCodexImportEntry(entry codexImportEntry) (*codexImportAccount, err case string: item.AccessToken = strings.TrimSpace(raw) case map[string]any: + if agentIdentity, ok := firstCodexMap(raw, []string{"agent_identity"}, []string{"agentIdentity"}); ok || strings.EqualFold(firstCodexString(raw, []string{"auth_mode"}, []string{"authMode"}), service.OpenAIAuthModeAgentIdentity) { + if !ok { + agentIdentity = raw + } + item.IsAgentIdentity = true + item.AgentRuntimeID = firstCodexString(agentIdentity, []string{"agent_runtime_id"}, []string{"agentRuntimeId"}) + item.AgentPrivateKey = firstCodexString(agentIdentity, []string{"agent_private_key"}, []string{"agentPrivateKey"}) + item.AgentTaskID = firstCodexString(agentIdentity, []string{"task_id"}, []string{"taskId"}) + item.AccountID = firstCodexString(agentIdentity, []string{"account_id"}, []string{"accountId"}) + item.UserID = firstCodexString(agentIdentity, []string{"chatgpt_user_id"}, []string{"chatgptUserId"}) + item.Email = firstCodexString(agentIdentity, []string{"email"}) + item.PlanType = firstCodexString(agentIdentity, []string{"plan_type"}, []string{"planType"}) + item.AgentFedRAMP = firstCodexBool(agentIdentity, []string{"chatgpt_account_is_fedramp"}, []string{"chatgptAccountIsFedramp"}) + if item.AgentRuntimeID == "" || item.AgentPrivateKey == "" || item.AccountID == "" || item.UserID == "" { + return nil, errors.New("Agent Identity 缺少必要字段") + } + if err := service.ValidateOpenAIAgentIdentityPrivateKey(item.AgentPrivateKey); err != nil { + return nil, errors.New("Agent Identity private key 格式无效") + } + item.Credentials["auth_mode"] = service.OpenAIAuthModeAgentIdentity + item.Credentials["agent_runtime_id"] = item.AgentRuntimeID + item.Credentials["agent_private_key"] = item.AgentPrivateKey + item.Credentials["chatgpt_account_id"] = item.AccountID + item.Credentials["chatgpt_user_id"] = item.UserID + item.Credentials["chatgpt_account_is_fedramp"] = item.AgentFedRAMP + setCodexCredentialIfNotEmpty(item.Credentials, "task_id", item.AgentTaskID) + setCodexCredentialIfNotEmpty(item.Credentials, "email", item.Email) + setCodexCredentialIfNotEmpty(item.Credentials, "plan_type", item.PlanType) + if item.AgentTaskID == "" { + item.WarningTexts = append(item.WarningTexts, "未包含 task_id，首次请求会使用现有 runtime 注册新 task") + } + item.IdentityKeys = buildCodexAgentIdentityKeys(item.AccountID, item.UserID, item.Email, item.AgentRuntimeID) + item.Name = buildCodexImportAccountName(item, entry.Index) + return item, nil + } item.AccessToken = firstCodexString(raw, []string{"tokens", "access_token"}, []string{"tokens", "accessToken"}, @@ -573,6 +613,9 @@ func normalizeCodexImportEntry(entry codexImportEntry) (*codexImportAccount, err return nil, fmt.Errorf("第 %d 条格式不支持", entry.Index) } + if item.IsAgentIdentity { + return item, nil + } if item.AccessToken == "" { return nil, errors.New("缺少 accessToken/access_token") } @@ -808,6 +851,9 @@ func sanitizeCodexImportCredentialExtras(input map[string]any) map[string]any { "openai_auth_mode": {}, "token_type": {}, "chatgpt_account_is_fedramp": {}, + "agent_runtime_id": {}, + "agent_private_key": {}, + "task_id": {}, } out := make(map[string]any, len(input)) for key, value := range input { @@ -838,6 +884,14 @@ func buildCodexImportIdentityKeys(accountID, userID, email, accessToken, refresh return buildCodexStoredIdentityKeys(accountID, userID, email, accessToken) } +func buildCodexAgentIdentityKeys(accountID, userID, email, runtimeID string) []string { + keys := buildCodexStoredIdentityKeys(accountID, userID, email, "") + if runtimeID = strings.TrimSpace(runtimeID); runtimeID != "" { + keys = append([]string{"agent:" + runtimeID}, keys...) + } + return keys +} + // buildCodexStoredIdentityKeys 生成存量账号索引键，保留 user/account 维度， // 让 accessToken-only 账号后续升级为完整 OAuth 时仍能命中并更新原账号。 func buildCodexStoredIdentityKeys(accountID, userID, email, accessToken string) []string { @@ -887,6 +941,10 @@ func (i *codexAccountIndex) Add(account service.Account) { for _, key := range keys { i.accountsByKey[key] = upsertCodexAccount(i.accountsByKey[key], account) } + if runtimeID := codexCredentialString(account.Credentials, "agent_runtime_id"); runtimeID != "" { + key := "agent:" + runtimeID + i.accountsByKey[key] = upsertCodexAccount(i.accountsByKey[key], account) + } } func (i *codexAccountIndex) remove(accountID int64) { @@ -1039,6 +1097,38 @@ func firstCodexString(obj map[string]any, paths ...[]string) string { return "" } +func firstCodexMap(obj map[string]any, paths ...[]string) (map[string]any, bool) { + for _, path := range paths { + value, ok := codexPathValue(obj, path) + if !ok || value == nil { + continue + } + if mapped, ok := value.(map[string]any); ok { + return mapped, true + } + } + return nil, false +} + +func firstCodexBool(obj map[string]any, paths ...[]string) bool { + for _, path := range paths { + value, ok := codexPathValue(obj, path) + if !ok { + continue + } + switch value := value.(type) { + case bool: + return value + case string: + parsed, err := strconv.ParseBool(strings.TrimSpace(value)) + if err == nil { + return parsed + } + } + } + return false +} + func copyCodexExtraString(obj map[string]any, extra map[string]any, key string, path []string) { value := firstCodexString(obj, path) if value != "" { diff --git a/backend/internal/handler/dto/credentials_redact_test.go b/backend/internal/handler/dto/credentials_redact_test.go index 431078fafd..9bf4f78039 100644 --- a/backend/internal/handler/dto/credentials_redact_test.go +++ b/backend/internal/handler/dto/credentials_redact_test.go @@ -20,6 +20,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) { "aws_secret_access_key": "aws-secret", "service_account_json": map[string]any{"private_key": "..."}, "private_key": "raw-key", + "agent_private_key": "agent-key-secret", // 非敏感 "base_url": "https://api.example.com", "model_mapping": map[string]any{"foo": "bar"}, @@ -35,6 +36,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) { require.NotContains(t, out, "aws_secret_access_key") require.NotContains(t, out, "service_account_json") require.NotContains(t, out, "private_key") + require.NotContains(t, out, "agent_private_key") require.Equal(t, "https://api.example.com", out["base_url"]) require.Equal(t, map[string]any{"foo": "bar"}, out["model_mapping"]) @@ -47,6 +49,7 @@ func TestRedactCredentials_StripsSensitiveKeysAndReportsStatus(t *testing.T) { require.True(t, status["has_aws_secret_access_key"]) require.True(t, status["has_service_account_json"]) require.True(t, status["has_private_key"]) + require.True(t, status["has_agent_private_key"]) // 状态 map 不应携带非敏感键的 has_* require.NotContains(t, status, "has_base_url") @@ -84,6 +87,7 @@ func TestRedactCredentials_AllKnownSensitiveKeys(t *testing.T) { "api_key", "session_key", "cookie", "aws_secret_access_key", "aws_session_token", "service_account_json", "service_account", "private_key", + "agent_private_key", } in := make(map[string]any, len(keys)) for _, k := range keys { diff --git a/backend/internal/service/account_credentials_redact.go b/backend/internal/service/account_credentials_redact.go index 76c2d1de5b..8eb70513ba 100644 --- a/backend/internal/service/account_credentials_redact.go +++ b/backend/internal/service/account_credentials_redact.go @@ -4,7 +4,7 @@ package service // dto 层做响应脱敏、service 层做更新合并都引用此清单——新增凭证类型时务必同步。 var SensitiveCredentialKeys = []string{ // OAuth - "access_token", "refresh_token", "id_token", + "access_token", "refresh_token", "id_token", "agent_private_key", // API Key 类 "api_key", "session_key", "cookie", // 云服务凭据 diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index b9dae84057..f83a99a7e2 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -15,6 +15,7 @@ import ( "net/http/httptest" "regexp" "strings" + "sync" "time" "github.com/Wei-Shaw/sub2api/internal/config" @@ -72,6 +73,7 @@ type AccountTestService struct { httpUpstream HTTPUpstream cfg *config.Config tlsFPProfileService *TLSFingerprintProfileService + agentIdentityTaskMu sync.Mutex } // NewAccountTestService creates a new AccountTestService @@ -544,9 +546,11 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account if credentialAccount.IsOAuth() { isOAuth = true - // OAuth - use Bearer token with ChatGPT internal API - authToken = credentialAccount.GetOpenAIAccessToken() - if authToken == "" { + // Agent Identity signs each request and does not retain the OAuth token. + if !credentialAccount.IsOpenAIAgentIdentity() { + authToken = credentialAccount.GetOpenAIAccessToken() + } + if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() { return s.sendErrorAndEnd(c, "No access token available") } @@ -597,7 +601,19 @@ func (s *AccountTestService) testOpenAIAccountConnection(c *gin.Context, account // Set common headers req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+authToken) + if credentialAccount.IsOpenAIAgentIdentity() { + authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount) + if authErr != nil { + return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication") + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } + } else { + req.Header.Set("Authorization", "Bearer "+authToken) + } // Set OAuth-specific headers for ChatGPT internal API if isOAuth { @@ -804,16 +820,26 @@ func (s *AccountTestService) testOpenAIChatCompletionsConnection( // resulting capability state on the account. func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account *Account, testModelID string) error { ctx := c.Request.Context() + credentialAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return s.sendErrorAndEnd(c, "Failed to resolve account credentials") + } + credentialAccount = resolved + } authToken := "" apiURL := "" isOAuth := false switch { - case account.IsOAuth(): + case credentialAccount.IsOAuth(): isOAuth = true - authToken = account.GetOpenAIAccessToken() - if authToken == "" { + if !credentialAccount.IsOpenAIAgentIdentity() { + authToken = credentialAccount.GetOpenAIAccessToken() + } + if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() { return s.sendErrorAndEnd(c, "No access token available") } apiURL = chatgptCodexAPIURL + "/compact" @@ -852,7 +878,19 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json") - req.Header.Set("Authorization", "Bearer "+authToken) + if credentialAccount.IsOpenAIAgentIdentity() { + authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount) + if authErr != nil { + return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication") + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } + } else { + req.Header.Set("Authorization", "Bearer "+authToken) + } req.Header.Set("OpenAI-Beta", "responses=experimental") req.Header.Set("Originator", "codex_cli_rs") req.Header.Set("User-Agent", codexCLIUserAgent) @@ -863,7 +901,7 @@ func (s *AccountTestService) testOpenAICompactConnection(c *gin.Context, account if isOAuth { req.Host = "chatgpt.com" - setOpenAIChatGPTAccountHeaders(req.Header, account) + setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount) } // 账号级请求头覆写：测试请求与真实转发保持一致的最终头 @@ -1677,8 +1715,19 @@ func (s *AccountTestService) testOpenAIImageAPIKey(c *gin.Context, ctx context.C // testOpenAIImageOAuth tests OpenAI image generation using an OAuth account via Codex /responses API. func (s *AccountTestService) testOpenAIImageOAuth(c *gin.Context, ctx context.Context, account *Account, modelID, prompt string) error { - authToken := account.GetOpenAIAccessToken() - if authToken == "" { + credentialAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return s.sendErrorAndEnd(c, "Failed to resolve account credentials") + } + credentialAccount = resolved + } + authToken := "" + if !credentialAccount.IsOpenAIAgentIdentity() { + authToken = credentialAccount.GetOpenAIAccessToken() + } + if authToken == "" && !credentialAccount.IsOpenAIAgentIdentity() { return s.sendErrorAndEnd(c, "No access token available") } @@ -1710,17 +1759,29 @@ func (s *AccountTestService) testOpenAIImageOAuth(c *gin.Context, ctx context.Co } req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) req.Host = "chatgpt.com" - req.Header.Set("Authorization", "Bearer "+authToken) + if credentialAccount.IsOpenAIAgentIdentity() { + authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, credentialAccount) + if authErr != nil { + return s.sendErrorAndEnd(c, "Failed to build Agent Identity authentication") + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } + } else { + req.Header.Set("Authorization", "Bearer "+authToken) + } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") req.Header.Set("OpenAI-Beta", "responses=experimental") req.Header.Set("originator", "codex_cli_rs") - if customUA := strings.TrimSpace(account.GetOpenAIUserAgent()); customUA != "" { + if customUA := strings.TrimSpace(credentialAccount.GetOpenAIUserAgent()); customUA != "" { req.Header.Set("User-Agent", customUA) } else { req.Header.Set("User-Agent", codexCLIUserAgent) } - setOpenAIChatGPTAccountHeaders(req.Header, account) + setOpenAIChatGPTAccountHeaders(req.Header, credentialAccount) // 与真实转发一致：originator 与最终 User-Agent 首段配套（原 opencode 与 Codex UA 错配会 404，issue #3901）。 enforceCodexIdentityHeaders(req.Header) diff --git a/backend/internal/service/account_usage_service.go b/backend/internal/service/account_usage_service.go index 281122d4f0..4dc09c4109 100644 --- a/backend/internal/service/account_usage_service.go +++ b/backend/internal/service/account_usage_service.go @@ -291,6 +291,7 @@ type AccountUsageService struct { cache *UsageCache identityCache IdentityCache tlsFPProfileService *TLSFingerprintProfileService + agentIdentityTaskMu sync.Mutex } // NewAccountUsageService 创建AccountUsageService实例 @@ -682,8 +683,11 @@ func (s *AccountUsageService) probeOpenAICodexSnapshot(ctx context.Context, acco if account == nil || !account.IsOAuth() { return nil, nil } - accessToken := account.GetOpenAIAccessToken() - if accessToken == "" { + accessToken := "" + if !account.IsOpenAIAgentIdentity() { + accessToken = account.GetOpenAIAccessToken() + } + if accessToken == "" && !account.IsOpenAIAgentIdentity() { return nil, fmt.Errorf("no access token available") } modelID := openaipkg.DefaultTestModel @@ -701,7 +705,19 @@ func (s *AccountUsageService) probeOpenAICodexSnapshot(ctx context.Context, acco } req.Host = "chatgpt.com" req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+accessToken) + if account.IsOpenAIAgentIdentity() { + authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, account) + if authErr != nil { + return nil, fmt.Errorf("build Agent Identity authentication: %w", authErr) + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } + } else { + req.Header.Set("Authorization", "Bearer "+accessToken) + } req.Header.Set("Accept", "text/event-stream") req.Header.Set("OpenAI-Beta", "responses=experimental") req.Header.Set("Originator", "codex_cli_rs") diff --git a/backend/internal/service/openai_agent_identity.go b/backend/internal/service/openai_agent_identity.go new file mode 100644 index 0000000000..c8818cd6f3 --- /dev/null +++ b/backend/internal/service/openai_agent_identity.go @@ -0,0 +1,431 @@ +package service + +import ( + "context" + "crypto" + "crypto/ed25519" + "crypto/sha512" + "crypto/x509" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/httpclient" + "golang.org/x/crypto/curve25519" + "golang.org/x/crypto/nacl/box" +) + +const ( + OpenAIAuthModeAgentIdentity = "agentIdentity" + agentIdentityAuthAPIBaseURL = "https://auth.openai.com/api/accounts" + agentIdentityTaskRegistrationTimeout = 30 * time.Second +) + +var openAIAgentIdentityAuthAPIBaseURL = agentIdentityAuthAPIBaseURL + +type agentIdentityKey struct { + runtimeID string + privateKey ed25519.PrivateKey + taskID string +} + +type agentIdentityTaskRegistrationResponse struct { + TaskID string `json:"task_id"` + TaskIDCamel string `json:"taskId"` + EncryptedTaskID string `json:"encrypted_task_id"` + EncryptedTaskIDCamel string `json:"encryptedTaskId"` +} + +type agentIdentityTaskRecoveredError struct{} + +func (e *agentIdentityTaskRecoveredError) Error() string { + return "agent identity task recovered" +} + +func (a *Account) IsOpenAIAgentIdentity() bool { + if a == nil || !a.IsOpenAIOAuth() { + return false + } + return strings.EqualFold(strings.TrimSpace(a.GetCredential(openAIAuthModeCredentialKey)), OpenAIAuthModeAgentIdentity) +} + +func agentIdentityPrivateKey(account *Account) (ed25519.PrivateKey, error) { + if account == nil { + return nil, errors.New("agent identity account is nil") + } + raw := strings.TrimSpace(account.GetCredential("agent_private_key")) + if raw == "" { + return nil, errors.New("agent identity private key is missing") + } + der, err := base64.StdEncoding.DecodeString(raw) + if err != nil { + return nil, errors.New("agent identity private key is not valid base64") + } + key, err := x509.ParsePKCS8PrivateKey(der) + if err != nil { + return nil, errors.New("agent identity private key is not valid PKCS#8") + } + privateKey, ok := key.(ed25519.PrivateKey) + if !ok || len(privateKey) != ed25519.PrivateKeySize { + return nil, errors.New("agent identity private key is not Ed25519") + } + return privateKey, nil +} + +// ValidateOpenAIAgentIdentityPrivateKey validates the stored PKCS#8 Ed25519 +// form without returning or logging the key material. +func ValidateOpenAIAgentIdentityPrivateKey(encoded string) error { + account := &Account{Credentials: map[string]any{"agent_private_key": encoded}} + _, err := agentIdentityPrivateKey(account) + return err +} + +func agentIdentityKeyFromAccount(account *Account) (agentIdentityKey, error) { + privateKey, err := agentIdentityPrivateKey(account) + if err != nil { + return agentIdentityKey{}, err + } + runtimeID := strings.TrimSpace(account.GetCredential("agent_runtime_id")) + if runtimeID == "" { + return agentIdentityKey{}, errors.New("agent identity runtime id is missing") + } + return agentIdentityKey{ + runtimeID: runtimeID, + privateKey: privateKey, + taskID: strings.TrimSpace(account.GetCredential("task_id")), + }, nil +} + +func buildAgentAssertion(key agentIdentityKey, now time.Time) (string, error) { + if key.runtimeID == "" || key.taskID == "" { + return "", errors.New("agent identity runtime or task id is missing") + } + timestamp := now.UTC().Format(time.RFC3339) + payload := []byte(key.runtimeID + ":" + key.taskID + ":" + timestamp) + signature, err := key.privateKey.Sign(nil, payload, crypto.Hash(0)) + if err != nil { + return "", errors.New("failed to sign agent assertion") + } + envelope := map[string]string{ + "agent_runtime_id": key.runtimeID, + "task_id": key.taskID, + "timestamp": timestamp, + "signature": base64.StdEncoding.EncodeToString(signature), + } + encoded, err := json.Marshal(envelope) + if err != nil { + return "", errors.New("failed to serialize agent assertion") + } + return "AgentAssertion " + base64.RawURLEncoding.EncodeToString(encoded), nil +} + +func signAgentTaskRegistration(key agentIdentityKey, timestamp time.Time) (string, string, error) { + if key.runtimeID == "" { + return "", "", errors.New("agent identity runtime id is missing") + } + formatted := timestamp.UTC().Format(time.RFC3339) + signature, err := key.privateKey.Sign(nil, []byte(key.runtimeID+":"+formatted), crypto.Hash(0)) + if err != nil { + return "", "", errors.New("failed to sign agent task registration") + } + return formatted, base64.StdEncoding.EncodeToString(signature), nil +} + +func decryptAgentTaskID(key agentIdentityKey, encoded string) (string, error) { + ciphertext, err := base64.StdEncoding.DecodeString(strings.TrimSpace(encoded)) + if err != nil { + return "", errors.New("encrypted agent task id is not valid base64") + } + seed := key.privateKey.Seed() + digest := sha512.Sum512(seed) + var curvePrivate [32]byte + copy(curvePrivate[:], digest[:32]) + curvePrivate[0] &= 248 + curvePrivate[31] &= 127 + curvePrivate[31] |= 64 + curvePublicBytes, err := curve25519.X25519(curvePrivate[:], curve25519.Basepoint) + if err != nil { + return "", errors.New("failed to derive agent identity decryption key") + } + var curvePublic [32]byte + copy(curvePublic[:], curvePublicBytes) + plaintext, ok := box.OpenAnonymous(nil, ciphertext, &curvePublic, &curvePrivate) + if !ok { + return "", errors.New("failed to decrypt encrypted agent task id") + } + taskID := strings.TrimSpace(string(plaintext)) + if taskID == "" { + return "", errors.New("decrypted agent task id is empty") + } + return taskID, nil +} + +func registerAgentIdentityTask(ctx context.Context, account *Account) (string, error) { + key, err := agentIdentityKeyFromAccount(account) + if err != nil { + return "", err + } + timestamp, signature, err := signAgentTaskRegistration(key, time.Now()) + if err != nil { + return "", err + } + proxyURL := "" + if account.ProxyID != nil && account.Proxy != nil { + proxyURL = account.Proxy.URL() + } + client, err := httpclient.GetClient(httpclient.Options{ + ProxyURL: proxyURL, + Timeout: agentIdentityTaskRegistrationTimeout, + ResponseHeaderTimeout: 15 * time.Second, + }) + if err != nil { + return "", errors.New("invalid proxy configuration for agent task registration") + } + body, err := json.Marshal(map[string]string{ + "timestamp": timestamp, + "signature": signature, + }) + if err != nil { + return "", errors.New("failed to serialize agent task registration") + } + url := strings.TrimRight(strings.TrimSpace(openAIAgentIdentityAuthAPIBaseURL), "/") + "/v1/agent/" + key.runtimeID + "/task/register" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(string(body))) + if err != nil { + return "", errors.New("failed to build agent task registration request") + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + resp, err := client.Do(req) + if err != nil { + return "", errors.New("agent task registration request failed") + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return "", fmt.Errorf("agent task registration returned status %d", resp.StatusCode) + } + var result agentIdentityTaskRegistrationResponse + if err := json.NewDecoder(io.LimitReader(resp.Body, 64*1024)).Decode(&result); err != nil { + return "", errors.New("agent task registration response is invalid") + } + if taskID := strings.TrimSpace(result.TaskID); taskID != "" { + return taskID, nil + } + if taskID := strings.TrimSpace(result.TaskIDCamel); taskID != "" { + return taskID, nil + } + encrypted := strings.TrimSpace(result.EncryptedTaskID) + if encrypted == "" { + encrypted = strings.TrimSpace(result.EncryptedTaskIDCamel) + } + if encrypted == "" { + return "", errors.New("agent task registration response omitted task id") + } + return decryptAgentTaskID(key, encrypted) +} + +func ensureAgentIdentityTaskForAccount(ctx context.Context, repo AccountRepository, pool *openAIWSConnPool, taskMu *sync.Mutex, account *Account, expectedTaskID string) error { + if account == nil || !account.IsOpenAIAgentIdentity() { + return nil + } + credAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, repo, account) + if err != nil { + return err + } + credAccount = resolved + } + if credAccount == nil || !credAccount.IsOpenAIAgentIdentity() { + return errors.New("agent identity credentials are unavailable") + } + currentTaskID := strings.TrimSpace(credAccount.GetCredential("task_id")) + if currentTaskID != "" && (expectedTaskID == "" || currentTaskID != expectedTaskID) { + return nil + } + if taskMu == nil { + return errors.New("agent identity task lock is unavailable") + } + taskMu.Lock() + defer taskMu.Unlock() + currentTaskID = strings.TrimSpace(credAccount.GetCredential("task_id")) + if currentTaskID != "" && (expectedTaskID == "" || currentTaskID != expectedTaskID) { + return nil + } + newTaskID, err := registerAgentIdentityTask(ctx, credAccount) + if err != nil { + return err + } + credentials := make(map[string]any, len(credAccount.Credentials)+1) + for key, value := range credAccount.Credentials { + credentials[key] = value + } + credentials["task_id"] = newTaskID + if err := persistAccountCredentials(ctx, repo, credAccount, credentials); err != nil { + return err + } + if pool != nil { + pool.ClearAccount(credAccount.ID) + } + return nil +} + +func (s *OpenAIGatewayService) ensureAgentIdentityTask(ctx context.Context, account *Account, expectedTaskID string) error { + if s == nil { + return errors.New("openai gateway service is nil") + } + return ensureAgentIdentityTaskForAccount(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, account, expectedTaskID) +} + +func isAgentIdentityTaskInvalidHTTPResponse(statusCode int, body []byte) bool { + if statusCode != http.StatusUnauthorized { + return false + } + lower := strings.ToLower(string(body)) + for _, marker := range []string{ + "invalid task", + "task_id", + "task id", + "task_not_found", + "task_expired", + "unknown task", + } { + if strings.Contains(lower, marker) { + return true + } + } + return false +} + +func (s *OpenAIGatewayService) buildOpenAIAuthenticationHeaders(ctx context.Context, account *Account, token string) (http.Header, error) { + if account == nil { + return nil, errors.New("account is nil") + } + credAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return nil, err + } + credAccount = resolved + } + headers := make(http.Header) + if credAccount != nil && credAccount.IsOpenAIAgentIdentity() { + agentHeaders, err := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, credAccount) + if err != nil { + return nil, err + } + return agentHeaders, nil + } + headers.Set("Authorization", "Bearer "+token) + return headers, nil +} + +func buildAgentIdentityAuthenticationHeaders(ctx context.Context, repo AccountRepository, pool *openAIWSConnPool, taskMu *sync.Mutex, account *Account) (http.Header, error) { + if account == nil || !account.IsOpenAIAgentIdentity() { + return nil, errors.New("agent identity account is required") + } + if err := ensureAgentIdentityTaskForAccount(ctx, repo, pool, taskMu, account, ""); err != nil { + return nil, err + } + key, err := agentIdentityKeyFromAccount(account) + if err != nil { + return nil, err + } + assertion, err := buildAgentAssertion(key, time.Now()) + if err != nil { + return nil, err + } + headers := make(http.Header) + headers.Set("Authorization", assertion) + return headers, nil +} + +func (s *OpenAIGatewayService) refreshOpenAIAgentIdentityHeaders(ctx context.Context, account *Account, headers http.Header) (http.Header, error) { + if account == nil { + return cloneHeader(headers), nil + } + credAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return nil, err + } + credAccount = resolved + } + if !credAccount.IsOpenAIAgentIdentity() { + return cloneHeader(headers), nil + } + refreshed := cloneHeader(headers) + if refreshed == nil { + refreshed = make(http.Header) + } + authHeaders, err := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.openaiWSPool, &s.agentIdentityTaskMu, credAccount) + if err != nil { + return nil, err + } + refreshed.Set("Authorization", authHeaders.Get("Authorization")) + return refreshed, nil +} + +func (s *OpenAIGatewayService) recoverAgentIdentityTask(ctx context.Context, account *Account, expectedTaskID string) error { + if account != nil && account.IsShadow() { + if resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account); err == nil && resolved != nil && strings.TrimSpace(expectedTaskID) == "" { + expectedTaskID = strings.TrimSpace(resolved.GetCredential("task_id")) + } + } + return s.ensureAgentIdentityTask(ctx, account, expectedTaskID) +} + +func (s *OpenAIGatewayService) isAgentIdentityAccount(ctx context.Context, account *Account) bool { + if account == nil { + return false + } + credAccount := account + if account.IsShadow() { + resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account) + if err != nil { + return false + } + credAccount = resolved + } + return credAccount != nil && credAccount.IsOpenAIAgentIdentity() +} + +// redactAgentIdentitySensitiveBody removes credential values before an +// upstream error can reach logs, ops events, or returned error text. Agent +// Identity responses should not echo these values, but keeping this boundary +// defensive prevents accidental disclosure if an upstream error does. +func (s *OpenAIGatewayService) redactAgentIdentitySensitiveBody(ctx context.Context, account *Account, body []byte) []byte { + if !s.isAgentIdentityAccount(ctx, account) || len(body) == 0 { + return body + } + credAccount := account + if account != nil && account.IsShadow() { + if resolved, err := resolveCredentialAccount(ctx, s.accountRepo, account); err == nil && resolved != nil { + credAccount = resolved + } + } + redacted := string(body) + for _, key := range []string{ + "agent_private_key", + "agent_runtime_id", + "task_id", + "access_token", + "refresh_token", + "id_token", + "api_key", + "session_key", + "cookie", + } { + if value := strings.TrimSpace(credAccount.GetCredential(key)); value != "" { + redacted = strings.ReplaceAll(redacted, value, "[redacted]") + } + } + return []byte(redacted) +} diff --git a/backend/internal/service/openai_agent_identity_compat_test.go b/backend/internal/service/openai_agent_identity_compat_test.go new file mode 100644 index 0000000000..5c0c996c42 --- /dev/null +++ b/backend/internal/service/openai_agent_identity_compat_test.go @@ -0,0 +1,247 @@ +package service + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestAccountTestServiceOpenAICompactAgentIdentityUsesFreshAssertion(t *testing.T) { + gin.SetMode(gin.TestMode) + key, privateKey := newTestAgentIdentityKey(t) + account := Account{ + ID: 21, + Name: "agent-identity", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + "task_id": key.taskID, + "chatgpt_account_id": "account-agent-test", + "chatgpt_account_is_fedramp": true, + }, + } + repo := &snapshotUpdateAccountRepo{stubOpenAIAccountRepo: stubOpenAIAccountRepo{accounts: []Account{account}}} + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"compact-agent","status":"completed"}`)), + }} + svc := &AccountTestService{accountRepo: repo, httpUpstream: upstream} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/21/test", bytes.NewReader(nil)) + + require.NoError(t, svc.TestAccountConnection(c, account.ID, "gpt-5.4", "", AccountTestModeCompact)) + require.Equal(t, "AgentAssertion", strings.SplitN(upstream.lastReq.Header.Get("Authorization"), " ", 2)[0]) + require.Equal(t, "account-agent-test", upstream.lastReq.Header.Get("chatgpt-account-id")) + require.Equal(t, "true", upstream.lastReq.Header.Get("x-openai-fedramp")) + require.NotContains(t, upstream.lastReq.Header.Get("Authorization"), privateKey) +} + +func TestOpenAIAgentIdentityPassthroughKeepsSessionAndPromptCacheHeaders(t *testing.T) { + gin.SetMode(gin.TestMode) + key, privateKey := newTestAgentIdentityKey(t) + account := &Account{ + ID: 24, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + "task_id": key.taskID, + "chatgpt_account_id": "account-agent-passthrough", + }, + } + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + body := []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":true,"prompt_cache_key":"cache-agent"}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("session_id", "client-session") + c.Request.Header.Set("conversation_id", "client-conversation") + c.Request.Header.Set("Authorization", "Bearer inbound-must-not-forward") + + svc := &OpenAIGatewayService{} + req, err := svc.buildUpstreamRequestOpenAIPassthrough(context.Background(), c, account, body, "") + require.NoError(t, err) + require.Equal(t, "AgentAssertion", strings.SplitN(req.Header.Get("Authorization"), " ", 2)[0]) + require.Equal(t, "account-agent-passthrough", req.Header.Get("chatgpt-account-id")) + require.NotEqual(t, "client-session", req.Header.Get("session_id")) + require.NotEqual(t, "client-conversation", req.Header.Get("conversation_id")) + require.Equal(t, isolateOpenAISessionID(0, "cache-agent"), req.Header.Get("session_id")) +} + +func TestOpenAIAgentIdentityErrorRedactionDoesNotLeakCredentialValues(t *testing.T) { + key, privateKey := newTestAgentIdentityKey(t) + account := &Account{ + ID: 25, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + "task_id": key.taskID, + "access_token": key.runtimeID + "-oauth-value", + }, + } + svc := &OpenAIGatewayService{} + oauthValue := account.GetCredential("access_token") + redacted := svc.redactAgentIdentitySensitiveBody(context.Background(), account, []byte(`{"message":"runtime-test task-test `+oauthValue+`"}`)) + require.NotContains(t, string(redacted), key.runtimeID) + require.NotContains(t, string(redacted), key.taskID) + require.NotContains(t, string(redacted), oauthValue) + require.Contains(t, string(redacted), "[redacted]") +} + +func TestOpenAIWSConnPoolHeadersFactoryRunsAtDialAndStalePrewarmIsDiscarded(t *testing.T) { + cfg := &config.Config{} + cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1 + pool := newOpenAIWSConnPool(cfg) + defer pool.Close() + pool.setClientDialerForTest(&openAIWSFakeDialer{}) + + accountID := int64(22) + ap := pool.getOrCreateAccountPool(accountID) + factoryCalls := 0 + latestHeader := "" + req := openAIWSAcquireRequest{ + Account: &Account{ID: accountID, Platform: PlatformOpenAI, Type: AccountTypeOAuth}, + WSURL: "wss://example.com/v1/responses", + HeadersFactory: func(_ context.Context, headers http.Header) (http.Header, error) { + factoryCalls++ + latestHeader = "AgentAssertion dial-" + string(rune('0'+factoryCalls)) + if headers == nil { + headers = make(http.Header) + } + headers.Set("Authorization", latestHeader) + return headers, nil + }, + } + ap.mu.Lock() + ap.lastAcquire = &req + generation := ap.generation + ap.mu.Unlock() + + pool.prewarmConns(accountID, req, 1, generation) + require.Equal(t, 1, factoryCalls, "prewarm must generate authorization inside the actual dial") + require.Equal(t, "AgentAssertion dial-1", latestHeader) + + pool.ClearAccount(accountID) + ap.mu.Lock() + require.Empty(t, ap.conns, "credential recovery must remove pooled connections") + require.Nil(t, ap.lastAcquire, "credential recovery must discard delayed acquire state") + require.Equal(t, generation+1, ap.generation) + ap.mu.Unlock() + + // A prewarm captured before ClearAccount must not be admitted after recovery. + pool.prewarmConns(accountID, req, 1, generation) + ap.mu.Lock() + require.Empty(t, ap.conns) + ap.mu.Unlock() +} + +func TestOpenAIAgentIdentityTaskInvalidRetriesExactlyOnce(t *testing.T) { + gin.SetMode(gin.TestMode) + key, privateKey := newTestAgentIdentityKey(t) + account := &Account{ + ID: 23, + Name: "agent-identity", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Status: StatusActive, + Schedulable: true, + Concurrency: 1, + Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + "task_id": "task-old", + "chatgpt_account_id": "account-agent-retry", + }, + } + repo := &agentIdentityForwardRepo{account: account} + registerCalls := 0 + registerServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + registerCalls++ + _, _ = io.WriteString(w, `{"task_id":"task-new"}`) + })) + defer registerServer.Close() + oldBase := openAIAgentIdentityAuthAPIBaseURL + openAIAgentIdentityAuthAPIBaseURL = registerServer.URL + t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase }) + + successBody := `{"id":"resp-agent-retry","object":"response","model":"gpt-5.4","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}` + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + {StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(successBody))}, + }} + svc := &OpenAIGatewayService{cfg: &config.Config{}, accountRepo: repo, httpUpstream: upstream} + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`)) + + _, err := svc.Forward(context.Background(), c, account, []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`)) + require.NoError(t, err) + require.Equal(t, 1, registerCalls) + require.Len(t, upstream.requests, 2) + require.NotEqual(t, upstream.requests[0].Header.Get("Authorization"), upstream.requests[1].Header.Get("Authorization")) + require.Equal(t, "task-new", decodeAgentAssertionTask(t, upstream.requests[1].Header.Get("Authorization"))) + + // Two consecutive invalid responses still produce only one retry for this + // request; the recovery path must not loop indefinitely. + upstream.responses = []*http.Response{ + {StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))}, + {StatusCode: http.StatusUnauthorized, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_task_id"}}`))}, + } + rec2 := httptest.NewRecorder() + c2, _ := gin.CreateTestContext(rec2) + c2.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`)) + _, err = svc.Forward(context.Background(), c2, account, []byte(`{"model":"gpt-5.4","instructions":"Reply OK","input":[],"stream":false}`)) + require.Error(t, err) + require.Equal(t, 2, registerCalls) + require.Len(t, upstream.requests, 4) +} + +func decodeAgentAssertionTask(t *testing.T, header string) string { + t.Helper() + encoded := strings.TrimPrefix(header, "AgentAssertion ") + decoded, err := base64.RawURLEncoding.DecodeString(encoded) + require.NoError(t, err) + var envelope struct { + TaskID string `json:"task_id"` + } + require.NoError(t, json.Unmarshal(decoded, &envelope)) + return envelope.TaskID +} + +type agentIdentityForwardRepo struct { + AccountRepository + account *Account +} + +func (r *agentIdentityForwardRepo) GetByID(_ context.Context, _ int64) (*Account, error) { + return r.account, nil +} + +func (r *agentIdentityForwardRepo) UpdateCredentials(_ context.Context, _ int64, credentials map[string]any) error { + r.account.Credentials = credentials + return nil +} diff --git a/backend/internal/service/openai_agent_identity_test.go b/backend/internal/service/openai_agent_identity_test.go new file mode 100644 index 0000000000..73ac0ddf08 --- /dev/null +++ b/backend/internal/service/openai_agent_identity_test.go @@ -0,0 +1,177 @@ +package service + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "crypto/sha512" + "crypto/x509" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/crypto/curve25519" + "golang.org/x/crypto/nacl/box" +) + +func newTestAgentIdentityKey(t *testing.T) (agentIdentityKey, string) { + t.Helper() + _, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + der, err := x509.MarshalPKCS8PrivateKey(privateKey) + require.NoError(t, err) + return agentIdentityKey{ + runtimeID: "runtime-test", + privateKey: privateKey, + taskID: "task-test", + }, base64.StdEncoding.EncodeToString(der) +} + +func TestBuildAgentAssertionMatchesCodexEnvelopeAndSignature(t *testing.T) { + key, _ := newTestAgentIdentityKey(t) + now := time.Date(2026, 7, 14, 8, 9, 10, 0, time.FixedZone("UTC+8", 8*60*60)) + assertion, err := buildAgentAssertion(key, now) + require.NoError(t, err) + require.True(t, strings.HasPrefix(assertion, "AgentAssertion ")) + + encoded := strings.TrimPrefix(assertion, "AgentAssertion ") + decoded, err := base64.RawURLEncoding.DecodeString(encoded) + require.NoError(t, err) + var envelope struct { + AgentRuntimeID string `json:"agent_runtime_id"` + TaskID string `json:"task_id"` + Timestamp string `json:"timestamp"` + Signature string `json:"signature"` + } + require.NoError(t, json.Unmarshal(decoded, &envelope)) + require.Equal(t, "runtime-test", envelope.AgentRuntimeID) + require.Equal(t, "task-test", envelope.TaskID) + require.Equal(t, "2026-07-14T00:09:10Z", envelope.Timestamp) + signature, err := base64.StdEncoding.DecodeString(envelope.Signature) + require.NoError(t, err) + require.True(t, ed25519.Verify(key.privateKey.Public().(ed25519.PublicKey), []byte("runtime-test:task-test:2026-07-14T00:09:10Z"), signature)) +} + +func TestDecryptAgentTaskIDSupportsCodexSealedBoxResponse(t *testing.T) { + key, _ := newTestAgentIdentityKey(t) + digest := sha512.Sum512(key.privateKey.Seed()) + var curvePrivate [32]byte + copy(curvePrivate[:], digest[:32]) + curvePrivate[0] &= 248 + curvePrivate[31] &= 127 + curvePrivate[31] |= 64 + curvePublicBytes, err := curve25519.X25519(curvePrivate[:], curve25519.Basepoint) + require.NoError(t, err) + var curvePublic [32]byte + copy(curvePublic[:], curvePublicBytes) + ciphertext, err := box.SealAnonymous(nil, []byte("task-sealed"), &curvePublic, rand.Reader) + require.NoError(t, err) + got, err := decryptAgentTaskID(key, base64.StdEncoding.EncodeToString(ciphertext)) + require.NoError(t, err) + require.Equal(t, "task-sealed", got) +} + +func TestRegisterAgentIdentityTaskAcceptsPlaintextAndEncryptedResponses(t *testing.T) { + key, privateKey := newTestAgentIdentityKey(t) + requestCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, http.MethodPost, r.Method) + require.Equal(t, "/v1/agent/runtime-test/task/register", r.URL.Path) + var request map[string]string + require.NoError(t, json.NewDecoder(r.Body).Decode(&request)) + require.NotEmpty(t, request["timestamp"]) + require.NotEmpty(t, request["signature"]) + requestCount++ + if requestCount == 2 { + digest := sha512.Sum512(key.privateKey.Seed()) + var curvePrivate [32]byte + copy(curvePrivate[:], digest[:32]) + curvePrivate[0] &= 248 + curvePrivate[31] &= 127 + curvePrivate[31] |= 64 + curvePublicBytes, curveErr := curve25519.X25519(curvePrivate[:], curve25519.Basepoint) + require.NoError(t, curveErr) + var curvePublic [32]byte + copy(curvePublic[:], curvePublicBytes) + ciphertext, sealErr := box.SealAnonymous(nil, []byte("task-encrypted"), &curvePublic, rand.Reader) + require.NoError(t, sealErr) + _, _ = fmt.Fprintf(w, `{"encrypted_task_id":%q}`, base64.StdEncoding.EncodeToString(ciphertext)) + return + } + _, _ = w.Write([]byte(`{"task_id":"task-plain"}`)) + })) + defer server.Close() + oldBase := openAIAgentIdentityAuthAPIBaseURL + openAIAgentIdentityAuthAPIBaseURL = server.URL + t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase }) + + account := &Account{ID: 1, Type: AccountTypeOAuth, Platform: PlatformOpenAI, Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + }} + taskID, err := registerAgentIdentityTask(context.Background(), account) + require.NoError(t, err) + require.Equal(t, "task-plain", taskID) + taskID, err = registerAgentIdentityTask(context.Background(), account) + require.NoError(t, err) + require.Equal(t, "task-encrypted", taskID) +} + +func TestEnsureAgentIdentityTaskPersistsAndRedactsCredentials(t *testing.T) { + key, privateKey := newTestAgentIdentityKey(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"task_id":"task-persisted"}`)) + })) + defer server.Close() + oldBase := openAIAgentIdentityAuthAPIBaseURL + openAIAgentIdentityAuthAPIBaseURL = server.URL + t.Cleanup(func() { openAIAgentIdentityAuthAPIBaseURL = oldBase }) + + repo := &agentIdentityCredentialsRepo{} + account := &Account{ID: 7, Type: AccountTypeOAuth, Platform: PlatformOpenAI, Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": key.runtimeID, + "agent_private_key": privateKey, + "chatgpt_account_id": "account-test", + }} + service := &OpenAIGatewayService{accountRepo: repo} + require.NoError(t, service.ensureAgentIdentityTask(context.Background(), account, "")) + require.Equal(t, "task-persisted", account.GetCredential("task_id")) + require.Equal(t, "task-persisted", repo.credentials["task_id"]) + require.True(t, IsSensitiveCredentialKey("agent_private_key")) + redacted := make(map[string]any) + for key, value := range account.Credentials { + if !IsSensitiveCredentialKey(key) { + redacted[key] = value + } + } + require.NotContains(t, string(mustJSON(t, redacted)), privateKey) +} + +type agentIdentityCredentialsRepo struct { + AccountRepository + credentials map[string]any + mu sync.Mutex +} + +func (r *agentIdentityCredentialsRepo) UpdateCredentials(_ context.Context, _ int64, credentials map[string]any) error { + r.mu.Lock() + defer r.mu.Unlock() + r.credentials = credentials + return nil +} + +func mustJSON(t *testing.T, value any) []byte { + t.Helper() + encoded, err := json.Marshal(value) + require.NoError(t, err) + return encoded +} diff --git a/backend/internal/service/openai_codex_models_service.go b/backend/internal/service/openai_codex_models_service.go index 8a919fa2b0..0eb2e1a071 100644 --- a/backend/internal/service/openai_codex_models_service.go +++ b/backend/internal/service/openai_codex_models_service.go @@ -42,7 +42,7 @@ func (s *OpenAIGatewayService) FetchCodexModelsManifest(ctx context.Context, acc return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_CODEX_MODELS_CREDENTIALS_FAILED", "resolve credential account: %v", err) } accessToken := credAccount.GetOpenAIAccessToken() - if accessToken == "" { + if accessToken == "" && !credAccount.IsOpenAIAgentIdentity() { return nil, infraerrors.New(http.StatusBadGateway, "OPENAI_CODEX_MODELS_TOKEN_MISSING", "account has no Codex backend access token") } @@ -58,7 +58,15 @@ func (s *OpenAIGatewayService) FetchCodexModelsManifest(ctx context.Context, acc if err != nil { return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_CODEX_MODELS_REQUEST_FAILED", "create codex models request: %v", err) } - req.Header.Set("Authorization", "Bearer "+accessToken) + authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, credAccount, accessToken) + if err != nil { + return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_AUTH_FAILED", "build Codex models authentication: %v", err) + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } req.Header.Set("Accept", "application/json") req.Header.Set("Originator", "codex_cli_rs") req.Header.Set("Version", clientVersion) diff --git a/backend/internal/service/openai_gateway_count_tokens.go b/backend/internal/service/openai_gateway_count_tokens.go index 7518a6073a..46b5a3843a 100644 --- a/backend/internal/service/openai_gateway_count_tokens.go +++ b/backend/internal/service/openai_gateway_count_tokens.go @@ -215,7 +215,15 @@ func (s *OpenAIGatewayService) buildInputTokensUpstreamRequest( return nil, err } req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) - req.Header.Set("authorization", "Bearer "+token) + authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token) + if err != nil { + return nil, err + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } req.Header.Set("content-type", "application/json") req.Header.Set("accept", "application/json") diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 53cc06989e..e1c9aea15f 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -487,6 +487,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco var wsResult *OpenAIForwardResult var wsErr error wsLastFailureReason := "" + agentTaskRecoveryTried := false wsPrevResponseRecoveryTried := false wsInvalidEncryptedContentRecoveryTried := false recoverPrevResponseNotFound := func(attempt int) bool { @@ -571,6 +572,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco startTime, attempt, wsLastFailureReason, + &agentTaskRecoveryTried, ) if wsErr == nil { break @@ -578,6 +580,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if c != nil && c.Writer != nil && c.Writer.Written() { break } + var taskRecoveredErr *agentIdentityTaskRecoveredError + if errors.As(wsErr, &taskRecoveredErr) { + continue + } reason, retryable := classifyOpenAIWSReconnectReason(wsErr) if reason != "" { @@ -684,6 +690,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } httpInvalidEncryptedContentRetryTried := false + agentTaskRecoveryTried := false for { // Build upstream request upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx) @@ -719,6 +726,16 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamCode := extractUpstreamErrorCode(respBody) + if !agentTaskRecoveryTried && s.isAgentIdentityAccount(ctx, account) && isAgentIdentityTaskInvalidHTTPResponse(resp.StatusCode, respBody) { + agentTaskRecoveryTried = true + expectedTaskID := account.GetCredential("task_id") + if err := s.recoverAgentIdentityTask(ctx, account, expectedTaskID); err != nil { + return nil, fmt.Errorf("agent identity task recovery failed: %w", err) + } + continue + } + respBody = s.redactAgentIdentitySensitiveBody(ctx, account, respBody) + resp.Body = io.NopCloser(bytes.NewReader(respBody)) if !httpInvalidEncryptedContentRetryTried && resp.StatusCode == http.StatusBadRequest && upstreamCode == "invalid_encrypted_content" { decoded, decodeErr := ensureReqBody() if decodeErr != nil { @@ -869,8 +886,17 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin. } req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) - // Set authentication header - req.Header.Set("authorization", "Bearer "+token) + // Build authentication for this request. Agent Identity signs a fresh + // assertion here; OAuth/PAT/API-key keep their existing Bearer behavior. + authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token) + if err != nil { + return nil, fmt.Errorf("build openai authentication headers: %w", err) + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } // Set headers specific to OAuth accounts (ChatGPT internal API) if account.Type == AccountTypeOAuth { diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 238d359b9e..2ea914b537 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -338,7 +338,15 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough( req.Header.Del("authorization") req.Header.Del("x-api-key") req.Header.Del("x-goog-api-key") - req.Header.Set("authorization", "Bearer "+token) + authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token) + if err != nil { + return nil, fmt.Errorf("build openai authentication headers: %w", err) + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } // OAuth 透传到 ChatGPT internal API 时补齐必要头。 if account.Type == AccountTypeOAuth { @@ -434,6 +442,7 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough( requestBody []byte, ) error { body := s.readUpstreamErrorBody(resp) + body = s.redactAgentIdentitySensitiveBody(ctx, account, body) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) @@ -477,6 +486,7 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough( ) error { MarkResponseCommitted(c) body := s.readUpstreamErrorBody(resp) + body = s.redactAgentIdentitySensitiveBody(ctx, account, body) // cyber_policy：透传账号本就把原始 body 回给客户端（下方 c.Data），此处仅打标记， // 供 handler 事后写风控/邮件。cyber 是上游网络安全策略拦截，不冷却账号， diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 0d03d83dcf..a17a6f6b05 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -356,6 +356,7 @@ type OpenAIGatewayService struct { openaiWSStateStoreOnce sync.Once openaiSchedulerOnce sync.Once openaiWSPassthroughDialerOnce sync.Once + agentIdentityTaskMu sync.Mutex openaiWSPool *openAIWSConnPool openaiWSStateStore OpenAIWSStateStore openaiScheduler OpenAIAccountScheduler @@ -1052,6 +1053,9 @@ func (s *OpenAIGatewayService) GetAccessToken(ctx context.Context, account *Acco } switch account.Type { case AccountTypeOAuth: + if account.IsOpenAIAgentIdentity() { + return "", OpenAIAuthModeAgentIdentity, nil + } if account.Platform == PlatformGrok { if s.grokTokenProvider != nil { accessToken, err := s.grokTokenProvider.GetAccessToken(ctx, account) diff --git a/backend/internal/service/openai_gateway_upstream_errors.go b/backend/internal/service/openai_gateway_upstream_errors.go index 2e661e7417..bc9bf91fef 100644 --- a/backend/internal/service/openai_gateway_upstream_errors.go +++ b/backend/internal/service/openai_gateway_upstream_errors.go @@ -279,6 +279,7 @@ func (s *OpenAIGatewayService) handleErrorResponse( requestedModel ...string, ) (*OpenAIForwardResult, error) { body := s.readUpstreamErrorBody(resp) + body = s.redactAgentIdentitySensitiveBody(ctx, account, body) // cyber_policy 硬阻断：透传上游原始错误体给客户端（不重包成通用 502），不冷却账号。 // 当前请求恒透传（需求1）；标记供 handler 事后写风控/邮件。400 cyber 不可 failover diff --git a/backend/internal/service/openai_images.go b/backend/internal/service/openai_images.go index fc7d37a173..b411cb3104 100644 --- a/backend/internal/service/openai_images.go +++ b/backend/internal/service/openai_images.go @@ -752,7 +752,15 @@ func (s *OpenAIGatewayService) buildOpenAIImagesRequest( return nil, err } req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) - req.Header.Set("Authorization", "Bearer "+token) + authHeaders, err := s.buildOpenAIAuthenticationHeaders(ctx, account, token) + if err != nil { + return nil, fmt.Errorf("build openai authentication headers: %w", err) + } + for key, values := range authHeaders { + for _, value := range values { + req.Header.Add(key, value) + } + } for key, values := range c.Request.Header { if !openaiPassthroughAllowedHeaders[strings.ToLower(key)] { continue diff --git a/backend/internal/service/openai_quota_service.go b/backend/internal/service/openai_quota_service.go index 337f8c1e8f..b8b766fcff 100644 --- a/backend/internal/service/openai_quota_service.go +++ b/backend/internal/service/openai_quota_service.go @@ -10,6 +10,7 @@ import ( "log/slog" "net/http" "strings" + "sync" "time" infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" @@ -118,6 +119,7 @@ type OpenAIQuotaService struct { proxyRepo ProxyRepository tokenProvider *OpenAITokenProvider privacyClientFactory PrivacyClientFactory + agentIdentityTaskMu sync.Mutex } // NewOpenAIQuotaService constructs a quota service. token provider is required — @@ -154,10 +156,14 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (* callCtx, cancel := context.WithTimeout(ctx, openaiQuotaUpstreamTimeout) defer cancel() + quotaHeaders, headerErr := s.buildCodexQuotaHeaders(callCtx, accountID, accessToken, chatGPTAccountID, fedRAMP) + if headerErr != nil { + return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_AUTH_FAILED", "failed to build upstream authentication: %v", headerErr) + } var payload OpenAIQuotaUsage resp, err := client.R(). SetContext(callCtx). - SetHeaders(buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)). + SetHeaders(quotaHeaders). SetSuccessResult(&payload). Get(chatGPTUsageURL) if err != nil { @@ -178,9 +184,14 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (* } func (s *OpenAIQuotaService) queryResetCreditDetails(ctx context.Context, client *req.Client, accessToken, chatGPTAccountID string, fedRAMP bool, accountID int64) []OpenAIRateLimitResetCreditDetail { + quotaHeaders, headerErr := s.buildCodexQuotaHeaders(ctx, accountID, accessToken, chatGPTAccountID, fedRAMP) + if headerErr != nil { + slog.Warn("openai_quota_reset_credit_details_auth_failed", "account_id", accountID, "error", headerErr) + return nil + } resp, err := client.R(). SetContext(ctx). - SetHeaders(buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP)). + SetHeaders(quotaHeaders). Get(chatGPTRateLimitCreditsURL) if err != nil { slog.Warn("openai_quota_reset_credit_details_failed", "account_id", accountID, "error", err) @@ -239,7 +250,10 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) ( callCtx, cancel := context.WithTimeout(ctx, openaiQuotaUpstreamTimeout) defer cancel() - headers := buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP) + headers, headerErr := s.buildCodexQuotaHeaders(callCtx, accountID, accessToken, chatGPTAccountID, fedRAMP) + if headerErr != nil { + return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_AUTH_FAILED", "failed to build upstream authentication: %v", headerErr) + } headers["content-type"] = "application/json" var payload OpenAIQuotaResetResult @@ -271,7 +285,7 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) ( // token via the shared TokenProvider, and resolves the chatgpt-account-id and // proxy URL. Centralized so QueryUsage / ResetCredit share validation. func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID int64) (accessToken, chatGPTAccountID, proxyURL string, fedRAMP bool, err error) { - if s == nil || s.accountRepo == nil || s.tokenProvider == nil || s.privacyClientFactory == nil { + if s == nil || s.accountRepo == nil || s.privacyClientFactory == nil { return "", "", "", false, infraerrors.New(http.StatusInternalServerError, "OPENAI_QUOTA_NOT_CONFIGURED", "openai quota service is not configured") } @@ -309,12 +323,17 @@ func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID return "", "", "", false, infraerrors.New(http.StatusBadRequest, "OPENAI_QUOTA_MISSING_ACCOUNT_ID", "chatgpt_account_id is missing; please re-authorize this account") } - accessToken, err = s.tokenProvider.GetAccessToken(ctx, account) - if err != nil { - return "", "", "", false, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "failed to acquire access token: %v", err) - } - if strings.TrimSpace(accessToken) == "" { - return "", "", "", false, infraerrors.New(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "access token is empty") + if !account.IsOpenAIAgentIdentity() { + if s.tokenProvider == nil { + return "", "", "", false, infraerrors.New(http.StatusInternalServerError, "OPENAI_QUOTA_NOT_CONFIGURED", "openai quota token provider is not configured") + } + accessToken, err = s.tokenProvider.GetAccessToken(ctx, account) + if err != nil { + return "", "", "", false, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "failed to acquire access token: %v", err) + } + if strings.TrimSpace(accessToken) == "" { + return "", "", "", false, infraerrors.New(http.StatusBadGateway, "OPENAI_QUOTA_TOKEN_UNAVAILABLE", "access token is empty") + } } fedRAMP = account.IsChatGPTAccountFedRAMP() @@ -337,6 +356,43 @@ func (s *OpenAIQuotaService) prepareUpstreamCall(ctx context.Context, accountID return accessToken, chatGPTAccountID, proxyURL, fedRAMP, nil } +func (s *OpenAIQuotaService) buildCodexQuotaHeaders(ctx context.Context, accountID int64, accessToken, chatGPTAccountID string, fedRAMP bool) (map[string]string, error) { + headers := buildCodexCommonHeaders(accessToken, chatGPTAccountID, fedRAMP) + if s == nil || s.accountRepo == nil { + return headers, nil + } + account, err := s.accountRepo.GetByID(ctx, accountID) + if err != nil || account == nil { + if strings.TrimSpace(accessToken) == "" { + return nil, fmt.Errorf("agent identity account credentials are unavailable") + } + return headers, nil + } + if account.IsShadow() { + if resolved, resolveErr := resolveCredentialAccount(ctx, s.accountRepo, account); resolveErr == nil && resolved != nil { + account = resolved + } else if strings.TrimSpace(accessToken) == "" { + return nil, fmt.Errorf("agent identity shadow credentials are unavailable") + } + } + if !account.IsOpenAIAgentIdentity() { + return headers, nil + } + if err := ensureAgentIdentityTaskForAccount(ctx, s.accountRepo, nil, &s.agentIdentityTaskMu, account, ""); err != nil { + return nil, err + } + key, err := agentIdentityKeyFromAccount(account) + if err != nil { + return nil, err + } + assertion, err := buildAgentAssertion(key, time.Now()) + if err != nil { + return nil, err + } + headers["authorization"] = assertion + return headers, nil +} + // buildCodexCommonHeaders sets the request headers expected by the chatgpt.com // backend so calls succeed past Cloudflare/WASM checks. func buildCodexCommonHeaders(accessToken, chatGPTAccountID string, fedRAMP bool) map[string]string { diff --git a/backend/internal/service/openai_quota_spark_window_test.go b/backend/internal/service/openai_quota_spark_window_test.go index c56600d3f5..d2b18af05b 100644 --- a/backend/internal/service/openai_quota_spark_window_test.go +++ b/backend/internal/service/openai_quota_spark_window_test.go @@ -2,6 +2,10 @@ package service import ( "context" + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -214,6 +218,45 @@ func TestPrepareUpstreamCallShadowResolve(t *testing.T) { "prepareUpstreamCall should use parent's chatgpt_account_id after shadow resolve") } +func TestQueryUsageAgentIdentityUsesAssertionWithoutOAuthToken(t *testing.T) { + _, privateKey, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + der, err := x509.MarshalPKCS8PrivateKey(privateKey) + require.NoError(t, err) + account := &Account{ + ID: 300, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Credentials: map[string]any{ + "auth_mode": OpenAIAuthModeAgentIdentity, + "agent_runtime_id": "runtime-quota", + "agent_private_key": base64.StdEncoding.EncodeToString(der), + "task_id": "task-quota", + "chatgpt_account_id": "account-quota", + "chatgpt_account_is_fedramp": true, + }, + } + repo := &stubQuotaAccountRepo{accounts: map[int64]*Account{account.ID: account}} + var authorization string + var accountHeader string + var fedrampHeader string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authorization = r.Header.Get("authorization") + accountHeader = r.Header.Get("chatgpt-account-id") + fedrampHeader = r.Header.Get("x-openai-fedramp") + w.Header().Set("content-type", "application/json") + _, _ = w.Write([]byte(`{"plan_type":"pro","rate_limit":{"allowed":true}}`)) + })) + defer srv.Close() + svc := NewOpenAIQuotaService(repo, nil, nil, newQuotaRedirectingFactory(srv)) + usage, err := svc.QueryUsage(context.Background(), account.ID) + require.NoError(t, err) + require.NotNil(t, usage) + require.True(t, strings.HasPrefix(authorization, "AgentAssertion ")) + require.Equal(t, "account-quota", accountHeader) + require.Equal(t, "true", fedrampHeader) +} + func TestParseOpenAIRateLimitResetCreditDetails_CompatibleContainers(t *testing.T) { tests := []struct { name string diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index 45e004b6d0..9a6453892a 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -539,6 +539,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( Account: account, WSURL: wsURL, Headers: wsHeaders, + HeadersFactory: func(factoryCtx context.Context, headers http.Header) (http.Header, error) { + return s.refreshOpenAIAgentIdentityHeaders(factoryCtx, account, headers) + }, ProxyURL: func() string { if account.ProxyID != nil && account.Proxy != nil { return account.Proxy.URL() @@ -602,7 +605,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( acquireTimeout = 30 * time.Second } - acquireTurnLease := func(turn int, preferred string, forcePreferredConn bool) (*openAIWSConnLease, error) { + agentTaskRecoveryTried := false + var acquireTurnLease func(int, string, bool) (*openAIWSConnLease, error) + acquireTurnLease = func(turn int, preferred string, forcePreferredConn bool) (*openAIWSConnLease, error) { req := cloneOpenAIWSAcquireRequest(baseAcquireReq) req.PreferredConnID = strings.TrimSpace(preferred) req.ForcePreferredConn = forcePreferredConn @@ -611,6 +616,14 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( acquireCtx, acquireCancel := context.WithTimeout(ctx, acquireTimeout) lease, acquireErr := pool.Acquire(acquireCtx, req) acquireCancel() + var dialErr *openAIWSDialError + if acquireErr != nil && s.isAgentIdentityAccount(ctx, account) && errors.As(acquireErr, &dialErr) && dialErr != nil && dialErr.StatusCode == http.StatusUnauthorized && !agentTaskRecoveryTried { + agentTaskRecoveryTried = true + if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil { + return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr) + } + return acquireTurnLease(turn, preferred, forcePreferredConn) + } if acquireErr != nil { dialStatus, dialClass, dialCloseStatus, dialCloseReason, dialRespServer, dialRespVia, dialRespCFRay, dialRespReqID := summarizeOpenAIWSDialError(acquireErr) logOpenAIWSModeInfo( diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index a4d47218e7..0aa4adc313 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -67,7 +67,9 @@ func (s *OpenAIGatewayService) buildOpenAIWSHeaders( promptCacheKey string, ) (http.Header, openAIWSSessionHeaderResolution, error) { headers := make(http.Header) - headers.Set("authorization", "Bearer "+token) + if account == nil || !account.IsOpenAIAgentIdentity() { + headers.Set("authorization", "Bearer "+token) + } sessionResolution := resolveOpenAIWSSessionHeaders(c, promptCacheKey) if c != nil && c.Request != nil { diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index f0f71648dd..65a6add7a5 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/backend/internal/service/openai_ws_forwarder_v2.go @@ -30,6 +30,7 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( startTime time.Time, attempt int, lastFailureReason string, + agentTaskRecoveryTried *bool, ) (*OpenAIForwardResult, error) { if s == nil || account == nil { return nil, wrapOpenAIWSFallback("invalid_state", errors.New("service or account is nil")) @@ -174,9 +175,12 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( defer acquireCancel() lease, err := s.getOpenAIWSConnPool().Acquire(acquireCtx, openAIWSAcquireRequest{ - Account: account, - WSURL: wsURL, - Headers: wsHeaders, + Account: account, + WSURL: wsURL, + Headers: wsHeaders, + HeadersFactory: func(factoryCtx context.Context, headers http.Header) (http.Header, error) { + return s.refreshOpenAIAgentIdentityHeaders(factoryCtx, account, headers) + }, PreferredConnID: preferredConnID, ForceNewConn: forceNewConn, ProxyURL: func() string { @@ -187,6 +191,14 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2( }(), }) if err != nil { + var agentDialErr *openAIWSDialError + if s.isAgentIdentityAccount(ctx, account) && errors.As(err, &agentDialErr) && agentDialErr != nil && agentDialErr.StatusCode == http.StatusUnauthorized && agentTaskRecoveryTried != nil && !*agentTaskRecoveryTried { + *agentTaskRecoveryTried = true + if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil { + return nil, fmt.Errorf("agent identity task recovery failed: %w", recoveryErr) + } + return nil, &agentIdentityTaskRecoveredError{} + } dialStatus, dialClass, dialCloseStatus, dialCloseReason, dialRespServer, dialRespVia, dialRespCFRay, dialRespReqID := summarizeOpenAIWSDialError(err) logOpenAIWSModeInfo( "acquire_fail account_id=%d account_type=%s transport=%s reason=%s dial_status=%d dial_class=%s dial_close_status=%s dial_close_reason=%s dial_resp_server=%s dial_resp_via=%s dial_resp_cf_ray=%s dial_resp_x_request_id=%s cause=%s preferred_conn_id=%s force_new_conn=%v ws_host=%s ws_path=%s proxy_enabled=%v", diff --git a/backend/internal/service/openai_ws_pool.go b/backend/internal/service/openai_ws_pool.go index 5950e02841..be81d611f9 100644 --- a/backend/internal/service/openai_ws_pool.go +++ b/backend/internal/service/openai_ws_pool.go @@ -60,9 +60,13 @@ func (e *openAIWSDialError) Unwrap() error { } type openAIWSAcquireRequest struct { - Account *Account - WSURL string - Headers http.Header + Account *Account + WSURL string + Headers http.Header + // HeadersFactory is evaluated inside dialConn. It exists so credentials + // whose authorization is per-dial (Agent Identity) are never cached in + // lastAcquire or delayed prewarm state. + HeadersFactory func(context.Context, http.Header) (http.Header, error) ProxyURL string PreferredConnID string // ForceNewConn: 强制本次获取新连接（避免复用导致连接内续链状态互相污染）。 @@ -517,6 +521,7 @@ type openAIWSAccountPool struct { conns map[string]*openAIWSConn pinnedConns map[string]int creating int + generation uint64 lastCleanupAt time.Time lastAcquire *openAIWSAcquireRequest prewarmActive bool @@ -1283,6 +1288,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) { } var req openAIWSAcquireRequest + generation := uint64(0) need := 0 ap, ok := p.getAccountPool(accountID) if !ok || ap == nil { @@ -1317,6 +1323,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) { return } req = cloneOpenAIWSAcquireRequest(*ap.lastAcquire) + generation = ap.generation ap.prewarmActive = true if cooldown := p.prewarmCooldown(); cooldown > 0 { ap.prewarmUntil = now.Add(cooldown) @@ -1324,7 +1331,7 @@ func (p *openAIWSConnPool) ensureTargetIdleAsync(accountID int64) { ap.creating += need p.metrics.scaleUpTotal.Add(int64(need)) - go p.prewarmConns(accountID, req, need) + go p.prewarmConns(accountID, req, need, generation) } func (p *openAIWSConnPool) targetConnCountLocked(ap *openAIWSAccountPool, maxConns int) int { @@ -1367,7 +1374,11 @@ func (p *openAIWSConnPool) targetConnCountLocked(ap *openAIWSAccountPool, maxCon return target } -func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequest, total int) { +func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequest, total int, generations ...uint64) { + generation := uint64(0) + if len(generations) > 0 { + generation = generations[0] + } defer func() { if ap, ok := p.getAccountPool(accountID); ok && ap != nil { ap.mu.Lock() @@ -1398,6 +1409,11 @@ func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequ ap.mu.Unlock() continue } + if ap.generation != generation || ap.lastAcquire == nil { + ap.mu.Unlock() + conn.close() + continue + } if len(ap.conns) >= p.effectiveMaxConnsByAccount(req.Account) { ap.mu.Unlock() conn.close() @@ -1410,6 +1426,35 @@ func (p *openAIWSConnPool) prewarmConns(accountID int64, req openAIWSAcquireRequ } } +// ClearAccount closes all pooled connections and discards delayed prewarm +// state for one account. The generation guard prevents an in-flight prewarm +// started before credential recovery from re-entering the pool afterwards. +func (p *openAIWSConnPool) ClearAccount(accountID int64) { + if p == nil || accountID <= 0 { + return + } + ap, ok := p.getAccountPool(accountID) + if !ok || ap == nil { + return + } + ap.mu.Lock() + ap.generation++ + conns := make([]*openAIWSConn, 0, len(ap.conns)) + for id, conn := range ap.conns { + delete(ap.conns, id) + delete(ap.pinnedConns, id) + if conn != nil { + conns = append(conns, conn) + } + } + ap.lastAcquire = nil + ap.prewarmUntil = time.Time{} + ap.prewarmFails = 0 + ap.prewarmFailAt = time.Time{} + ap.mu.Unlock() + closeOpenAIWSConns(conns) +} + func (p *openAIWSConnPool) evictConn(accountID int64, connID string) { if p == nil || accountID <= 0 || stringsTrim(connID) == "" { return @@ -1485,7 +1530,15 @@ func (p *openAIWSConnPool) dialConn(ctx context.Context, req openAIWSAcquireRequ if p == nil || p.clientDialer == nil { return nil, errors.New("openai ws client dialer is nil") } - conn, status, handshakeHeaders, err := p.clientDialer.Dial(ctx, req.WSURL, req.Headers, req.ProxyURL) + headers := cloneHeader(req.Headers) + var err error + if req.HeadersFactory != nil { + headers, err = req.HeadersFactory(ctx, headers) + if err != nil { + return nil, err + } + } + conn, status, handshakeHeaders, err := p.clientDialer.Dial(ctx, req.WSURL, headers, req.ProxyURL) if err != nil { return nil, &openAIWSDialError{ StatusCode: status, diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index d822d181cd..01b66c376b 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -359,10 +359,28 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( return errors.New("openai ws passthrough dialer is nil") } - dialCtx, cancelDial := context.WithTimeout(ctx, s.openAIWSDialTimeout()) - defer cancelDial() - upstreamConn, statusCode, handshakeHeaders, err := dialer.Dial(dialCtx, wsURL, headers, proxyURL) - if err != nil { + agentTaskRecoveryTried := false + var upstreamConn openAIWSClientConn + statusCode := 0 + var handshakeHeaders http.Header + for { + headers, err = s.refreshOpenAIAgentIdentityHeaders(ctx, account, headers) + if err != nil { + return fmt.Errorf("refresh ws authentication headers: %w", err) + } + dialCtx, cancelDial := context.WithTimeout(ctx, s.openAIWSDialTimeout()) + upstreamConn, statusCode, handshakeHeaders, err = dialer.Dial(dialCtx, wsURL, headers, proxyURL) + cancelDial() + if err == nil { + break + } + if s.isAgentIdentityAccount(ctx, account) && statusCode == http.StatusUnauthorized && !agentTaskRecoveryTried { + agentTaskRecoveryTried = true + if recoveryErr := s.recoverAgentIdentityTask(ctx, account, account.GetCredential("task_id")); recoveryErr != nil { + return fmt.Errorf("agent identity task recovery failed: %w", recoveryErr) + } + continue + } logOpenAIWSV2Passthrough( "relay_dial_failed account_id=%d status_code=%d err=%s", account.ID,