fix(xai): stop batch import from discarding rotated refresh tokens

prepare_import exchanged refresh tokens entry by entry but aborted the
whole batch on any later failure, dropping every already-rotated token:
the source file's old tokens were consumed upstream, so a re-import hit
invalid_grant and the credentials were unrecoverable. A document containing
the same account twice triggered this deterministically via the duplicate-
subject rejection.

Deduplicate identical refresh tokens at document parse (side-effect-free,
first entry wins) so duplicated entries can no longer invalidate each
other's exchange. In prepare_import, make entries independent: a failed
entry is skipped and the first failure is returned only when no entry
succeeded; a second grant for an already-imported subject is skipped
instead of failing the batch.

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
zyycn
2026-07-26 23:38:05 +08:00
co-authored by Claude
parent c953f6f44d
commit 97b91d3409
3 changed files with 69 additions and 24 deletions
+42 -22
View File
@@ -316,36 +316,56 @@ impl ProviderAdmin for XaiAdminProvider {
.map_err(map_provider_store_error)?; .map_err(map_provider_store_error)?;
let mut subjects = BTreeSet::new(); let mut subjects = BTreeSet::new();
let mut credentials = Vec::new(); let mut credentials = Vec::new();
// 逐条目独立成败:verify 会真实轮换 RT,后续条目的任何失败都不得
// 丢弃已轮换成功的凭据(旧 RT 已被上游作废,丢弃即不可逆报废)。
// 失败条目跳过;仅当没有任何条目成功时返回首个失败。
let mut first_failure: Option<ProviderAdminError> = None;
for entry in document.into_entries() { for entry in document.into_entries() {
let name = entry.name().to_owned(); let name = entry.name().to_owned();
let email = entry.email().map(str::to_owned); let email = entry.email().map(str::to_owned);
let tokens = self let tokens = match self
.oauth .oauth
.verify_imported_credential(&discovery, entry.into_candidate()) .verify_imported_credential(&discovery, entry.into_candidate())
.await .await
.map_err(|error| map_failure_class(error.class()))?; {
Ok(tokens) => tokens,
Err(error) => {
first_failure.get_or_insert_with(|| map_failure_class(error.class()));
continue;
}
};
// 同一账号的另一份 grant:账号已由更早条目入册,本条目跳过。
if !subjects.insert(tokens.evidence().subject().to_owned()) { if !subjects.insert(tokens.evidence().subject().to_owned()) {
return Err(provider_error(ProviderAdminErrorKind::Invalid)); continue;
} }
let prepared = GrokCredentialAdmin let account = VerifiedGrokAccount {
.prepare_verified_account( account_id: ProviderAccountId::new(format!("acct_{}", Uuid::now_v7().simple()))
&VerifiedGrokAccount { .map_err(|_| provider_error(ProviderAdminErrorKind::Internal))?,
account_id: ProviderAccountId::new(format!( name,
"acct_{}", email,
Uuid::now_v7().simple() upstream_account_id: None,
)) plan_type: None,
.map_err(|_| provider_error(ProviderAdminErrorKind::Internal))?, tokens,
name, enabled: true,
email, };
upstream_account_id: None, let prepared = match GrokCredentialAdmin.prepare_verified_account(&account, policy) {
plan_type: None, Ok(prepared) => prepared,
tokens, Err(error) => {
enabled: true, first_failure.get_or_insert_with(|| map_repository_error(error));
}, continue;
policy, }
) };
.map_err(map_repository_error)?; match prepared_create(prepared, Utc::now()) {
credentials.push(prepared_create(prepared, Utc::now())?); Ok(credential) => credentials.push(credential),
Err(error) => {
first_failure.get_or_insert(error);
}
}
}
if credentials.is_empty() {
return Err(
first_failure.unwrap_or_else(|| provider_error(ProviderAdminErrorKind::Invalid))
);
} }
Ok(PreparedCredentialImport { Ok(PreparedCredentialImport {
provider_kind: self.provider_kind.clone(), provider_kind: self.provider_kind.clone(),
@@ -172,6 +172,12 @@ impl GrokOAuthImportDocument {
entries.push(entry); entries.push(entry);
} }
} }
// 同一 refresh token 的重复条目只保留首个:RT exchange 会作废旧
// grant,重复条目逐条交换会让先完成的轮换被后续条目作废。
let mut seen_refresh_tokens = HashSet::new();
entries.retain(|entry| {
seen_refresh_tokens.insert(entry.candidate.refresh_token.expose().to_owned())
});
if entries.is_empty() { if entries.is_empty() {
return Err(GrokOAuthImportError::InvalidField("document")); return Err(GrokOAuthImportError::InvalidField("document"));
} }
@@ -283,7 +283,7 @@ fn minimal_oauth_credential_should_not_require_a_source_format() {
} }
#[test] #[test]
fn oauth_account_document_should_not_require_unique_source_names() { fn duplicate_refresh_token_entries_should_collapse_to_the_first() {
let mut document: serde_json::Value = let mut document: serde_json::Value =
serde_json::from_slice(&oauth_account_document()).expect("fixture JSON"); serde_json::from_slice(&oauth_account_document()).expect("fixture JSON");
let duplicate = document["accounts"][0].clone(); let duplicate = document["accounts"][0].clone();
@@ -292,10 +292,29 @@ fn oauth_account_document_should_not_require_unique_source_names() {
.expect("accounts array") .expect("accounts array")
.push(duplicate); .push(duplicate);
// 同一 refresh token 重复出现时解析层去重,避免逐条 RT exchange
// 让先完成的轮换被后续重复条目作废。
let parsed = GrokOAuthImportDocument::parse_json( let parsed = GrokOAuthImportDocument::parse_json(
&serde_json::to_vec(&document).expect("serialize fixture"), &serde_json::to_vec(&document).expect("serialize fixture"),
) )
.expect("upstream identity verification owns duplicate detection"); .expect("duplicate refresh tokens collapse at parse");
assert_eq!(parsed.into_entries().len(), 1);
// 不同 refresh token 的同名条目不受影响,重复检测仍归上游身份验证。
let mut document: serde_json::Value =
serde_json::from_slice(&oauth_account_document()).expect("fixture JSON");
let mut distinct = document["accounts"][0].clone();
distinct["credentials"]["refresh_token"] =
serde_json::Value::String("fixture-refresh-token-second".to_owned());
document["accounts"]
.as_array_mut()
.expect("accounts array")
.push(distinct);
let parsed = GrokOAuthImportDocument::parse_json(
&serde_json::to_vec(&document).expect("serialize fixture"),
)
.expect("distinct refresh tokens keep their entries");
assert_eq!(parsed.into_entries().len(), 2); assert_eq!(parsed.into_entries().len(), 2);
} }