From 98652d4c68d70c986077aa8a2f023fc53109c40f Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 6 Aug 2026 14:22:13 +0800 Subject: [PATCH] fix(openai): write oauth recovery to app logs Co-authored-by: Codex --- backend/Cargo.lock | 1 + backend/crates/gateway-host/src/logging.rs | 34 +++- backend/crates/providers/openai/Cargo.toml | 1 + backend/crates/providers/openai/src/config.rs | 6 +- .../providers/openai/src/credential/admin.rs | 64 +++++-- .../providers/openai/src/credential/mod.rs | 1 + .../providers/openai/src/credential/oauth.rs | 45 ++++- .../openai/src/credential/recovery_log.rs | 58 +++++++ .../openai/src/credential/refresh.rs | 24 ++- .../openai/src/credential/token_client.rs | 8 +- backend/crates/providers/openai/src/lib.rs | 1 + .../openai/tests/credential/admin.rs | 76 ++++++++ .../providers/openai/tests/credential/mod.rs | 69 ++++++++ .../openai/tests/credential/oauth.rs | 163 +++++++++++++++++- .../openai/tests/credential/refresh.rs | 82 +++++++++ .../openai/tests/credential/token_client.rs | 22 +++ deploy/README.md | 12 +- 17 files changed, 631 insertions(+), 36 deletions(-) create mode 100644 backend/crates/providers/openai/src/credential/recovery_log.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index c965814c3..a02be1235 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2628,6 +2628,7 @@ dependencies = [ "tokio-tungstenite 0.28.0", "tokio-util", "tracing", + "tracing-subscriber", "tungstenite 0.27.0", "url", "uuid", diff --git a/backend/crates/gateway-host/src/logging.rs b/backend/crates/gateway-host/src/logging.rs index 2c782e845..0f704ff76 100644 --- a/backend/crates/gateway-host/src/logging.rs +++ b/backend/crates/gateway-host/src/logging.rs @@ -13,6 +13,8 @@ use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::Subscriber use crate::config::LoggingConfig; const LOG_FILE_PREFIX: &str = "codex-proxy-rs"; +/// OpenAI OAuth 恢复记录含原始 AT/RT，文件日志开启时不能被常规级别过滤掉。 +const OPENAI_OAUTH_RECOVERY_LOG_TARGET: &str = "openai_oauth_recovery"; /// non-blocking 日志 writer 的进程级守卫。 pub struct LogGuard { @@ -34,7 +36,7 @@ pub enum LogError { /// 按自然日、单文件大小、保留天数和文件总数初始化结构化日志。 pub fn initialize_logging(config: &LoggingConfig) -> Result { let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone()); - let file_filter = EnvFilter::try_new(&directive).map_err(|_| LogError::InvalidFilter)?; + let file_filter = file_filter(&directive)?; let stdout_filter = EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled)) .map_err(|_| LogError::InvalidFilter)?; @@ -97,6 +99,15 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result Ok(LogGuard { _writers: guards }) } +fn file_filter(directive: &str) -> Result { + let recovery_directive = format!("{OPENAI_OAUTH_RECOVERY_LOG_TARGET}=info") + .parse() + .expect("static OpenAI OAuth recovery log directive is valid"); + EnvFilter::try_new(directive) + .map(|filter| filter.add_directive(recovery_directive)) + .map_err(|_| LogError::InvalidFilter) +} + fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String { if !persistent_log_enabled { directive.to_owned() @@ -105,6 +116,27 @@ fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> Str } } +#[cfg(test)] +mod tests { + use tracing_subscriber::prelude::*; + + use super::{OPENAI_OAUTH_RECOVERY_LOG_TARGET, file_filter}; + + #[test] + fn oauth_recovery_file_filter_overrides_global_log_level() { + let subscriber = tracing_subscriber::registry() + .with(file_filter("off,openai_oauth_recovery=off").expect("valid global filter")); + + tracing::subscriber::with_default(subscriber, || { + assert!(tracing::enabled!( + target: OPENAI_OAUTH_RECOVERY_LOG_TARGET, + tracing::Level::INFO + )); + assert!(!tracing::enabled!(target: "other", tracing::Level::INFO)); + }); + } +} + struct RotatingLogWriter { directory: PathBuf, maximum_bytes: u64, diff --git a/backend/crates/providers/openai/Cargo.toml b/backend/crates/providers/openai/Cargo.toml index 2aae438d8..9040b73f2 100644 --- a/backend/crates/providers/openai/Cargo.toml +++ b/backend/crates/providers/openai/Cargo.toml @@ -58,6 +58,7 @@ zstd = "0.13" [dev-dependencies] tempfile = "3.27.0" tokio = { version = "1.52.3", default-features = false, features = ["rt-multi-thread", "test-util"] } +tracing-subscriber = { version = "0.3.23", features = ["json"] } wiremock = "0.6.5" [[test]] diff --git a/backend/crates/providers/openai/src/config.rs b/backend/crates/providers/openai/src/config.rs index 9dc45fe6f..5e9b83d40 100644 --- a/backend/crates/providers/openai/src/config.rs +++ b/backend/crates/providers/openai/src/config.rs @@ -43,10 +43,8 @@ impl OpenAiConfig { self.quota.validate()?; self.auth.validate()?; self.wire_profile.validate()?; - self.identity_secret_path = source_dir - .parent() - .unwrap_or(source_dir) - .join(".runtime/data/identity_hmac_secret"); + let project_root = source_dir.parent().unwrap_or(source_dir); + self.identity_secret_path = project_root.join(".runtime/data/identity_hmac_secret"); Ok(()) } diff --git a/backend/crates/providers/openai/src/credential/admin.rs b/backend/crates/providers/openai/src/credential/admin.rs index 8d808d3f1..e54ca76c4 100644 --- a/backend/crates/providers/openai/src/credential/admin.rs +++ b/backend/crates/providers/openai/src/credential/admin.rs @@ -28,6 +28,7 @@ use thiserror::Error; use super::identity::{ CodexAuthenticatedAccountSource, CodexIdentityExpectation, CodexIdentityVerificationError, }; +use super::recovery_log::{CodexOAuthRecoveryOperation, record_oauth_recovery}; use super::security::CodexCredentialCodec; use super::token_client::{RefreshFailure, TokenRefresher}; use super::types::{ @@ -729,13 +730,6 @@ impl CodexCredentialAdminService { .refresh(refresh_token.expose_secret()) .await .map_err(map_refresh_failure)?; - if tokens.access_token.trim().is_empty() || tokens.expires_in.is_zero() { - return Err(CodexCredentialAdminError::InvalidCredential); - } - let access_token_expires_at = SystemTime::now() - .checked_add(tokens.expires_in) - .map(DateTime::::from) - .ok_or(CodexCredentialAdminError::InvalidCredential)?; let secret = CodexOAuthSecret { access_token: SecretString::from(tokens.access_token), refresh_token: tokens @@ -744,6 +738,18 @@ impl CodexCredentialAdminService { .or_else(|| oauth.refresh_token.clone()), id_token: oauth.id_token.clone(), }; + Self::record_recovery_log( + CodexOAuthRecoveryOperation::ManualRefresh, + Some(account_id.as_str()), + &secret, + ); + if secret.access_token.expose_secret().trim().is_empty() || tokens.expires_in.is_zero() { + return Err(CodexCredentialAdminError::InvalidCredential); + } + let access_token_expires_at = SystemTime::now() + .checked_add(tokens.expires_in) + .map(DateTime::::from) + .ok_or(CodexCredentialAdminError::InvalidCredential)?; let next_refresh_at = refresh_time( policy, &account_id, @@ -793,8 +799,14 @@ impl CodexCredentialAdminService { if !account_ids.insert(account_id.clone()) { return Err(CodexCredentialAdminError::InvalidInput); } + let typed_account_id = ProviderAccountId::new(account_id.clone()) + .map_err(|_| CodexCredentialAdminError::InvalidInput)?; let (secret, access_token_expires_at) = self - .resolve_import_tokens(access_token.clone(), refresh_token.clone()) + .resolve_import_tokens( + &typed_account_id, + access_token.clone(), + refresh_token.clone(), + ) .await?; let mut verified_account = complete_oauth_account_profile( self.account_source.as_ref(), @@ -823,8 +835,7 @@ impl CodexCredentialAdminService { }; let next_refresh_at = refresh_time( refresh_policy, - &ProviderAccountId::new(account_id.clone()) - .map_err(|_| CodexCredentialAdminError::InvalidInput)?, + &typed_account_id, verified_account.access_token_expires_at, secret.refresh_token.is_some(), )?; @@ -877,6 +888,7 @@ impl CodexCredentialAdminService { async fn resolve_import_tokens( &self, + account_id: &ProviderAccountId, access_token: Option, refresh_token: Option, ) -> Result<(CodexOAuthSecret, Option>), CodexCredentialAdminError> { @@ -894,6 +906,11 @@ impl CodexCredentialAdminService { refresh_token: refresh_token.map(SecretString::from), id_token: None, }; + Self::record_recovery_log( + CodexOAuthRecoveryOperation::ImportDirect, + Some(account_id.as_str()), + &secret, + ); return Ok((secret, None)); } let refresh_token = refresh_token.ok_or(CodexCredentialAdminError::InvalidCredential)?; @@ -902,9 +919,6 @@ impl CodexCredentialAdminService { .refresh(&refresh_token) .await .map_err(map_refresh_failure)?; - if tokens.access_token.trim().is_empty() || tokens.expires_in.is_zero() { - return Err(CodexCredentialAdminError::InvalidCredential); - } let secret = CodexOAuthSecret { access_token: SecretString::from(tokens.access_token), // RT 未轮换时仍保留导入提供的 RT，保证本次补全不会丢失后续刷新能力。 @@ -914,6 +928,14 @@ impl CodexCredentialAdminService { .or_else(|| Some(SecretString::from(refresh_token))), id_token: None, }; + Self::record_recovery_log( + CodexOAuthRecoveryOperation::ImportRefreshToken, + Some(account_id.as_str()), + &secret, + ); + if secret.access_token.expose_secret().trim().is_empty() || tokens.expires_in.is_zero() { + return Err(CodexCredentialAdminError::InvalidCredential); + } let access_token_expires_at = SystemTime::now() .checked_add(tokens.expires_in) .map(DateTime::::from) @@ -921,6 +943,22 @@ impl CodexCredentialAdminService { Ok((secret, Some(access_token_expires_at))) } + fn record_recovery_log( + operation: CodexOAuthRecoveryOperation, + account_id: Option<&str>, + secret: &CodexOAuthSecret, + ) { + record_oauth_recovery( + operation, + account_id, + secret.access_token.expose_secret(), + secret + .refresh_token + .as_ref() + .map(ExposeSecret::expose_secret), + ); + } + fn prepare_agent_identity_import( &self, candidate: &ParsedCodexImportAccount, diff --git a/backend/crates/providers/openai/src/credential/mod.rs b/backend/crates/providers/openai/src/credential/mod.rs index 63bdb98d6..6d24c5e69 100644 --- a/backend/crates/providers/openai/src/credential/mod.rs +++ b/backend/crates/providers/openai/src/credential/mod.rs @@ -8,6 +8,7 @@ mod cookie; mod identity; mod oauth; mod quota; +mod recovery_log; mod refresh; mod repository; mod security; diff --git a/backend/crates/providers/openai/src/credential/oauth.rs b/backend/crates/providers/openai/src/credential/oauth.rs index 8001e2a56..707e3e541 100644 --- a/backend/crates/providers/openai/src/credential/oauth.rs +++ b/backend/crates/providers/openai/src/credential/oauth.rs @@ -17,7 +17,7 @@ use gateway_admin::model::{ }, }; use gateway_core::engine::credential::{ - NewProviderAccount, ProviderAccountId, ProviderAccountStore, + LoadedCredential, NewProviderAccount, ProviderAccountId, ProviderAccountStore, }; use gateway_core::provider_ports::ProviderRuntimePolicyPort; use secrecy::{ExposeSecret, SecretString}; @@ -31,6 +31,8 @@ use super::admin::{ PreparedCodexCredentialRotation, complete_oauth_account_profile, refresh_time, }; use super::identity::{CodexAuthenticatedAccountSource, CodexIdentityVerificationError}; +use super::recovery_log::{CodexOAuthRecoveryOperation, record_oauth_recovery}; +use super::security::CodexCredentialCodec; use super::token_client::{ AuthorizationCodeExchangeError, AuthorizationCodeExchanger, AuthorizationCodeGrant, OFFICIAL_CODEX_OAUTH_CLIENT_ID, OFFICIAL_CODEX_REDIRECT_URI, @@ -489,9 +491,6 @@ impl CodexOAuthAdminService { callback_url: &str, ) -> Result<(PendingAuthorizationMutation, CompletedCodexOAuthCredential), CodexOAuthAdminError> { - let (mut secret, id_token, expires_in) = - self.exchange_pending(&pending, callback_url).await?; - let mutation = pending.mutation.clone(); let current = if let Some(target) = pending.reauthorization() { let current = self .store @@ -502,6 +501,15 @@ impl CodexOAuthAdminService { } else { None }; + let fallback_refresh_token = current + .as_ref() + .map(current_oauth_refresh_token) + .transpose()? + .flatten(); + let (mut secret, id_token, expires_in) = self + .exchange_pending(&pending, callback_url, fallback_refresh_token) + .await?; + let mutation = pending.mutation.clone(); let access_token_expires_at = SystemTime::now() .checked_add(expires_in) .map(DateTime::::from) @@ -661,6 +669,7 @@ impl CodexOAuthAdminService { &self, pending: &CodexPendingAuthorization, callback_url: &str, + fallback_refresh_token: Option, ) -> Result<(super::types::CodexOAuthSecret, SecretString, Duration), CodexOAuthAdminError> { let (code, callback_state) = callback_parts(callback_url)?; @@ -678,10 +687,36 @@ impl CodexOAuthAdminService { }) .await .map_err(map_exchange_error)?; - Ok((tokens.secret, tokens.id_token, tokens.expires_in)) + let mut secret = tokens.secret; + if secret.refresh_token.is_none() { + secret.refresh_token = fallback_refresh_token; + } + record_oauth_recovery( + CodexOAuthRecoveryOperation::AuthorizationCode, + pending + .reauthorization() + .map(|account_id| account_id.as_str()), + secret.access_token.expose_secret(), + secret + .refresh_token + .as_ref() + .map(ExposeSecret::expose_secret), + ); + Ok((secret, tokens.id_token, tokens.expires_in)) } } +fn current_oauth_refresh_token( + current: &LoadedCredential, +) -> Result, CodexOAuthAdminError> { + let runtime = CodexCredentialCodec::decode(¤t.credential) + .map_err(|_| CodexOAuthAdminError::Credential)?; + Ok(runtime + .authentication + .oauth() + .and_then(|oauth| oauth.refresh_token.clone())) +} + fn authorization_url( pending: &CodexPendingAuthorization, oauth_client_id: &str, diff --git a/backend/crates/providers/openai/src/credential/recovery_log.rs b/backend/crates/providers/openai/src/credential/recovery_log.rs new file mode 100644 index 000000000..6177b46f8 --- /dev/null +++ b/backend/crates/providers/openai/src/credential/recovery_log.rs @@ -0,0 +1,58 @@ +//! 成功取得 OpenAI OAuth token 后写入普通应用日志的恢复记录。 +//! +//! 此日志会刻意保存原始 AT/RT，避免后续数据库写入或校验失败时，已成功 +//! 交换的账号无法恢复。记录直接进入现有结构化日志，沿用其滚动和清理策略。 + +const RECOVERY_LOG_TARGET: &str = "openai_oauth_recovery"; + +/// 产生可恢复 token 组的 OAuth 操作。 +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum CodexOAuthRecoveryOperation { + /// 直接提供 AT/RT 的导入。 + ImportDirect, + /// 仅提供 RT 的导入，先交换得到 AT。 + ImportRefreshToken, + /// 管理员手动触发的 RT 交换。 + ManualRefresh, + /// 调度器触发的 RT 交换。 + ScheduledRefresh, + /// 授权码加 PKCE 的 token 交换。 + AuthorizationCode, +} + +impl CodexOAuthRecoveryOperation { + const fn as_str(self) -> &'static str { + match self { + Self::ImportDirect => "import_direct", + Self::ImportRefreshToken => "import_refresh_token", + Self::ManualRefresh => "manual_refresh", + Self::ScheduledRefresh => "scheduled_refresh", + Self::AuthorizationCode => "authorization_code", + } + } +} + +/// 将已取得的 AT/RT 写入普通结构化日志。 +/// +/// 调用方必须在任何可能拒绝响应的本地校验、画像补全或数据库写入之前调用。 +/// 该调用不执行额外 I/O，也不返回业务错误。Host 负责普通日志文件的分割、清理和 +/// 输出；日志系统不可用时按其既有策略处理，不能阻断 OAuth 业务流程。 +pub(crate) fn record_oauth_recovery( + operation: CodexOAuthRecoveryOperation, + account_id: Option<&str>, + access_token: &str, + refresh_token: Option<&str>, +) { + let has_refresh_token = refresh_token.is_some(); + let refresh_token = refresh_token.unwrap_or_default(); + tracing::info!( + target: RECOVERY_LOG_TARGET, + event = "openai_oauth_recovery", + operation = operation.as_str(), + account_id = account_id.unwrap_or_default(), + access_token, + refresh_token, + has_refresh_token, + "OpenAI OAuth recovery token record" + ); +} diff --git a/backend/crates/providers/openai/src/credential/refresh.rs b/backend/crates/providers/openai/src/credential/refresh.rs index eb5013172..6538404bc 100644 --- a/backend/crates/providers/openai/src/credential/refresh.rs +++ b/backend/crates/providers/openai/src/credential/refresh.rs @@ -17,6 +17,7 @@ use secrecy::{ExposeSecret, SecretString}; use thiserror::Error; use super::identity::CodexAccountIdentityVerifier; +use super::recovery_log::{CodexOAuthRecoveryOperation, record_oauth_recovery}; use super::repository::{CodexCredentialRepository, CredentialRepositoryError}; use super::token_client::{RefreshFailure, TokenPair, TokenRefresher}; use super::types::CodexOAuthSecret; @@ -387,13 +388,6 @@ impl CodexCredentialRefreshService { refresh_token: rotated_refresh_token, expires_in, } = tokens; - if access_token.is_empty() || expires_in.is_zero() { - return Err(CodexCredentialRefreshError::InvalidRefreshResponse); - } - let observed_at = SystemTime::now(); - let access_token_expires_at = observed_at - .checked_add(expires_in) - .ok_or(CodexCredentialRefreshError::InvalidRefreshResponse)?; let refresh_token = rotated_refresh_token .map(SecretString::from) .or(due.secret.refresh_token); @@ -402,6 +396,22 @@ impl CodexCredentialRefreshService { refresh_token, id_token: due.secret.id_token, }; + record_oauth_recovery( + CodexOAuthRecoveryOperation::ScheduledRefresh, + Some(due.account.id().as_str()), + secret.access_token.expose_secret(), + secret + .refresh_token + .as_ref() + .map(ExposeSecret::expose_secret), + ); + if secret.access_token.expose_secret().is_empty() || expires_in.is_zero() { + return Err(CodexCredentialRefreshError::InvalidRefreshResponse); + } + let observed_at = SystemTime::now(); + let access_token_expires_at = observed_at + .checked_add(expires_in) + .ok_or(CodexCredentialRefreshError::InvalidRefreshResponse)?; let next_refresh_at = policy .next_attempt_at(due.account.id(), access_token_expires_at, observed_at) .map(DateTime::::from) diff --git a/backend/crates/providers/openai/src/credential/token_client.rs b/backend/crates/providers/openai/src/credential/token_client.rs index 91cbcb424..696486521 100644 --- a/backend/crates/providers/openai/src/credential/token_client.rs +++ b/backend/crates/providers/openai/src/credential/token_client.rs @@ -258,7 +258,7 @@ impl AuthorizationCodeExchanger for OpenAiTokenClient { || tokens .refresh_token .as_deref() - .is_some_and(|token| token.is_empty()) + .is_some_and(|token| token.trim().is_empty()) { return Err(AuthorizationCodeExchangeError::Rejected); } @@ -305,7 +305,11 @@ fn parse_token_pair(body: &[u8]) -> Result { } Ok(TokenPair { access_token: tokens.access_token, - refresh_token: tokens.refresh_token, + // OAuth 刷新响应可能省略未变更的 RT。将空字段同样视为省略，调用方 + // 才能保留当前仍可使用的 token。 + refresh_token: tokens + .refresh_token + .filter(|refresh_token| !refresh_token.trim().is_empty()), expires_in: Duration::from_secs(tokens.expires_in), }) } diff --git a/backend/crates/providers/openai/src/lib.rs b/backend/crates/providers/openai/src/lib.rs index d9955c21e..0cf49b202 100644 --- a/backend/crates/providers/openai/src/lib.rs +++ b/backend/crates/providers/openai/src/lib.rs @@ -143,6 +143,7 @@ pub async fn initialize( .with_session_identity(session_identity), ); + tracing::info!(target: "openai_oauth_recovery", "OpenAI OAuth recovery records use normal application logs"); let token_client = Arc::new( credential::token_client::openai_token_client(config.token_client_config()) .map_err(|_| OpenAiInitializeError::TokenClient)?, diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index 44df4991d..f833bed3c 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/backend/crates/providers/openai/tests/credential/admin.rs @@ -29,6 +29,7 @@ use provider_openai::credential::{ }; use secrecy::ExposeSecret; +use super::{OAuthRecoveryLogCapture, lock_oauth_recovery_log_capture}; use crate::support::{MemoryAccountStore, codex_account, profile, runtime_policy, secret}; fn import(id: &str, token: &str) -> ImportCodexOAuthCredential { @@ -522,6 +523,7 @@ fn refreshed_tokens() -> TokenPair { #[tokio::test] async fn manual_refresh_prepares_rotation_without_identity_verification() { + let _capture_lock = lock_oauth_recovery_log_capture().await; let (store, refresher, leases, service) = manual_refresh_fixture(Ok(refreshed_tokens()), true).await; let account_id = ProviderAccountId::new("acct_manual_refresh").expect("account id"); @@ -529,10 +531,14 @@ async fn manual_refresh_prepares_rotation_without_identity_verification() { .load_current_credential(&account_id) .await .expect("current credential"); + let recovery_log = OAuthRecoveryLogCapture::default(); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); let prepared = service .manual_refresh(current) .await .expect("manual refresh"); + drop(guard); assert_eq!(prepared.credential.account_id(), &account_id); assert_eq!(prepared.credential.expected_revision().get(), 1); @@ -571,6 +577,12 @@ async fn manual_refresh_prepares_rotation_without_identity_verification() { ["rt-old-access"] ); assert_eq!(leases.drops.load(Ordering::SeqCst), 0); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "manual_refresh"); + assert_eq!(record["fields"]["account_id"], "acct_manual_refresh"); + assert_eq!(record["fields"]["access_token"], "refreshed-access"); + assert_eq!(record["fields"]["refresh_token"], "rotated-refresh"); + assert_eq!(record["fields"]["has_refresh_token"], true); drop(prepared); assert_eq!(leases.drops.load(Ordering::SeqCst), 2); } @@ -697,6 +709,37 @@ async fn oauth_import_accepts_opaque_access_token_and_completes_profile_from_acc ); } +#[tokio::test] +async fn direct_import_logs_raw_tokens_before_account_profile_failure() { + let _capture_lock = lock_oauth_recovery_log_capture().await; + let recovery_log = OAuthRecoveryLogCapture::default(); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); + let error = import_service_with_source(unused_import_refresher(), account_source([])) + .prepare_import_document(serde_json::json!({ + "sourceFormat": "cpr", + "accounts": [{ + "token": "logged-direct-access", + "refreshToken": "logged-direct-refresh" + }] + })) + .await + .expect_err("profile source fails after direct import logging"); + drop(guard); + + assert_eq!(error, CodexCredentialAdminError::AccountProfileUnavailable); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "import_direct"); + assert!( + record["fields"]["account_id"] + .as_str() + .is_some_and(|account_id| account_id.starts_with("acct_")) + ); + assert_eq!(record["fields"]["access_token"], "logged-direct-access"); + assert_eq!(record["fields"]["refresh_token"], "logged-direct-refresh"); + assert_eq!(record["fields"]["has_refresh_token"], true); +} + #[tokio::test] async fn refresh_token_only_import_persists_exchanged_tokens_and_completes_profile() { let refreshed_access = "opaque-refreshed-import"; @@ -768,6 +811,39 @@ async fn refresh_token_only_import_persists_exchanged_tokens_and_completes_profi ); } +#[tokio::test] +async fn refresh_token_only_import_logs_fallback_refresh_token_before_account_profile_failure() { + let _capture_lock = lock_oauth_recovery_log_capture().await; + let recovery_log = OAuthRecoveryLogCapture::default(); + let refresher = Arc::new(ManualRefresher { + outcome: Mutex::new(Some(Ok(TokenPair { + access_token: "logged-import-access".to_owned(), + refresh_token: None, + expires_in: Duration::from_secs(3_600), + }))), + seen: Mutex::new(Vec::new()), + }); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); + let error = import_service_with_source(refresher, account_source([])) + .prepare_import_document(serde_json::json!({ + "refreshToken": "logged-import-original-refresh" + })) + .await + .expect_err("profile source fails after refresh-token import logging"); + drop(guard); + + assert_eq!(error, CodexCredentialAdminError::AccountProfileUnavailable); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "import_refresh_token"); + assert_eq!(record["fields"]["access_token"], "logged-import-access"); + assert_eq!( + record["fields"]["refresh_token"], + "logged-import-original-refresh" + ); + assert_eq!(record["fields"]["has_refresh_token"], true); +} + #[tokio::test] async fn cpr_batch_allows_distinct_users_in_the_same_workspace() { let source = account_source([ diff --git a/backend/crates/providers/openai/tests/credential/mod.rs b/backend/crates/providers/openai/tests/credential/mod.rs index 37f8fcad7..e8d664c7c 100644 --- a/backend/crates/providers/openai/tests/credential/mod.rs +++ b/backend/crates/providers/openai/tests/credential/mod.rs @@ -1,3 +1,72 @@ +use std::io::{self, Write}; +use std::sync::{Arc, Mutex}; + +use serde_json::Value; +use tokio::sync::{Mutex as AsyncMutex, MutexGuard}; +use tracing_subscriber::fmt::MakeWriter; + +#[derive(Clone, Default)] +pub(super) struct OAuthRecoveryLogCapture(Arc>>); + +static OAUTH_RECOVERY_LOG_CAPTURE_LOCK: AsyncMutex<()> = AsyncMutex::const_new(()); + +pub(super) struct OAuthRecoveryLogWriter(Arc>>); + +impl Write for OAuthRecoveryLogWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0 + .lock() + .expect("OAuth recovery log buffer lock") + .extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl<'a> MakeWriter<'a> for OAuthRecoveryLogCapture { + type Writer = OAuthRecoveryLogWriter; + + fn make_writer(&'a self) -> Self::Writer { + OAuthRecoveryLogWriter(Arc::clone(&self.0)) + } +} + +impl OAuthRecoveryLogCapture { + pub(super) fn dispatch(&self) -> tracing::Dispatch { + let subscriber = tracing_subscriber::fmt() + .json() + .with_target(true) + .without_time() + .with_writer(self.clone()) + .finish(); + tracing::Dispatch::new(subscriber) + } + + pub(super) fn recovery_record(&self) -> Value { + let bytes = self + .0 + .lock() + .expect("OAuth recovery log buffer lock") + .clone(); + std::str::from_utf8(&bytes) + .expect("JSON log output is UTF-8") + .lines() + .map(|line| serde_json::from_str(line).expect("JSON log record")) + .find(|record: &Value| { + record["target"] == "openai_oauth_recovery" + && record["fields"]["event"] == "openai_oauth_recovery" + }) + .expect("OAuth recovery log record") + } +} + +pub(super) async fn lock_oauth_recovery_log_capture() -> MutexGuard<'static, ()> { + OAUTH_RECOVERY_LOG_CAPTURE_LOCK.lock().await +} + mod admin; mod agent_identity; mod catalog; diff --git a/backend/crates/providers/openai/tests/credential/oauth.rs b/backend/crates/providers/openai/tests/credential/oauth.rs index 14935c93a..a7426a45b 100644 --- a/backend/crates/providers/openai/tests/credential/oauth.rs +++ b/backend/crates/providers/openai/tests/credential/oauth.rs @@ -30,6 +30,7 @@ use secrecy::{ExposeSecret, SecretString}; use sha2::{Digest as _, Sha256}; use url::Url; +use super::{OAuthRecoveryLogCapture, lock_oauth_recovery_log_capture}; use crate::support::{MemoryAccountStore, profile, runtime_policy, secret}; #[derive(Default)] @@ -169,6 +170,29 @@ impl AuthorizationCodeExchanger for Exchanger { } } +struct NoRefreshTokenExchanger; + +#[async_trait] +impl AuthorizationCodeExchanger for NoRefreshTokenExchanger { + async fn exchange_authorization_code( + &self, + grant: AuthorizationCodeGrant, + ) -> Result { + if grant.code.expose_secret() != "authorization-code" { + return Err(AuthorizationCodeExchangeError::Rejected); + } + Ok(AuthorizationTokenSet { + secret: CodexOAuthSecret { + access_token: SecretString::from("oauth-access-without-refresh-token"), + refresh_token: None, + id_token: None, + }, + id_token: SecretString::from("oauth-id-token"), + expires_in: Duration::from_secs(3_600), + }) + } +} + #[derive(Default)] struct StaticAccountSource { calls: AtomicUsize, @@ -198,6 +222,19 @@ impl CodexAuthenticatedAccountSource for StaticAccountSource { } } +struct UnavailableAccountSource; + +#[async_trait] +impl CodexAuthenticatedAccountSource for UnavailableAccountSource { + async fn fetch( + &self, + _secret: &CodexOAuthSecret, + _expectation: &CodexIdentityExpectation, + ) -> Result { + Err(CodexIdentityVerificationError::Unavailable) + } +} + fn service() -> CodexOAuthAdminService { service_with_store(Arc::new(MemoryAccountStore::default())) } @@ -209,10 +246,18 @@ fn service_with_store(store: Arc) -> CodexOAuthAdminService fn service_with_account_source( store: Arc, account_source: Arc, +) -> CodexOAuthAdminService { + service_with_exchanger(store, account_source, Arc::new(Exchanger)) +} + +fn service_with_exchanger( + store: Arc, + account_source: Arc, + exchanger: Arc, ) -> CodexOAuthAdminService { CodexOAuthAdminService::new( Arc::new(PendingStore::default()), - Arc::new(Exchanger), + exchanger, account_source, store, runtime_policy(), @@ -333,6 +378,46 @@ async fn completion_populates_create_profile_from_authenticated_account_source() ); } +#[tokio::test] +async fn authorization_code_profile_failure_is_upstream_unavailable() { + let _capture_lock = lock_oauth_recovery_log_capture().await; + let service = service_with_account_source( + Arc::new(MemoryAccountStore::default()), + Arc::new(UnavailableAccountSource), + ); + let started = started(&service).await; + let state = Url::parse(&started.authorization_url) + .expect("authorization URL") + .query_pairs() + .find_map(|(key, value)| (key == "state").then(|| value.into_owned())) + .expect("state"); + + let recovery_log = OAuthRecoveryLogCapture::default(); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); + let error = service + .complete_authorization(CompleteCodexOAuthAuthorization { + owner_ref: owner_ref(), + flow_id: started.flow_id, + callback_url: SecretString::from(format!( + "http://localhost:1455/auth/callback?code=authorization-code&state={state}" + )), + }) + .await + .expect_err("profile source fails after successful authorization-code exchange"); + drop(guard); + + assert_eq!(error, CodexOAuthAdminError::UpstreamUnavailable); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "authorization_code"); + assert_eq!(record["fields"]["account_id"], ""); + assert_eq!(record["fields"]["access_token"], "oauth-access-token"); + assert_eq!(record["fields"]["refresh_token"], "oauth-refresh-token"); + assert_eq!(record["fields"]["has_refresh_token"], true); + assert!(record["fields"].get("flow_id").is_none()); + assert!(record["fields"].get("id_token").is_none()); +} + #[tokio::test] async fn completion_accepts_parseable_non_localhost_callback_url() { let service = service(); @@ -556,3 +641,79 @@ async fn reauthorization_preserves_existing_principal_without_identity_verificat 2 ); } + +#[tokio::test] +async fn reauthorization_preserves_existing_refresh_token_when_exchange_omits_one() { + let _capture_lock = lock_oauth_recovery_log_capture().await; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_oauth_credential(ImportCodexOAuthCredential { + account_id: "acct_oauth_reauth_fallback".to_owned(), + name: "reauthorize fallback".to_owned(), + secret: secret("old-oauth-access"), + verified_account: profile("chatgpt-oauth"), + next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)), + enabled: true, + }) + .await; + let service = service_with_exchanger( + store, + Arc::new(StaticAccountSource::default()), + Arc::new(NoRefreshTokenExchanger), + ); + let started = service + .start_authorization(StartCodexOAuthAuthorization { + mutation: reauthorization_mutation( + "request-reauth-fallback", + "acct_oauth_reauth_fallback", + ), + }) + .await + .expect("start reauthorization"); + let state = Url::parse(&started.authorization_url) + .expect("authorization URL") + .query_pairs() + .find_map(|(key, value)| (key == "state").then(|| value.into_owned())) + .expect("state"); + + let recovery_log = OAuthRecoveryLogCapture::default(); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); + let prepared = service + .complete_authorization(CompleteCodexOAuthAuthorization { + owner_ref: owner_ref(), + flow_id: started.flow_id, + callback_url: SecretString::from(format!( + "http://localhost:1455/auth/callback?code=authorization-code&state={state}" + )), + }) + .await + .expect("complete reauthorization"); + drop(guard); + let CompletedCodexOAuthCredential::Reauthorize(prepared) = prepared.credential else { + panic!("expected prepared reauthorization"); + }; + let runtime = + provider_openai::credential::CodexCredentialCodec::decode(prepared.credential.credential()) + .expect("prepared credential"); + assert_eq!( + runtime + .authentication + .oauth() + .expect("OAuth credential") + .refresh_token + .as_ref() + .expect("preserved refresh token") + .expose_secret(), + "rt-old-oauth-access" + ); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "authorization_code"); + assert_eq!(record["fields"]["account_id"], "acct_oauth_reauth_fallback"); + assert_eq!( + record["fields"]["access_token"], + "oauth-access-without-refresh-token" + ); + assert_eq!(record["fields"]["refresh_token"], "rt-old-oauth-access"); + assert_eq!(record["fields"]["has_refresh_token"], true); +} diff --git a/backend/crates/providers/openai/tests/credential/refresh.rs b/backend/crates/providers/openai/tests/credential/refresh.rs index 64cff741b..8bc34851a 100644 --- a/backend/crates/providers/openai/tests/credential/refresh.rs +++ b/backend/crates/providers/openai/tests/credential/refresh.rs @@ -17,6 +17,7 @@ use provider_openai::credential::{ }; use secrecy::{ExposeSecret, SecretString}; +use super::{OAuthRecoveryLogCapture, lock_oauth_recovery_log_capture}; use crate::support::{MemoryAccountStore, profile, runtime_policy, secret}; const OAUTH_BACKOFF_ATTEMPTS: u32 = 5; @@ -378,6 +379,87 @@ async fn successful_refresh_uses_redis_lease_and_cas_rotates_plaintext_tokens() ); } +#[tokio::test] +async fn scheduled_refresh_preserves_existing_refresh_token_when_exchange_omits_one() { + let (store, _, service) = setup( + Ok(TokenPair { + access_token: "scheduled-access-without-rotated-refresh".to_owned(), + refresh_token: None, + expires_in: Duration::from_secs(3_600), + }), + true, + ) + .await; + + let outcomes = service.refresh_due().await.expect("scheduled refresh"); + + assert!(matches!( + outcomes.as_slice(), + [CodexCredentialRefreshOutcome::Refreshed { account_id, .. }] if account_id == "acct_refresh" + )); + let account = store.account("acct_refresh").expect("rotated account"); + assert_eq!(account.revision().get(), 2); + let runtime = store + .repository() + .load_runtime_credential(&account) + .await + .expect("rotated credential"); + assert_eq!( + runtime + .authentication + .oauth() + .expect("OAuth credential") + .access_token + .expose_secret(), + "scheduled-access-without-rotated-refresh" + ); + assert_eq!( + runtime + .authentication + .oauth() + .expect("OAuth credential") + .refresh_token + .as_ref() + .expect("preserved refresh token") + .expose_secret(), + "rt-old-access" + ); +} + +#[tokio::test] +async fn invalid_refresh_response_is_logged_before_local_rejection() { + let _capture_lock = lock_oauth_recovery_log_capture().await; + let (_, _, service) = setup( + Ok(TokenPair { + access_token: "rejected-access".to_owned(), + refresh_token: None, + expires_in: Duration::ZERO, + }), + true, + ) + .await; + let recovery_log = OAuthRecoveryLogCapture::default(); + let dispatch = recovery_log.dispatch(); + let guard = tracing::dispatcher::set_default(&dispatch); + + let outcomes = service + .refresh_due() + .await + .expect("refresh cycle completes"); + drop(guard); + + assert!(matches!( + outcomes.as_slice(), + [CodexCredentialRefreshOutcome::Failed { account_id }] if account_id == "acct_refresh" + )); + let record = recovery_log.recovery_record(); + assert_eq!(record["fields"]["operation"], "scheduled_refresh"); + assert_eq!(record["fields"]["account_id"], "acct_refresh"); + assert_eq!(record["fields"]["access_token"], "rejected-access"); + assert_eq!(record["fields"]["refresh_token"], "rt-old-access"); + assert_eq!(record["fields"]["has_refresh_token"], true); +} + #[tokio::test] async fn refreshed_tokens_persist_when_identity_verification_rejects() { let (store, _, _) = setup(Ok(success_tokens()), true).await; diff --git a/backend/crates/providers/openai/tests/credential/token_client.rs b/backend/crates/providers/openai/tests/credential/token_client.rs index 697c782ad..a6ff6a164 100644 --- a/backend/crates/providers/openai/tests/credential/token_client.rs +++ b/backend/crates/providers/openai/tests/credential/token_client.rs @@ -75,6 +75,28 @@ async fn bounded_oauth_response_should_parse_lifetime_and_rotated_token() { assert_eq!(tokens.expires_in, Duration::from_secs(3600)); } +#[tokio::test] +async fn refresh_response_with_blank_rotated_token_keeps_the_existing_token_eligible() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "access-rotated", + "refresh_token": " ", + "expires_in": 3600 + }))) + .expect(1) + .mount(&server) + .await; + + let tokens = client(&server) + .refresh("refresh-initial") + .await + .expect("blank rotation field is treated as absent"); + + assert_eq!(tokens.refresh_token, None); +} + #[tokio::test] async fn refresh_should_exchange_the_official_form_fields() { let server = MockServer::start().await; diff --git a/deploy/README.md b/deploy/README.md index 6c309ea5b..ce0242fcf 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -111,9 +111,10 @@ PostgreSQL 是账号、Client Key、运行设置、请求记录与审计的权 Provider schema 以明文 JSON 保存于 PostgreSQL。Redis 只保存可重建、可过期的协调状态，例如 会话亲和、lease、cooldown、OAuth pending flow 与套餐模型目录 cache。 -备份必须包含 `.runtime/postgres`。若希望保留短期 Redis 状态、OpenAI 会话亲和和更新状态， -同时备份 `.runtime/redis` 与 `.runtime/data`；后两者丢失不会造成账号 credential 丢失，但会使 -缓存和会话锚点重新建立。 +备份必须包含 `.runtime/postgres`。要保留 OpenAI OAuth 的 AT/RT 恢复能力，必须保持 +`host.logging.file.enabled: true` 并备份 `.runtime/logs`；该记录受普通日志的 `retention_days` 和 +`max_files` 约束。若希望保留短期 Redis 状态和会话锚点，也同时备份 `.runtime/redis` 与 +`.runtime/data`。 完整运行时、Provider、revision 与恢复边界见 [架构文档](../docs/architecture.md)。 @@ -202,6 +203,11 @@ Release 必须提供当前 OS/架构的 `codex-proxy-rs___.ta Debian 官方源补 `libpq5` 依赖，不引入第三方 APT 源，也不依赖运行时动态安装。 - 备份暂存目录为 `/app/.runtime/data/backup-staging`，由 `.runtime/data` 卷持久化，权限 `0700`（仅 `cpr` 用户可读写）。部署卷至少预留一个最大数据库归档的空间。 +- OpenAI OAuth 每次成功取得 AT/RT 后，都会在账号资料补全、过期时间计算和数据库写入前，以 + `openai_oauth_recovery` 结构化日志事件写入普通应用日志。事件含原始 AT/RT，完全沿用 + `.runtime/logs` 的按日、按大小分割和保留清理策略；文件日志开启时不会被普通日志级别筛掉。 + 它遵循普通日志现有的非阻塞写入机制，不会阻断导入、授权或刷新。 + `.runtime/logs` 因此属于敏感数据，必须按现有普通日志的访问控制和加密备份策略处理。 - S3/R2 存储、Cron 计划、保留策略与备份记录都保存在 PostgreSQL（`backup_settings` / `backup_records`），备份记录行在删除成功后硬删除，操作历史进入 `admin_audit_events`。