diff --git a/internal/control/credential_import_batch.go b/internal/control/credential_import_batch.go index 059610f2..97fc5615 100644 --- a/internal/control/credential_import_batch.go +++ b/internal/control/credential_import_batch.go @@ -181,15 +181,16 @@ func (s *Service) ImportCredentialFiles( item.ErrorCode = credentialImportItemError(ctx, classifyCredentialImportError(driver, importErr)) break } - if _, duplicate := identities[credential.Identity()]; duplicate { - item.Status, item.ErrorCode = "skipped", "duplicate_account" - break - } + // 刷新可能补全用户或组织身份，必须按准备完成后的身份判重。 credential, importErr = s.prepareTransientSubscriptionCredential(ctx, channelID, driver, credential) if importErr != nil { item.ErrorCode = credentialImportItemError(ctx, importErr) break } + if _, duplicate := identities[credential.Identity()]; duplicate { + item.Status, item.ErrorCode = "skipped", "duplicate_account" + break + } stage, persistErr := s.persistReadyCredentialStage(ctx, channelID, "oauth_file", credential, network) if persistErr != nil { item.ErrorCode = credentialImportItemError(ctx, persistErr) diff --git a/internal/control/credential_stages.go b/internal/control/credential_stages.go index 3b6984d6..9536886c 100644 --- a/internal/control/credential_stages.go +++ b/internal/control/credential_stages.go @@ -297,7 +297,7 @@ func (s *Service) prepareTransientSubscriptionCredential( } return subscriptionruntime.Credential{}, app_errors.ErrCredentialAuthOutcomeUnknown } - if refreshed.Identity() == "" || refreshed.Identity() != credential.Identity() { + if !subscriptionruntime.RefreshPreservesIdentity(driver, credential, refreshed) { return subscriptionruntime.Credential{}, app_errors.ErrCredentialReauthorizationRequired } return refreshed, nil @@ -393,8 +393,8 @@ func (s *Service) prepareReadySubscriptionStageCredential( } return subscriptionruntime.Credential{}, apiErr } - if refreshed.Identity() == "" || refreshed.Identity() != credential.Identity() || - s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), refreshed.Identity()) != row.IdentityFingerprint { + if !subscriptionruntime.RefreshPreservesIdentity(driver, credential, refreshed) || + s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), credential.Identity()) != row.IdentityFingerprint { if err := s.finishCredentialStageRefreshFailure( ctx, row.ID, @@ -496,7 +496,7 @@ func (s *Service) finishCredentialStageRefresh( "status": models.CredentialStageReady, "encrypted_payload": ciphertext, "payload_schema_version": stagedSubscriptionSchemaV2, "safe_summary_json": models.JSON(summaryJSON), - "identity_fingerprint": row.IdentityFingerprint, + "identity_fingerprint": s.subscriptionIdentityFingerprint(channel.ID(row.ChannelID), credential.Identity()), "expires_at_ms": row.ExpiresAtMS, "error_code": "", "updated_at_ms": s.now().UnixMilli(), }) diff --git a/internal/control/subscription_identity_test.go b/internal/control/subscription_identity_test.go new file mode 100644 index 00000000..4483efc8 --- /dev/null +++ b/internal/control/subscription_identity_test.go @@ -0,0 +1,277 @@ +package control + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "testing" + + "gpt-load/internal/channel" + app_errors "gpt-load/internal/platform/errors" + "gpt-load/internal/storage/models" + subscriptionruntime "gpt-load/internal/subscription/runtime" +) + +func subscriptionIdentityJSON(t *testing.T, channelID channel.ID, account, scope, token, expires string) []byte { + t.Helper() + values := map[string]string{"type": string(channelID), "access_token": "access-" + token, "refresh_token": "refresh-" + token, "email": token + "@example.com", "expired": expires} + if channelID == channel.Codex { + values["account_id"] = account + if scope != "" { + claims, err := json.Marshal(map[string]any{"https://api.openai.com/auth": map[string]string{"chatgpt_account_id": account, "chatgpt_user_id": scope}}) + if err != nil { + t.Fatal(err) + } + values["id_token"] = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".signature" + } + } else { + values["account_uuid"] = account + if scope != "" { + values["organization_uuid"] = scope + } + } + raw, err := json.Marshal(values) + if err != nil { + t.Fatal(err) + } + return raw +} + +func TestSubscriptionConnectionUsesCompleteIdentity(t *testing.T) { + t.Parallel() + for _, test := range []struct { + name string + channel channel.ID + account, scope string + added int + }{ + {"codex different users", channel.Codex, "account-one", "scope-two", 1}, + {"codex different workspaces", channel.Codex, "account-two", "scope-one", 1}, + {"codex reauthorization", channel.Codex, "account-one", "scope-one", 0}, + {"claude different organizations", channel.Claude, "account-one", "scope-two", 1}, + {"claude different users", channel.Claude, "account-two", "scope-one", 1}, + {"claude reauthorization", channel.Claude, "account-one", "scope-one", 0}, + } { + t.Run(test.name, func(t *testing.T) { + for _, authState := range []models.CredentialAuthState{models.CredentialAuthStateReady, models.CredentialAuthStateReauthorizationRequired} { + t.Run(string(authState), func(t *testing.T) { + f := newServiceFixture(t) + importStage := func(account, scope, token string) CredentialStageResult { + result, err := f.service.ImportCredentialStage(t.Context(), test.channel, subscriptionIdentityJSON(t, test.channel, account, scope, token, "2099-01-01T00:00:00Z")) + if err != nil { + t.Fatal(err) + } + return result + } + first := importStage("account-one", "scope-one", "first") + group, err := f.service.CreateGroup(t.Context(), GroupCreateRequest{Name: stringPointer("identity"), ChannelID: test.channel, ConnectionType: models.ConnectionTypeSubscription, Models: optionalGroupModels{Set: true, Values: []GroupModel{}}, StagedCredentialIDs: []string{first.StageID}}) + if err != nil { + t.Fatal(err) + } + var original models.Credential + if err := f.db.Where("group_id = ?", group.GroupID).Take(&original).Error; err != nil { + t.Fatal(err) + } + if err := f.db.Model(&original).Update("auth_state", authState).Error; err != nil { + t.Fatal(err) + } + second := importStage(test.account, test.scope, "second") + result, err := f.service.ConnectGroupCredentials(t.Context(), group.GroupID, []string{second.StageID}) + if err != nil { + t.Fatal(err) + } + wantAdded := test.added + if authState == models.CredentialAuthStateReauthorizationRequired { + wantAdded = 1 + } + if result.CredentialsAdded != wantAdded || result.CredentialsDuplicated != 1-wantAdded { + t.Fatalf("added/duplicated = %d/%d, want %d/%d", result.CredentialsAdded, result.CredentialsDuplicated, wantAdded, 1-wantAdded) + } + var rows []models.Credential + if err := f.db.Where("group_id = ?", group.GroupID).Order("id ASC").Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != 1+test.added { + t.Fatalf("credential count = %d, want %d", len(rows), 1+test.added) + } + if test.added == 1 && rows[0].Data != original.Data { + t.Fatal("another identity replaced the existing credential") + } + }) + } + }) + } +} + +func TestSubscriptionRefreshIdentityAllowsEnrichmentWithoutDowngrade(t *testing.T) { + t.Parallel() + for _, channelID := range []channel.ID{channel.Codex, channel.Claude} { + t.Run(string(channelID), func(t *testing.T) { + for _, test := range []struct { + name, oldScope, newAccount, newScope string + allowed bool + }{ + {"same identity", "scope-one", "account-one", "scope-one", true}, + {"legacy metadata still absent", "", "account-one", "", true}, + {"enriched metadata", "", "account-one", "scope-one", true}, + {"missing metadata", "scope-one", "account-one", "", false}, + {"different scope", "scope-one", "account-one", "scope-two", false}, + {"different account", "scope-one", "account-two", "scope-one", false}, + } { + t.Run(test.name, func(t *testing.T) { + for _, path := range []string{"transient", "ready stage"} { + t.Run(path, func(t *testing.T) { + f := newServiceFixture(t) + driver, ok := subscriptionsDriver(f.service.subscriptions, channelID) + if !ok { + t.Fatal("driver missing") + } + current, err := driver.Parse(subscriptionIdentityJSON(t, channelID, "account-one", test.oldScope, "current", "2000-01-01T00:00:00Z")) + if err != nil { + t.Fatal(err) + } + refreshed, err := driver.Parse(subscriptionIdentityJSON(t, channelID, test.newAccount, test.newScope, "next", "2099-01-01T00:00:00Z")) + if err != nil { + t.Fatal(err) + } + f.service.refreshSubscriptionCredential = func(context.Context, channel.ID, subscriptionruntime.Credential) (subscriptionruntime.Credential, error) { + return refreshed, nil + } + if path == "transient" { + _, err = f.service.prepareTransientSubscriptionCredential(t.Context(), channelID, driver, current) + } else { + stage, stageErr := f.service.persistReadyCredentialStage(t.Context(), channelID, "oauth_file", current) + if stageErr != nil { + t.Fatal(stageErr) + } + row, loadErr := f.service.loadCredentialStage(t.Context(), stage.StageID) + if loadErr != nil { + t.Fatal(loadErr) + } + _, err = f.service.prepareReadySubscriptionStageCredential(t.Context(), row, driver, current, true) + if test.allowed && err == nil { + // 正常刷新后的短期暂存必须仍可消费，不能留下过期的身份指纹。 + _, err = f.service.CreateGroup(t.Context(), GroupCreateRequest{Name: stringPointer("refreshed identity"), ChannelID: channelID, ConnectionType: models.ConnectionTypeSubscription, Models: optionalGroupModels{Set: true, Values: []GroupModel{}}, StagedCredentialIDs: []string{stage.StageID}}) + } + } + if test.allowed && err != nil { + t.Fatalf("same authorization with optional metadata change rejected: %v", err) + } + if !test.allowed && !errors.Is(err, app_errors.ErrCredentialReauthorizationRequired) { + t.Fatalf("changed identity error = %v, want reauthorization required", err) + } + }) + } + }) + } + }) + } +} + +func TestSubscriptionBatchImportSeparatesIdentityScopes(t *testing.T) { + t.Parallel() + for _, channelID := range []channel.ID{channel.Codex, channel.Claude} { + t.Run(string(channelID), func(t *testing.T) { + f := newServiceFixture(t) + items := []json.RawMessage{ + subscriptionIdentityJSON(t, channelID, "account", "scope-one", "first", "2099-01-01T00:00:00Z"), + subscriptionIdentityJSON(t, channelID, "account", "scope-two", "second", "2099-01-01T00:00:00Z"), + subscriptionIdentityJSON(t, channelID, "account", "scope-one", "renewed", "2099-01-01T00:00:00Z"), + } + raw, err := json.Marshal(items) + if err != nil { + t.Fatal(err) + } + result, err := f.service.ImportCredentialBatch(t.Context(), channelID, raw, 0, nil) + if err != nil { + t.Fatal(err) + } + if len(result.Items) != 3 || result.Items[0].Status != "ready" || result.Items[1].Status != "ready" || result.Items[2].ErrorCode != "duplicate_account" { + t.Fatal("batch import did not distinguish scopes and deduplicate the same identity") + } + }) + } +} + +func TestSubscriptionBatchDeduplicatesIdentityAfterRefresh(t *testing.T) { + t.Parallel() + for _, ch := range []channel.ID{channel.Codex, channel.Claude} { + t.Run(string(ch), func(t *testing.T) { + f := newServiceFixture(t) + driver, ok := subscriptionsDriver(f.service.subscriptions, ch) + if !ok { + t.Fatal("driver missing") + } + f.service.refreshSubscriptionCredential = func(_ context.Context, _ channel.ID, current subscriptionruntime.Credential) (subscriptionruntime.Credential, error) { + var fields map[string]any + if err := json.Unmarshal(current.Canonical(), &fields); err != nil { + t.Fatal(err) + } + return driver.Parse(subscriptionIdentityJSON(t, ch, "account", "scope", fields["refresh_token"].(string), "2099-01-01T00:00:00Z")) + } + raw, err := json.Marshal([]json.RawMessage{ + subscriptionIdentityJSON(t, ch, "account", "", "one", "2000-01-01T00:00:00Z"), + subscriptionIdentityJSON(t, ch, "account", "", "two", "2000-01-01T00:00:00Z"), + }) + if err != nil { + t.Fatal(err) + } + result, err := f.service.ImportCredentialBatch(t.Context(), ch, raw, 0, nil) + if err != nil { + t.Fatal(err) + } + if len(result.Items) != 2 || result.Items[0].Status != "ready" || result.Items[1].Status != "skipped" || result.Items[1].ErrorCode != "duplicate_account" { + t.Fatal("batch did not deduplicate the identity enriched by refresh") + } + var count int64 + if err := f.db.Model(&models.CredentialStage{}).Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("persisted stages = %d, want 1", count) + } + }) + } +} + +func TestSubscriptionBatchRefreshesBeforeCoarseIdentityDeduplication(t *testing.T) { + t.Parallel() + for _, ch := range []channel.ID{channel.Codex, channel.Claude} { + t.Run(string(ch), func(t *testing.T) { + f := newServiceFixture(t) + driver, ok := subscriptionsDriver(f.service.subscriptions, ch) + if !ok { + t.Fatal("driver missing") + } + f.service.refreshSubscriptionCredential = func(_ context.Context, _ channel.ID, current subscriptionruntime.Credential) (subscriptionruntime.Credential, error) { + var fields map[string]any + if err := json.Unmarshal(current.Canonical(), &fields); err != nil { + t.Fatal(err) + } + return driver.Parse(subscriptionIdentityJSON(t, ch, "account", "scope", fields["refresh_token"].(string), "2099-01-01T00:00:00Z")) + } + raw, err := json.Marshal([]json.RawMessage{ + subscriptionIdentityJSON(t, ch, "account", "", "one", "2099-01-01T00:00:00Z"), + subscriptionIdentityJSON(t, ch, "account", "", "two", "2000-01-01T00:00:00Z"), + }) + if err != nil { + t.Fatal(err) + } + result, err := f.service.ImportCredentialBatch(t.Context(), ch, raw, 0, nil) + if err != nil { + t.Fatal(err) + } + if len(result.Items) != 2 || result.Items[0].Status != "ready" || result.Items[1].Status != "ready" { + t.Fatal("batch skipped a credential before refresh could complete its identity") + } + var count int64 + if err := f.db.Model(&models.CredentialStage{}).Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 2 { + t.Fatalf("persisted stages = %d, want 2", count) + } + }) + } +} diff --git a/internal/state/loader/subscription_identity_test.go b/internal/state/loader/subscription_identity_test.go new file mode 100644 index 00000000..35840b4d --- /dev/null +++ b/internal/state/loader/subscription_identity_test.go @@ -0,0 +1,81 @@ +package loader_test + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "strings" + "testing" + + "gpt-load/internal/channel" + "gpt-load/internal/platform/encryption" + "gpt-load/internal/state" + "gpt-load/internal/state/loader" + "gpt-load/internal/storage/models" +) + +func TestLoadPreservesLegacySubscriptionDataAndIdentity(t *testing.T) { + t.Parallel() + token := "e30." + base64.RawURLEncoding.EncodeToString([]byte(`{"https://api.openai.com/auth":{"chatgpt_account_id":"account","chatgpt_user_id":"scope"}}`)) + ".signature" + for _, test := range []struct { + channel channel.ID + canonical string + }{ + {channel.Codex, fmt.Sprintf(`{"type":"codex","id_token":%q,"access_token":"access","refresh_token":"refresh","account_id":"account"}`, token)}, + {channel.Claude, fmt.Sprintf(`{"type":"claude","access_token":"access","refresh_token":"refresh","account_uuid":"account","organization_uuid":"scope","claude_device_ids":[%q],"expired":"2099-01-01T00:00:00Z"}`, strings.Repeat("a", 64))}, + } { + t.Run(string(test.channel), func(t *testing.T) { + db := openMigratedDatabase(t) + group := models.Group{Name: "legacy identity", ChannelID: string(test.channel), ConnectionType: models.ConnectionTypeSubscription, Params: models.JSON(`{}`), Models: models.JSON(`[]`), Overrides: models.JSON(`{}`), Enabled: true} + mustCreate(t, db, &group) + crypto, err := encryption.NewService("legacy-identity-test-key-material") + if err != nil { + t.Fatal(err) + } + ciphertext, err := crypto.Encrypt(test.canonical) + if err != nil { + t.Fatal(err) + } + legacyIdentity := crypto.Hash("credential-identity/v1|" + string(test.channel) + "|" + string(test.channel) + "|account") + row := models.Credential{GroupID: group.ID, Data: ciphertext, Fingerprint: crypto.Hash(test.canonical), IdentityFingerprint: legacyIdentity, SecretVersion: 1, AuthState: models.CredentialAuthStateReady, Status: models.CredentialStatusActive} + mustCreate(t, db, &row) + observation := models.CredentialObservation{CredentialID: row.ID, IdentityFingerprint: legacyIdentity, SchemaVersion: 1, ObservationVersion: 1, SnapshotJSON: models.JSON(`{}`), State: models.CredentialObservationFresh} + mustCreate(t, db, &observation) + channels, subscriptions := testSubscriptionRuntime(t) + driver, ok := subscriptions.Driver(test.channel) + if !ok { + t.Fatal("missing driver") + } + parsed, err := driver.Parse([]byte(test.canonical)) + if err != nil { + t.Fatal(err) + } + if parsed.Identity() != "account/scope" { + t.Fatalf("derived identity = %q", parsed.Identity()) + } + registry := state.NewCredentialRegistry() + if err := loader.NewWithCredentialValidation(db, state.NewManager(), registry, channels, subscriptions, crypto).Load(t.Context()); err != nil { + t.Fatal(err) + } + var after models.Credential + if err := db.First(&after, row.ID).Error; err != nil { + t.Fatal(err) + } + if after.Data != row.Data || after.Fingerprint != row.Fingerprint || after.IdentityFingerprint != legacyIdentity || after.SecretVersion != 1 { + t.Fatal("loading rewrote existing subscription data") + } + ref, ok := registry.CredentialRef(row.ID) + expectedGeneration := loader.CredentialIdentityGeneration(legacyIdentity, string(test.channel), "subscription", json.RawMessage(`{}`)) + if !ok || ref.IdentityGeneration != expectedGeneration { + t.Fatal("loading changed the stored identity generation") + } + var savedObservation models.CredentialObservation + if err := db.First(&savedObservation, row.ID).Error; err != nil { + t.Fatal(err) + } + if savedObservation.IdentityFingerprint != after.IdentityFingerprint || savedObservation.State != models.CredentialObservationFresh { + t.Fatal("loading invalidated the existing observation") + } + }) + } +} diff --git a/internal/subscription/credential_identity_test.go b/internal/subscription/credential_identity_test.go new file mode 100644 index 00000000..95149d05 --- /dev/null +++ b/internal/subscription/credential_identity_test.go @@ -0,0 +1,79 @@ +package subscription + +import ( + "context" + "encoding/base64" + "encoding/json" + "testing" + "time" + + "gpt-load/internal/channel" + "gpt-load/internal/storage/models" + "gpt-load/internal/subscription/providers/codex" + subscriptionruntime "gpt-load/internal/subscription/runtime" +) + +func TestCredentialRefreshPreservesStoredIdentityWithOptionalUserClaims(t *testing.T) { + t.Parallel() + for _, test := range []struct { + name, oldUser, newAccount, newUser string + allowed bool + }{ + {"same identity", "user-one", "account-one", "user-one", true}, + {"legacy metadata still absent", "", "account-one", "", true}, + {"enriched metadata", "", "account-one", "user-one", true}, + {"missing metadata", "user-one", "account-one", "", false}, + {"different user", "user-one", "account-one", "user-two", false}, + {"different account", "user-one", "account-two", "user-one", false}, + } { + t.Run(test.name, func(t *testing.T) { + makeCredential := func(account, user, token string, expires time.Time) []byte { + value := codex.Credential{Type: "codex", AccountID: account, AccessToken: "access-" + token, RefreshToken: "refresh-" + token, Expire: expires.UTC().Format(time.RFC3339)} + if user != "" { + claims, err := json.Marshal(map[string]any{"https://api.openai.com/auth": map[string]string{"chatgpt_account_id": account, "chatgpt_user_id": user}}) + if err != nil { + t.Fatal(err) + } + value.IDToken = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".signature" + } + raw, err := codex.MarshalCredential(value) + if err != nil { + t.Fatal(err) + } + return raw + } + manager, db, registry, crypto, row := newCredentialManagerFixture(t, makeCredential("account-one", test.oldUser, "current", time.Now().Add(time.Minute))) + before, ok := registry.CredentialRef(row.ID) + if !ok { + t.Fatal("missing credential") + } + manager.refresh = func(_ context.Context, driver subscriptionruntime.Driver, _ subscriptionruntime.Credential) (subscriptionruntime.Credential, error) { + return driver.Parse(makeCredential(test.newAccount, test.newUser, "next", time.Now().Add(time.Hour))) + } + _, evidence := manager.Prepare(t.Context(), channel.Codex, credentialSnapshot(t, row, crypto), false) + if test.allowed && evidence != nil { + t.Fatalf("refresh rejected optional identity metadata: %s", evidence.Code) + } + if !test.allowed && (evidence == nil || evidence.Code != "refresh_identity_changed") { + t.Fatalf("changed identity evidence = %#v", evidence) + } + var stored models.Credential + if err := db.First(&stored, row.ID).Error; err != nil { + t.Fatal(err) + } + after, ok := registry.CredentialRef(row.ID) + if !ok { + t.Fatal("missing refreshed credential") + } + if stored.IdentityFingerprint != row.IdentityFingerprint || after.IdentityGeneration != before.IdentityGeneration { + t.Fatal("refresh rewrote durable identity or runtime generation") + } + if test.allowed && (stored.SecretVersion != 2 || stored.AuthState != models.CredentialAuthStateReady) { + t.Fatal("refreshed credential was not persisted") + } + if !test.allowed && (stored.Data != row.Data || stored.SecretVersion != 1 || stored.AuthState != models.CredentialAuthStateReauthorizationRequired) { + t.Fatal("changed identity overwrote the credential") + } + }) + } +} diff --git a/internal/subscription/credential_manager.go b/internal/subscription/credential_manager.go index da4d0d5e..f559df68 100644 --- a/internal/subscription/credential_manager.go +++ b/internal/subscription/credential_manager.go @@ -300,7 +300,7 @@ func (manager *CredentialManager) refreshCredentialLocked( } return subscriptionruntime.Credential{}, authEvidence(evidenceCode) } - if refreshed.Identity() == "" || refreshed.Identity() != current.Identity() { + if !subscriptionruntime.RefreshPreservesIdentity(driver, current, refreshed) { if err := manager.transitionAuthState(ctx, row, row.SecretVersion, models.CredentialAuthStateReauthorizationRequired, "refresh_identity_changed"); err != nil { manager.registry.SetCredentialAuthState(row.ID, state.CredentialAuthStateOutcomeUnknown) return subscriptionruntime.Credential{}, localEvidence("refresh_state_commit_failed", "subscription credential state could not be saved") diff --git a/internal/subscription/providers/claude/driver.go b/internal/subscription/providers/claude/driver.go index 7af2bb92..f80817ea 100644 --- a/internal/subscription/providers/claude/driver.go +++ b/internal/subscription/providers/claude/driver.go @@ -58,6 +58,19 @@ func (*claudeDriver) Refresh(ctx context.Context, current subscriptionruntime.Cr return claudeRuntimeCredential(refreshed, canonical), nil } +func (*claudeDriver) MatchesRefreshIdentity(current, refreshed subscriptionruntime.Credential) bool { + before, err := ParseCredentialJSON(current.Canonical()) + if err != nil { + return false + } + after, err := ParseCredentialJSON(refreshed.Canonical()) + if err != nil || before.AccountUUID != after.AccountUUID { + return false + } + // 桥接会保留未返回的旧组织；已经确认的组织不能被清空或替换。 + return before.OrganizationUUID == "" || before.OrganizationUUID == after.OrganizationUUID +} + func (*claudeDriver) ClassifyRefreshFailure(err error) subscriptionruntime.RefreshFailureDecision { var tokenErr *TokenEndpointError if errors.Is(err, ErrCredentialIdentityChanged) || @@ -214,7 +227,7 @@ func claudeRuntimeCredential(value Credential, canonical []byte) subscriptionrun } return subscriptionruntime.NewCredential( canonical, - strings.TrimSpace(value.AccountUUID), + credentialIdentity(value), account, expiresAt, expires, @@ -223,3 +236,11 @@ func claudeRuntimeCredential(value Credential, canonical []byte) subscriptionrun } var _ subscriptionruntime.BrowserAuthorizationDriver = (*claudeDriver)(nil) + +func credentialIdentity(value Credential) string { + accountID := strings.TrimSpace(value.AccountUUID) + if organizationID := strings.TrimSpace(value.OrganizationUUID); organizationID != "" { + return accountID + "/" + organizationID + } + return accountID +} diff --git a/internal/subscription/providers/claude/driver_test.go b/internal/subscription/providers/claude/driver_test.go index 7235a3b2..daed8cf5 100644 --- a/internal/subscription/providers/claude/driver_test.go +++ b/internal/subscription/providers/claude/driver_test.go @@ -26,7 +26,7 @@ func TestClaudeDriverProducesProviderNeutralCredential(t *testing.T) { if err != nil { t.Fatal(err) } - if credential.Identity() != "account-one" || credential.Account().Email != "owner@example.com" || + if credential.Identity() != "account-one/org-one" || credential.Account().Email != "owner@example.com" || !credential.Account().ExpiresAtKnown { t.Fatalf("credential metadata = %q %#v", credential.Identity(), credential.Account()) } diff --git a/internal/subscription/providers/codex/codex.go b/internal/subscription/providers/codex/codex.go index d92e7440..01f26eb9 100644 --- a/internal/subscription/providers/codex/codex.go +++ b/internal/subscription/providers/codex/codex.go @@ -44,7 +44,11 @@ func ParseCredentialJSON(raw []byte) (Credential, error) { if err != nil { return Credential{}, err } - return credentialFromBridge(parsed), nil + value := credentialFromBridge(parsed) + if err := validateCredentialIdentity(value); err != nil { + return Credential{}, err + } + return value, nil } // MarshalCredential returns the canonical persisted representation. diff --git a/internal/subscription/providers/codex/driver.go b/internal/subscription/providers/codex/driver.go index ca353829..f01bb554 100644 --- a/internal/subscription/providers/codex/driver.go +++ b/internal/subscription/providers/codex/driver.go @@ -266,7 +266,7 @@ func codexRuntimeCredential(value Credential, canonical []byte) subscriptionrunt if refreshed, err := time.Parse(time.RFC3339, strings.TrimSpace(value.LastRefresh)); err == nil { account.LastRefresh, account.LastRefreshKnown = refreshed, true } - return subscriptionruntime.NewCredential(canonical, strings.TrimSpace(value.AccountID), account, expiresAt, expires, value.SecretValues()) + return subscriptionruntime.NewCredential(canonical, credentialIdentity(value), account, expiresAt, expires, value.SecretValues()) } var _ subscriptionruntime.BrowserAuthorizationDriver = (*codexDriver)(nil) diff --git a/internal/subscription/providers/codex/identity.go b/internal/subscription/providers/codex/identity.go new file mode 100644 index 00000000..65a128a8 --- /dev/null +++ b/internal/subscription/providers/codex/identity.go @@ -0,0 +1,79 @@ +package codex + +import ( + "encoding/base64" + "encoding/json" + "strings" + + subscriptionruntime "gpt-load/internal/subscription/runtime" +) + +func (*codexDriver) MatchesRefreshIdentity(current, refreshed subscriptionruntime.Credential) bool { + before, err := ParseCredentialJSON(current.Canonical()) + if err != nil { + return false + } + after, err := ParseCredentialJSON(refreshed.Canonical()) + if err != nil || before.AccountID != after.AccountID { + return false + } + beforeUser, afterUser := credentialUserID(before), credentialUserID(after) + // 旧凭据允许补全身份，但不能把已经确认的用户降级为未知用户。 + return beforeUser == "" || beforeUser == afterUser +} + +// 身份只从既有令牌派生，不能改变参与持久化内容指纹计算的 canonical JSON。 +func credentialIdentity(value Credential) string { + accountID := strings.TrimSpace(value.AccountID) + if userID := credentialUserID(value); userID != "" { + return accountID + "/" + userID + } + return accountID +} + +func credentialUserID(value Credential) string { + for _, token := range []string{value.IDToken, value.AccessToken} { + if userID := tokenUserID(token); userID != "" { + return userID + } + } + return "" +} + +// 不能让旧 ID token 掩盖实际用于请求的新 access token 所属用户。 +func validateCredentialIdentity(value Credential) error { + idUser, accessUser := tokenUserID(value.IDToken), tokenUserID(value.AccessToken) + if idUser != "" && accessUser != "" && idUser != accessUser { + return ErrCredentialIdentityChanged + } + return nil +} + +// JWT 仅用于读取已持有凭据的身份元数据，不作为签名或登录认证。 +func tokenUserID(token string) string { + parts := strings.Split(strings.TrimSpace(token), ".") + if len(parts) != 3 { + return "" + } + payload, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + payload, err = base64.URLEncoding.DecodeString(parts[1]) + if err != nil { + return "" + } + } + defer clear(payload) + var claims struct { + Auth struct { + ChatGPTUserID string `json:"chatgpt_user_id"` + UserID string `json:"user_id"` + } `json:"https://api.openai.com/auth"` + } + if json.Unmarshal(payload, &claims) != nil { + return "" + } + if userID := strings.TrimSpace(claims.Auth.ChatGPTUserID); userID != "" { + return userID + } + return strings.TrimSpace(claims.Auth.UserID) +} diff --git a/internal/subscription/providers/codex/identity_test.go b/internal/subscription/providers/codex/identity_test.go new file mode 100644 index 00000000..57c5539f --- /dev/null +++ b/internal/subscription/providers/codex/identity_test.go @@ -0,0 +1,131 @@ +package codex + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "testing" + + subscriptionruntime "gpt-load/internal/subscription/runtime" +) + +func identityTestToken(t *testing.T, claims map[string]string, padded bool) string { + t.Helper() + raw, err := json.Marshal(map[string]any{"https://api.openai.com/auth": claims}) + if err != nil { + t.Fatal(err) + } + encoding := base64.RawURLEncoding + if padded { + encoding = base64.URLEncoding + } + return "e30." + encoding.EncodeToString(raw) + ".signature" +} + +func TestCodexIdentityUsesUserClaimsWithoutChangingCanonical(t *testing.T) { + t.Parallel() + user := identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one"}, false) + alias := identityTestToken(t, map[string]string{"user_id": "user-two"}, true) + for _, test := range []struct{ name, idToken, accessToken, want string }{ + {"id token", user, "access", "workspace/user-one"}, + {"access token", "not-a-jwt", user, "workspace/user-one"}, + {"user id alias", alias, "access", "workspace/user-two"}, + {"access token alias", "", alias, "workspace/user-two"}, + {"claim precedence", identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one", "user_id": "user-two"}, false), "access", "workspace/user-one"}, + {"blank primary claim", identityTestToken(t, map[string]string{"chatgpt_user_id": " ", "user_id": "user-two"}, false), "access", "workspace/user-two"}, + {"missing claims", identityTestToken(t, map[string]string{}, false), "access", "workspace"}, + {"invalid token", "e30.!.signature", "access", "workspace"}, + } { + t.Run(test.name, func(t *testing.T) { + // 固定旧格式的字段顺序，身份推导不得改变 canonical 字节或持久化派生字段。 + raw := []byte(fmt.Sprintf(`{"type":"codex"%s,"access_token":%q,"refresh_token":"refresh","account_id":"workspace","email":"owner@example.com"}`, identityTestIDTokenField(test.idToken), test.accessToken)) + credential, err := newCodexDriver().Parse(raw) + if err != nil { + t.Fatal(err) + } + if string(credential.Canonical()) != string(raw) { + t.Fatal("identity derivation changed canonical credential bytes") + } + if credential.Identity() != test.want { + t.Fatalf("identity = %q, want %q", credential.Identity(), test.want) + } + }) + } +} + +func identityTestIDTokenField(token string) string { + if token == "" { + return "" + } + return fmt.Sprintf(`,"id_token":%q`, token) +} + +func TestCodexIdentityRejectsConflictingTokenUsers(t *testing.T) { + t.Parallel() + for _, accessClaim := range []string{"chatgpt_user_id", "user_id"} { + t.Run(accessClaim, func(t *testing.T) { + value := Credential{Type: "codex", AccountID: "workspace", RefreshToken: "refresh", IDToken: identityTestToken(t, map[string]string{"chatgpt_user_id": "user-one"}, false), AccessToken: identityTestToken(t, map[string]string{accessClaim: "user-two"}, false)} + raw, err := json.Marshal(value) + if err != nil { + t.Fatal(err) + } + if _, err := newCodexDriver().Parse(raw); err == nil { + t.Fatal("conflicting token identities were accepted") + } + if _, err := MarshalCredential(value); err == nil { + t.Fatal("conflicting token identities were canonicalized") + } + }) + } +} + +func TestCodexRefreshWithoutNewIDTokenPreservesKnownIdentity(t *testing.T) { + previous := http.DefaultTransport + t.Cleanup(func() { http.DefaultTransport = previous }) + http.DefaultTransport = targetRoundTripper(func(r *http.Request) (*http.Response, error) { + if r.Method != http.MethodPost || r.URL.String() != "https://auth.openai.com/oauth/token" { + t.Errorf("unexpected refresh request: %s %s", r.Method, r.URL) + } + return &http.Response{ + StatusCode: http.StatusOK, Header: make(http.Header), Request: r, + Body: io.NopCloser(strings.NewReader(`{"access_token":"new-access","refresh_token":"new-refresh","expires_in":3600}`)), + }, nil + }) + userToken := identityTestToken(t, map[string]string{"chatgpt_account_id": "workspace", "chatgpt_user_id": "user-one"}, false) + for _, test := range []struct { + name, idToken, accessToken string + allowed bool + }{ + {"retained ID token", userToken, "old-access", true}, + {"last user claim only in replaced access token", "", userToken, false}, + } { + t.Run(test.name, func(t *testing.T) { + raw, err := MarshalCredential(Credential{Type: "codex", AccountID: "workspace", IDToken: test.idToken, AccessToken: test.accessToken, RefreshToken: "old-refresh"}) + if err != nil { + t.Fatal(err) + } + driver := newCodexDriver() + current, err := driver.Parse(raw) + if err != nil { + t.Fatal(err) + } + refreshed, err := driver.Refresh(t.Context(), current) + if err != nil { + t.Fatal(err) + } + value, err := ParseCredentialJSON(refreshed.Canonical()) + if err != nil { + t.Fatal(err) + } + if value.IDToken != test.idToken || value.AccessToken != "new-access" || value.RefreshToken != "new-refresh" { + t.Fatal("refresh did not retain the old ID token while replacing access and refresh tokens") + } + if allowed := subscriptionruntime.RefreshPreservesIdentity(driver, current, refreshed); allowed != test.allowed { + t.Fatalf("refresh allowed = %v, want %v", allowed, test.allowed) + } + }) + } +} diff --git a/internal/subscription/runtime/runtime.go b/internal/subscription/runtime/runtime.go index 2d83e7ef..4c0d7ca2 100644 --- a/internal/subscription/runtime/runtime.go +++ b/internal/subscription/runtime/runtime.go @@ -57,7 +57,7 @@ func NewCredential(canonical []byte, identity string, account Account, expiresAt // Canonical returns an independent copy suitable for encryption or adapter decoding. func (credential Credential) Canonical() []byte { return append([]byte(nil), credential.canonical...) } -// Identity returns the stable provider account identity used across refreshes. +// Identity returns the provider account identity used for duplicate checks. func (credential Credential) Identity() string { return credential.identity } // Account returns safe display metadata. @@ -157,6 +157,22 @@ type Driver interface { ClassifyRefreshFailure(error) RefreshFailureDecision } +// RefreshIdentityMatcher 由身份包含可选字段的渠道实现；允许补全，但禁止已知身份丢失或变化。 +type RefreshIdentityMatcher interface { + MatchesRefreshIdentity(current, refreshed Credential) bool +} + +// RefreshPreservesIdentity 默认要求身份完全相同；可选字段的规则由渠道负责。 +func RefreshPreservesIdentity(driver Driver, current, refreshed Credential) bool { + if current.Identity() == "" || refreshed.Identity() == "" { + return false + } + if matcher, ok := driver.(RefreshIdentityMatcher); ok { + return matcher.MatchesRefreshIdentity(current, refreshed) + } + return current.Identity() == refreshed.Identity() +} + // CredentialFileImporter is an optional, narrow preprocessing capability for // OAuth file formats that cannot safely provide a canonical credential without // one provider-bound verification step. Existing drivers continue to use Parse.