Compare commits

...
3 Commits
Author SHA1 Message Date
ZyyandCodex 5ce36931a8 chore: bump version to 3.2.9
Co-authored-by: Codex <noreply@openai.com>
2026-08-17 11:38:49 +08:00
ZyyandCodex 0d826d9eba fix(openai): align OAuth token refresh
Co-authored-by: Codex <noreply@openai.com>
2026-08-17 11:14:03 +08:00
ZyyandCodex 304242957a fix(accounts): preserve upstream error messages
Co-authored-by: Codex <noreply@openai.com>
2026-08-16 13:28:28 +08:00
13 changed files with 558 additions and 117 deletions
@@ -31,23 +31,51 @@ pub enum ProviderAdminErrorKind {
Internal,
}
/// 不泄漏 OAuth material 与 Provider 响应体的管理错误。
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
/// 不携带 OAuth 请求材料的管理错误。
///
/// `message` 用于当前认证管理请求展示;上游没有标准 `error.message` 时,
/// Provider 可以按其官方契约回退为完整的非成功响应正文。
#[derive(Clone, PartialEq, Eq, thiserror::Error)]
#[error("provider admin operation failed: {kind:?}")]
pub struct ProviderAdminError {
kind: ProviderAdminErrorKind,
message: Option<String>,
}
impl std::fmt::Debug for ProviderAdminError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ProviderAdminError")
.field("kind", &self.kind)
.field("message", &self.message.as_ref().map(|_| "<redacted>"))
.finish()
}
}
impl ProviderAdminError {
#[must_use]
pub const fn new(kind: ProviderAdminErrorKind) -> Self {
Self { kind }
Self {
kind,
message: None,
}
}
#[must_use]
pub fn with_message(mut self, message: impl Into<String>) -> Self {
self.message = Some(message.into());
self
}
#[must_use]
pub const fn kind(&self) -> ProviderAdminErrorKind {
self.kind
}
#[must_use]
pub fn message(&self) -> Option<&str> {
self.message.as_deref()
}
}
/// 一个具体 Provider 对管理控制面提供的解析、验证、上游交互与运行时资源回收能力。
@@ -565,11 +565,7 @@ impl AccountsService for DefaultAccountsService {
let upstream_body = error
.upstream_response()
.map(|response| String::from_utf8_lossy(response.body()).into_owned());
let message = upstream_body
.as_deref()
.filter(|body| !body.is_empty())
.unwrap_or_else(|| error.client_message())
.to_owned();
let message = error.client_message().to_owned();
vec![AccountConnectionTestEvent::Failed {
message,
provider_error_code: error.client_error_code().map(ToOwned::to_owned),
@@ -59,7 +59,11 @@ fn map_provider_error(
ProviderAdminErrorKind::Unavailable => AdminErrorKind::Unavailable,
ProviderAdminErrorKind::Internal => AdminErrorKind::Internal,
};
AdminError::new(kind, format!("{resource} operation failed"))
let message = error
.message()
.map(ToOwned::to_owned)
.unwrap_or_else(|| format!("{resource} operation failed"));
AdminError::new(kind, message)
}
async fn publish_committed(
@@ -60,7 +60,7 @@ pub(super) type EventLog = Arc<Mutex<Vec<&'static str>>>;
pub(super) struct FakeProviderAdmin {
kind: ProviderKind,
events: EventLog,
failure: Mutex<Option<ProviderAdminErrorKind>>,
failure: Mutex<Option<ProviderAdminError>>,
quota_failure: Mutex<Option<ProviderAdminErrorKind>>,
pending: Arc<Mutex<Option<PendingAuthorizationMutation>>>,
retry_authorization_after_abort: Mutex<bool>,
@@ -89,7 +89,12 @@ impl FakeProviderAdmin {
}
pub(super) fn fail_next(&self, kind: ProviderAdminErrorKind) {
*self.failure.lock().expect("provider failure") = Some(kind);
*self.failure.lock().expect("provider failure") = Some(ProviderAdminError::new(kind));
}
fn fail_next_with_message(&self, kind: ProviderAdminErrorKind, message: &str) {
*self.failure.lock().expect("provider failure") =
Some(ProviderAdminError::new(kind).with_message(message));
}
pub(super) fn fail_next_quota(&self, kind: ProviderAdminErrorKind) {
@@ -145,7 +150,7 @@ impl FakeProviderAdmin {
fn require_available(&self) -> Result<(), ProviderAdminError> {
match self.failure.lock().expect("provider failure").take() {
Some(kind) => Err(ProviderAdminError::new(kind)),
Some(error) => Err(error),
None => Ok(()),
}
}
@@ -1650,6 +1655,32 @@ async fn accounts_refresh_provider_failure_should_not_call_store_commit() {
);
}
#[tokio::test]
async fn accounts_refresh_should_preserve_the_provider_failure_message() {
let events = events();
let provider = FakeProviderAdmin::new("openai", events.clone());
let upstream_message = "Your refresh token has already been used.";
provider.fail_next_with_message(ProviderAdminErrorKind::Conflict, upstream_message);
let store = FakeAccountStore::new("openai", events.clone());
let services = accounts_service(provider, store).await;
let error = services
.accounts()
.refresh(
&context("refresh-provider-message"),
ProviderAccountId::new("acct_test").expect("account ID"),
)
.await
.expect_err("Provider preparation must fail");
assert_eq!(error.kind(), gateway_admin::model::AdminErrorKind::Conflict);
assert_eq!(error.to_string(), upstream_message);
assert_eq!(
recorded(&events),
["store.load_account", "provider.prepare_refresh"]
);
}
#[tokio::test]
async fn accounts_refresh_store_failure_should_drop_guard_after_commit_attempt() {
let events = events();
@@ -1119,7 +1119,8 @@ pub struct AccountStateChange {
pub observed_at: SystemTime,
/// 受控错误原因;恢复 Ready 时必须清空。
pub error_reason: Option<AccountErrorReason>,
/// 已脱敏的上游诊断信息;恢复 Ready 时必须清空。
/// 供管理端展示的错误消息;结构化上游失败应保留原始 message,不能写入整个正文。
/// 恢复 Ready 时必须清空。
pub message: Option<String>,
}
+16 -10
View File
@@ -258,7 +258,7 @@ impl ProviderAdmin for OpenAiAdminProvider {
))
.await
.inspect_err(|error| {
log_import_failure("prepare_document", credential_admin_error_code(*error));
log_import_failure("prepare_document", credential_admin_error_code(error));
})
.map_err(map_credential_admin_error)?;
let observed_at = Utc::now();
@@ -1275,31 +1275,37 @@ fn map_store_error(error: gateway_core::error::StoreError) -> ProviderAdminError
fn map_credential_admin_error(error: CodexCredentialAdminError) -> ProviderAdminError {
use CodexCredentialAdminError as Error;
provider_admin_error(match error {
let upstream_message = error.upstream_message().map(ToOwned::to_owned);
let kind = match error {
Error::InvalidInput
| Error::InvalidCredential
| Error::MissingRefreshToken
| Error::RefreshRejected
| Error::AccountBanned => ProviderAdminErrorKind::Invalid,
| Error::RefreshRejected { .. }
| Error::AccountBanned { .. } => ProviderAdminErrorKind::Invalid,
Error::NotFound => ProviderAdminErrorKind::NotFound,
Error::RefreshLeaseUnavailable | Error::RefreshAmbiguous => {
Error::RefreshLeaseUnavailable | Error::RefreshAmbiguous { .. } => {
ProviderAdminErrorKind::Conflict
}
Error::RefreshUnavailable => ProviderAdminErrorKind::Unavailable,
})
};
let error = provider_admin_error(kind);
match upstream_message {
Some(message) => error.with_message(message),
None => error,
}
}
const fn credential_admin_error_code(error: CodexCredentialAdminError) -> &'static str {
const fn credential_admin_error_code(error: &CodexCredentialAdminError) -> &'static str {
match error {
CodexCredentialAdminError::InvalidInput => "invalid_input",
CodexCredentialAdminError::InvalidCredential => "invalid_credential",
CodexCredentialAdminError::NotFound => "not_found",
CodexCredentialAdminError::MissingRefreshToken => "missing_refresh_token",
CodexCredentialAdminError::RefreshLeaseUnavailable => "refresh_lease_unavailable",
CodexCredentialAdminError::RefreshRejected => "refresh_rejected",
CodexCredentialAdminError::AccountBanned => "account_banned",
CodexCredentialAdminError::RefreshRejected { .. } => "refresh_rejected",
CodexCredentialAdminError::AccountBanned { .. } => "account_banned",
CodexCredentialAdminError::RefreshUnavailable => "refresh_unavailable",
CodexCredentialAdminError::RefreshAmbiguous => "refresh_ambiguous",
CodexCredentialAdminError::RefreshAmbiguous { .. } => "refresh_ambiguous",
}
}
@@ -376,7 +376,7 @@ impl fmt::Debug for PreparedCodexCredentialRotationGuard {
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum CodexCredentialAdminError {
#[error("Codex account input is invalid")]
InvalidInput,
@@ -389,13 +389,30 @@ pub enum CodexCredentialAdminError {
#[error("Codex refresh lease is unavailable")]
RefreshLeaseUnavailable,
#[error("Codex refresh token was rejected")]
RefreshRejected,
RefreshRejected { message: Option<String> },
#[error("Codex account is banned")]
AccountBanned,
AccountBanned { message: Option<String> },
#[error("Codex refresh service is unavailable")]
RefreshUnavailable,
#[error("Codex refresh send state is ambiguous")]
RefreshAmbiguous,
RefreshAmbiguous { message: Option<String> },
}
impl CodexCredentialAdminError {
#[must_use]
pub fn upstream_message(&self) -> Option<&str> {
match self {
Self::RefreshRejected { message }
| Self::AccountBanned { message }
| Self::RefreshAmbiguous { message } => message.as_deref(),
Self::InvalidInput
| Self::InvalidCredential
| Self::NotFound
| Self::MissingRefreshToken
| Self::RefreshLeaseUnavailable
| Self::RefreshUnavailable => None,
}
}
}
/// 无状态的 Codex Admin command preparer。
@@ -800,6 +817,7 @@ impl CodexCredentialAdminService {
.refresher
.refresh(refresh_token.expose_secret())
.await
.inspect_err(|error| log_manual_refresh_failure(&account_id, error))
.map_err(map_refresh_failure)?;
let access_token_rotated = tokens.access_token.is_some();
let secret = CodexOAuthSecret {
@@ -1133,15 +1151,33 @@ fn china_rfc3339(value: DateTime<Utc>) -> String {
.to_rfc3339()
}
const fn map_refresh_failure(error: RefreshFailure) -> CodexCredentialAdminError {
fn map_refresh_failure(error: RefreshFailure) -> CodexCredentialAdminError {
match error {
RefreshFailure::InvalidGrant => CodexCredentialAdminError::RefreshRejected,
RefreshFailure::Banned => CodexCredentialAdminError::AccountBanned,
RefreshFailure::RetryableTransport => CodexCredentialAdminError::RefreshUnavailable,
RefreshFailure::Transport => CodexCredentialAdminError::RefreshAmbiguous,
RefreshFailure::InvalidGrant { message, .. } => {
CodexCredentialAdminError::RefreshRejected { message }
}
RefreshFailure::Banned { message, .. } => {
CodexCredentialAdminError::AccountBanned { message }
}
RefreshFailure::RetryableTransport { .. } => CodexCredentialAdminError::RefreshUnavailable,
RefreshFailure::Transport { message, .. } => {
CodexCredentialAdminError::RefreshAmbiguous { message }
}
}
}
fn log_manual_refresh_failure(account_id: &ProviderAccountId, error: &RefreshFailure) {
let upstream = error.upstream();
tracing::warn!(
account_id = %account_id,
failure_class = error.classification(),
upstream_message = ?error.message(),
upstream_status = ?upstream.map(super::token_client::RefreshUpstreamFailure::status),
upstream_body = ?upstream.map(super::token_client::RefreshUpstreamFailure::body),
"OpenAI OAuth manual refresh failed"
);
}
fn parse_import_document(
payload: &Value,
) -> Result<Vec<ParsedCodexImportAccount>, CodexCredentialAdminError> {
@@ -20,7 +20,7 @@ use thiserror::Error;
use super::recovery_log::{CodexOAuthRecoveryOperation, record_oauth_recovery};
use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
use super::token_client::{RefreshFailure, TokenPair, TokenRefresher};
use super::token_client::{RefreshFailure, RefreshUpstreamFailure, TokenPair, TokenRefresher};
use super::types::CodexOAuthSecret;
const PROVIDER_NAME: &str = "openai";
@@ -91,12 +91,17 @@ fn log_refresh_deferred(
access_token_expires_at: Option<SystemTime>,
attempt: u32,
reason: &'static str,
upstream_message: Option<&str>,
upstream: Option<&RefreshUpstreamFailure>,
retry_at: SystemTime,
) {
tracing::warn!(
account_id = %account_id,
attempt,
reason,
upstream_message = ?upstream_message,
upstream_status = ?upstream.map(RefreshUpstreamFailure::status),
upstream_body = ?upstream.map(RefreshUpstreamFailure::body),
retry_at = %DateTime::<Utc>::from(retry_at),
access_token_expires_at = ?access_token_expires_at.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(access_token_expires_at)
@@ -105,6 +110,26 @@ fn log_refresh_deferred(
);
}
struct RefreshFailureContext<'a> {
reason: &'static str,
message: Option<String>,
upstream: Option<&'a RefreshUpstreamFailure>,
}
impl<'a> RefreshFailureContext<'a> {
const fn new(
reason: &'static str,
message: Option<String>,
upstream: Option<&'a RefreshUpstreamFailure>,
) -> Self {
Self {
reason,
message,
upstream,
}
}
}
fn refresh_due_at(
account: &ProviderAccount,
policy: ProviderRefreshPolicy,
@@ -257,7 +282,7 @@ impl CodexCredentialRefreshService {
&due.account,
CredentialState::Expired,
AccountErrorReason::CredentialExpired,
"refresh_recovery_window_exhausted",
RefreshFailureContext::new("refresh_recovery_window_exhausted", None, None),
CodexCredentialRefreshOutcome::Invalidated { account_id },
)
.await;
@@ -293,29 +318,33 @@ impl CodexCredentialRefreshService {
.ok_or(CodexCredentialRefreshError::InvalidRefreshResponse)?;
match self.refresher.refresh(refresh_token.expose_secret()).await {
Ok(tokens) => self.persist_success(due, tokens).await,
Err(RefreshFailure::InvalidGrant) => {
Err(RefreshFailure::InvalidGrant { message, upstream }) => {
self.persist_terminal(
&due.account,
CredentialState::Expired,
AccountErrorReason::CredentialExpired,
"refresh_invalid_grant",
RefreshFailureContext::new(
"refresh_invalid_grant",
message,
upstream.as_deref(),
),
CodexCredentialRefreshOutcome::Invalidated { account_id },
)
.await
}
Err(RefreshFailure::Banned) => {
Err(RefreshFailure::Banned { message, upstream }) => {
self.persist_terminal(
&due.account,
CredentialState::Banned,
AccountErrorReason::AccountBanned,
"account_banned",
RefreshFailureContext::new("account_banned", message, upstream.as_deref()),
CodexCredentialRefreshOutcome::Banned { account_id },
)
.await
}
Err(RefreshFailure::RetryableTransport) => {
Err(RefreshFailure::RetryableTransport { message }) => {
if self
.defer_refresh(&due.account, "transport-not-sent")
.defer_refresh(&due.account, "transport-not-sent", Some(&message), None)
.await?
{
Ok(CodexCredentialRefreshOutcome::Transient { account_id })
@@ -323,11 +352,16 @@ impl CodexCredentialRefreshService {
Ok(CodexCredentialRefreshOutcome::Stale { account_id })
}
}
Err(RefreshFailure::Transport) => {
Err(RefreshFailure::Transport { message, upstream }) => {
// 上游瞬态(429/5xx/超时/畸形响应等,发送状态未知)保留现有凭据、
// 推进退避重试;仅 invalid_grant/banned 才终态失效账号。
if self
.defer_refresh(&due.account, "transport-ambiguous")
.defer_refresh(
&due.account,
"transport-ambiguous",
message.as_deref(),
upstream.as_deref(),
)
.await?
{
Ok(CodexCredentialRefreshOutcome::Transient { account_id })
@@ -483,9 +517,14 @@ impl CodexCredentialRefreshService {
account: &ProviderAccount,
credential_state: CredentialState,
error_reason: AccountErrorReason,
reason: &'static str,
failure: RefreshFailureContext<'_>,
outcome: CodexCredentialRefreshOutcome,
) -> Result<CodexCredentialRefreshOutcome, CodexCredentialRefreshError> {
let RefreshFailureContext {
reason,
message,
upstream,
} = failure;
match self.repository.load_runtime_credential(account).await {
Ok(_) => {}
Err(CredentialRepositoryError::RevisionConflict) => {
@@ -495,6 +534,7 @@ impl CodexCredentialRefreshService {
}
Err(error) => return Err(error.into()),
}
let persisted_message = message.as_deref().unwrap_or(reason).to_owned();
match self
.repository
.apply_state_with_reason(
@@ -502,7 +542,7 @@ impl CodexCredentialRefreshService {
credential_state,
SystemTime::now(),
Some(error_reason),
Some(reason.to_owned()),
Some(persisted_message),
)
.await
{
@@ -511,6 +551,9 @@ impl CodexCredentialRefreshService {
account_id = %account.id(),
?credential_state,
reason,
upstream_message = ?message.as_deref(),
upstream_status = ?upstream.map(RefreshUpstreamFailure::status),
upstream_body = ?upstream.map(RefreshUpstreamFailure::body),
access_token_expires_at = ?account.access_token_expires_at()
.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(account.access_token_expires_at())
@@ -532,6 +575,8 @@ impl CodexCredentialRefreshService {
&self,
account: &ProviderAccount,
reason: &'static str,
upstream_message: Option<&str>,
upstream: Option<&RefreshUpstreamFailure>,
) -> Result<bool, CodexCredentialRefreshError> {
let attempt = self
.credential_state
@@ -552,6 +597,8 @@ impl CodexCredentialRefreshService {
account.access_token_expires_at(),
attempt,
reason,
upstream_message,
upstream,
retry_at,
);
Ok(true)
@@ -3,7 +3,7 @@
use async_trait::async_trait;
use reqwest::{Client, StatusCode, redirect::Policy};
use secrecy::{ExposeSecret, SecretString};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::time::Duration;
@@ -51,16 +51,106 @@ impl fmt::Debug for TokenPair {
}
/// Codex token 刷新的稳定失败分类。
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[derive(Clone, PartialEq, Eq, thiserror::Error)]
pub enum RefreshFailure {
#[error("refresh token is invalid or expired")]
InvalidGrant,
InvalidGrant {
message: Option<String>,
upstream: Option<Box<RefreshUpstreamFailure>>,
},
#[error("account is banned")]
Banned,
Banned {
message: Option<String>,
upstream: Option<Box<RefreshUpstreamFailure>>,
},
#[error("refresh transport failed before server processing")]
RetryableTransport,
RetryableTransport { message: String },
#[error("refresh transport failed after possible server processing")]
Transport,
Transport {
message: Option<String>,
upstream: Option<Box<RefreshUpstreamFailure>>,
},
}
impl fmt::Debug for RefreshFailure {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RefreshFailure")
.field("classification", &self.classification())
.field("message", &self.message().map(|_| "<redacted>"))
.field("upstream", &self.upstream())
.finish()
}
}
impl RefreshFailure {
#[must_use]
pub fn message(&self) -> Option<&str> {
match self {
Self::InvalidGrant { message, .. }
| Self::Banned { message, .. }
| Self::Transport { message, .. } => message.as_deref(),
Self::RetryableTransport { message } => Some(message),
}
}
#[must_use]
pub fn upstream(&self) -> Option<&RefreshUpstreamFailure> {
match self {
Self::InvalidGrant { upstream, .. }
| Self::Banned { upstream, .. }
| Self::Transport { upstream, .. } => upstream.as_deref(),
Self::RetryableTransport { .. } => None,
}
}
#[must_use]
pub const fn classification(&self) -> &'static str {
match self {
Self::InvalidGrant { .. } => "invalid-grant",
Self::Banned { .. } => "account-banned",
Self::RetryableTransport { .. } => "transport-not-sent",
Self::Transport { .. } => "transport-ambiguous",
}
}
}
/// 当前 OAuth 刷新请求收到的完整非成功响应。
///
/// 该值不持久化;`Debug` 不输出正文。调用方仅在受控刷新失败日志中显式记录正文。
#[derive(Clone, PartialEq, Eq)]
pub struct RefreshUpstreamFailure {
status: u16,
body: String,
}
impl RefreshUpstreamFailure {
fn new(status: StatusCode, body: &[u8]) -> Self {
Self {
status: status.as_u16(),
body: String::from_utf8_lossy(body).into_owned(),
}
}
#[must_use]
pub const fn status(&self) -> u16 {
self.status
}
#[must_use]
pub fn body(&self) -> &str {
&self.body
}
}
impl fmt::Debug for RefreshUpstreamFailure {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RefreshUpstreamFailure")
.field("status", &self.status)
.field("body", &"<redacted>")
.finish()
}
}
/// Codex token 刷新端口。
@@ -186,6 +276,51 @@ struct RefreshTokenResponse {
expires_in: Option<u64>,
}
#[derive(Serialize)]
struct RefreshTokenRequest<'a> {
client_id: &'a str,
grant_type: &'static str,
refresh_token: &'a str,
}
#[derive(Deserialize)]
struct RefreshErrorResponse {
error: Option<RefreshError>,
code: Option<String>,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum RefreshError {
Details(RefreshErrorDetails),
Code(String),
}
#[derive(Deserialize)]
struct RefreshErrorDetails {
code: Option<String>,
message: Option<String>,
}
impl RefreshErrorResponse {
fn code(&self) -> Option<&str> {
match self.error.as_ref() {
Some(RefreshError::Details(error)) => error.code.as_deref(),
Some(RefreshError::Code(code)) => Some(code.as_str()),
None => None,
}
.or(self.code.as_deref())
}
fn message(&self) -> Option<String> {
match self.error.as_ref() {
Some(RefreshError::Details(error)) => error.message.as_deref(),
Some(RefreshError::Code(_)) | None => None,
}
.map(ToOwned::to_owned)
}
}
#[derive(Deserialize)]
struct AuthorizationCodeResponse {
access_token: String,
@@ -200,11 +335,11 @@ impl TokenRefresher for OpenAiTokenClient {
let response = self
.client
.post(&self.config.token_endpoint)
.form(&[
("grant_type", "refresh_token"),
("client_id", self.config.client_id.as_str()),
("refresh_token", refresh_token),
])
.json(&RefreshTokenRequest {
client_id: self.config.client_id.as_str(),
grant_type: "refresh_token",
refresh_token,
})
.send()
.await
.map_err(|error| refresh_transport_failure(&error))?;
@@ -212,7 +347,10 @@ impl TokenRefresher for OpenAiTokenClient {
if !status.is_success() {
return Err(classify_refresh_failure(status, &body));
}
parse_token_pair(&body).map_err(|()| RefreshFailure::Transport)
parse_token_pair(&body).map_err(|()| RefreshFailure::Transport {
message: Some("OpenAI OAuth refresh returned an invalid success response".to_owned()),
upstream: None,
})
}
}
@@ -273,13 +411,21 @@ async fn read_bounded_response(
while let Some(chunk) = response
.chunk()
.await
.map_err(|_| RefreshFailure::Transport)?
.map_err(|error| RefreshFailure::Transport {
message: Some(error.to_string()),
upstream: None,
})?
{
let next_len = body
.len()
.checked_add(chunk.len())
.filter(|length| *length <= MAX_OAUTH_RESPONSE_BYTES)
.ok_or(RefreshFailure::Transport)?;
.ok_or_else(|| RefreshFailure::Transport {
message: Some(format!(
"OpenAI OAuth response exceeded {MAX_OAUTH_RESPONSE_BYTES} bytes"
)),
upstream: None,
})?;
body.reserve(next_len.saturating_sub(body.len()));
body.extend_from_slice(&chunk);
}
@@ -312,32 +458,77 @@ fn optional_expiry(value: Option<u64>) -> Option<Duration> {
}
fn classify_refresh_failure(status: StatusCode, body: &[u8]) -> RefreshFailure {
// 官方刷新错误的消息与错误码分别位于 `error.message`、`error.code`;
// `error` 字符串与顶层 `code` 仅用于兼容官方客户端自身的错误码提取契约。
let error = serde_json::from_slice::<RefreshErrorResponse>(body).ok();
let message = Some(refresh_failure_message(error.as_ref(), body));
let upstream = || Some(Box::new(RefreshUpstreamFailure::new(status, body)));
// 5xx 与 429 按状态码判为瞬态:其正文(CDN/网关页等)不是权威 OAuth 错误响应,
// 不据此终态失效账号。
// 不据此终态失效账号,但官方 `error.message` 仍用于本次诊断。
if status.is_server_error() || status == StatusCode::TOO_MANY_REQUESTS {
return RefreshFailure::Transport;
return RefreshFailure::Transport {
message,
upstream: upstream(),
};
}
// 4xx 正文才是权威 OAuth 错误响应,据其区分真过期/封禁与其余瞬态。
let lower = String::from_utf8_lossy(body).to_ascii_lowercase();
if lower.contains("account has been deactivated") {
return RefreshFailure::Banned;
let Some(error) = error else {
return RefreshFailure::Transport {
message,
upstream: upstream(),
};
};
if message.as_deref().is_some_and(|message| {
message
.to_ascii_lowercase()
.contains("account has been deactivated")
}) {
return RefreshFailure::Banned {
message,
upstream: upstream(),
};
}
if lower.contains("invalid_grant")
|| lower.contains("invalid_token")
|| lower.contains("access_denied")
|| lower.contains("refresh_token_expired")
|| lower.contains("refresh_token_reused")
{
return RefreshFailure::InvalidGrant;
let normalized_code = error.code().map(str::to_ascii_lowercase);
if matches!(
normalized_code.as_deref(),
Some(
"invalid_grant"
| "invalid_token"
| "access_denied"
| "refresh_token_expired"
| "refresh_token_reused"
| "refresh_token_invalidated"
)
) {
return RefreshFailure::InvalidGrant {
message,
upstream: upstream(),
};
}
RefreshFailure::Transport
RefreshFailure::Transport {
message,
upstream: upstream(),
}
}
fn refresh_failure_message(error: Option<&RefreshErrorResponse>, body: &[u8]) -> String {
if let Some(message) = error.and_then(RefreshErrorResponse::message) {
return message;
}
if body.is_empty() {
return "Unknown error".to_owned();
}
String::from_utf8_lossy(body).into_owned()
}
fn refresh_transport_failure(error: &reqwest::Error) -> RefreshFailure {
let message = error.to_string();
if is_safe_to_retry_refresh_transport(error) {
RefreshFailure::RetryableTransport
RefreshFailure::RetryableTransport { message }
} else {
RefreshFailure::Transport
RefreshFailure::Transport {
message: Some(message),
upstream: None,
}
}
}
@@ -78,7 +78,21 @@ impl TokenRefresher for SingleUseRefresher {
.lock()
.expect("refresh response lock")
.take()
.ok_or(RefreshFailure::Transport)
.ok_or_else(|| RefreshFailure::Transport {
message: Some("test refresh response is exhausted".to_owned()),
upstream: None,
})
}
}
struct FailingRefresher {
failure: RefreshFailure,
}
#[async_trait]
impl TokenRefresher for FailingRefresher {
async fn refresh(&self, _: &str) -> Result<TokenPair, RefreshFailure> {
Err(self.failure.clone())
}
}
@@ -267,6 +281,51 @@ async fn scheduled_refresh_uses_the_current_margin_without_persisting_a_normal_s
);
}
#[tokio::test]
async fn scheduled_refresh_persists_the_original_upstream_error_message() {
let store = Arc::new(MemoryAccountStore::default());
let policy = MutableRuntimePolicy::new(Duration::from_secs(5 * 60));
let upstream_message = "Refresh token has already been used.";
let service = CodexCredentialRefreshService::new(
store.repository(),
Arc::new(FailingRefresher {
failure: RefreshFailure::InvalidGrant {
message: Some(upstream_message.to_owned()),
upstream: None,
},
}),
Arc::new(RefreshLeases),
Arc::new(RefreshCredentialState),
policy,
);
let account_id = "acct_invalid_refresh_token";
seed_refreshable_account(
&store,
account_id,
SystemTime::now()
.checked_add(Duration::from_secs(120))
.expect("test expiry"),
None,
)
.await;
let outcomes = service.refresh_due().await.expect("refresh cycle");
assert!(matches!(
outcomes.as_slice(),
[CodexCredentialRefreshOutcome::Invalidated {
account_id: invalidated_account_id,
}] if invalidated_account_id == account_id
));
assert_eq!(
store
.account(account_id)
.expect("invalidated account")
.last_error_message(),
Some(upstream_message)
);
}
#[tokio::test]
async fn scheduled_refresh_persists_a_returned_id_token() {
let store = Arc::new(MemoryAccountStore::default());
@@ -46,7 +46,10 @@ async fn oversized_chunked_oauth_response_should_fail_closed_and_redact_body() {
.expect_err("oversized response must fail closed before body classification");
let diagnostic = format!("{failure:?} {failure}");
assert_eq!(failure, RefreshFailure::Transport);
assert!(matches!(
failure,
RefreshFailure::Transport { upstream: None, .. }
));
assert!(!diagnostic.contains(marker));
assert!(!diagnostic.contains("refresh-secret-request-marker"));
}
@@ -95,7 +98,10 @@ async fn refresh_response_should_reject_a_malformed_rotated_id_token() {
.await
.expect_err("malformed ID token must not replace the stored token set");
assert_eq!(failure, RefreshFailure::Transport);
assert!(matches!(
failure,
RefreshFailure::Transport { upstream: None, .. }
));
}
#[tokio::test]
@@ -142,7 +148,7 @@ async fn refresh_response_keeps_the_upstream_rotated_token_verbatim() {
}
#[tokio::test]
async fn refresh_should_exchange_the_official_form_fields() {
async fn refresh_should_exchange_the_official_json_fields() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
@@ -160,11 +166,24 @@ async fn refresh_should_exchange_the_official_form_fields() {
.await
.expect("refresh succeeds");
let requests = server.received_requests().await.expect("received request");
let body = String::from_utf8(requests[0].body.clone()).expect("form body is UTF-8");
let body: serde_json::Value =
serde_json::from_slice(&requests[0].body).expect("JSON request body");
assert!(body.contains("grant_type=refresh_token"));
assert!(body.contains("client_id=test-public-client"));
assert!(body.contains("refresh_token=refresh+secret"));
assert_eq!(
body,
serde_json::json!({
"client_id": "test-public-client",
"grant_type": "refresh_token",
"refresh_token": "refresh secret"
})
);
assert_eq!(
requests[0]
.headers
.get("content-type")
.and_then(|value| value.to_str().ok()),
Some("application/json")
);
}
#[tokio::test]
@@ -273,52 +292,77 @@ async fn authorization_code_exchange_requires_id_token_and_json_response() {
#[tokio::test]
async fn refresh_token_reuse_should_be_classified_as_invalid_grant() {
let upstream_message = "Your refresh token has already been used to generate a new access token. Please try signing in again.";
let failure = refresh_failure(
400,
r#"{"error":"invalid_grant","error_description":"refresh_token_reused"}"#,
r#"{"error":{"message":"Your refresh token has already been used to generate a new access token. Please try signing in again.","type":"invalid_request_error","param":null,"code":"refresh_token_reused"}}"#,
)
.await;
assert_eq!(failure, RefreshFailure::InvalidGrant);
let RefreshFailure::InvalidGrant { message, upstream } = failure else {
panic!("refresh-token reuse must be terminal");
};
assert_eq!(message.as_deref(), Some(upstream_message));
let upstream = upstream.expect("complete upstream failure");
assert_eq!(upstream.status(), 400);
assert!(upstream.body().contains(r#""code":"refresh_token_reused""#));
}
#[tokio::test]
async fn oauth_error_without_nested_message_should_use_the_complete_body() {
let body = r#"{"error":"invalid_grant","error_description":"not the official refresh error message field"}"#;
let failure = refresh_failure(400, body).await;
let diagnostic = format!("{failure:?} {failure}");
let RefreshFailure::InvalidGrant { message, upstream } = failure else {
panic!("invalid_grant must be terminal");
};
assert_eq!(message.as_deref(), Some(body));
assert_eq!(upstream.expect("complete upstream failure").body(), body);
assert!(!diagnostic.contains("error_description"));
}
#[tokio::test]
async fn deactivated_account_should_be_classified_as_banned() {
let failure = refresh_failure(
403,
r#"{"error":"access_denied","error_description":"account has been deactivated"}"#,
r#"{"error":{"code":"access_denied","message":"account has been deactivated"}}"#,
)
.await;
assert_eq!(failure, RefreshFailure::Banned);
let RefreshFailure::Banned { message, upstream } = failure else {
panic!("deactivated account must be banned");
};
assert_eq!(message.as_deref(), Some("account has been deactivated"));
assert_eq!(upstream.expect("complete upstream failure").status(), 403);
}
#[tokio::test]
async fn generic_banned_text_should_not_impersonate_the_deactivation_contract() {
let failure = refresh_failure(403, "account is banned").await;
assert_eq!(failure, RefreshFailure::Transport);
assert_transport_failure(&failure, 403, "account is banned", "account is banned");
}
#[tokio::test]
async fn unregistered_disabled_account_text_should_remain_a_transport_failure() {
let failure = refresh_failure(400, "account disabled").await;
assert_eq!(failure, RefreshFailure::Transport);
assert_transport_failure(&failure, 400, "account disabled", "account disabled");
}
#[tokio::test]
async fn quota_text_should_not_disable_the_oauth_credential() {
let failure = refresh_failure(400, "quota exceeded").await;
assert_eq!(failure, RefreshFailure::Transport);
assert_transport_failure(&failure, 400, "quota exceeded", "quota exceeded");
}
#[tokio::test]
async fn token_revoked_text_without_invalid_grant_should_remain_temporary() {
let failure = refresh_failure(400, "token_revoked").await;
assert_eq!(failure, RefreshFailure::Transport);
assert_transport_failure(&failure, 400, "token_revoked", "token_revoked");
}
#[tokio::test]
@@ -328,17 +372,32 @@ async fn server_error_or_rate_limit_must_stay_transient_even_with_oauth_error_bo
for status in [500, 502, 503, 429] {
let failure = refresh_failure(
status,
r#"{"error":"invalid_grant","error_description":"refresh_token_expired"}"#,
r#"{"error":{"code":"refresh_token_expired","message":"Refresh token expired."}}"#,
)
.await;
assert_eq!(
failure,
RefreshFailure::Transport,
"status {status} must classify as transient"
assert_transport_failure(
&failure,
status,
"Refresh token expired.",
r#"{"error":{"code":"refresh_token_expired","message":"Refresh token expired."}}"#,
);
}
}
fn assert_transport_failure(failure: &RefreshFailure, status: u16, message: &str, body: &str) {
let RefreshFailure::Transport {
message: actual_message,
upstream,
} = failure
else {
panic!("status {status} must classify as transient");
};
assert_eq!(actual_message.as_deref(), Some(message));
let upstream = upstream.as_deref().expect("complete upstream failure");
assert_eq!(upstream.status(), status);
assert_eq!(upstream.body(), body);
}
async fn refresh_failure(status: u16, body: &str) -> RefreshFailure {
let server = MockServer::start().await;
Mock::given(method("POST"))
@@ -42,26 +42,13 @@ interface ConnectionTestEvent {
model?: string
payload?: ConnectionTestRequestPayload
success?: boolean
upstreamBody?: string
upstreamStatus?: number
error?: string
}
function connectionTestErrorText(event: ConnectionTestEvent) {
if (typeof event.upstreamBody === 'string' && event.upstreamBody.length > 0)
return event.upstreamBody
if (Number.isInteger(event.upstreamStatus))
return `HTTP ${event.upstreamStatus}`
return event.error || '测试连接失败'
}
function connectionTestFailureLabel(event: ConnectionTestEvent) {
const hasUpstreamResponse
= (typeof event.upstreamBody === 'string' && event.upstreamBody.length > 0)
|| Number.isInteger(event.upstreamStatus)
return hasUpstreamResponse
? '上游返回错误'
: '未能向上游发起请求'
if (typeof event.error === 'string' && event.error.trim().length > 0)
return event.error
return '测试连接失败'
}
export function useAccountConnectionTest(options: { reload: () => Promise<unknown> }) {
@@ -284,11 +271,7 @@ export function useAccountConnectionTest(options: { reload: () => Promise<unknow
}
if (event.type === 'error') {
connectionTestError.value = connectionTestErrorText(event)
appendConnectionTestLog(
connectionTestFailureLabel(event),
'danger',
connectionTestError.value,
)
appendConnectionTestLog(connectionTestError.value, 'danger')
finishConnectionTest('error')
clearConnectionTestRun()
void options.reload()
+1 -1
View File
@@ -1 +1 @@
version: "3.2.8"
version: "3.2.9"