fix(openai): write oauth recovery to app logs

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-06 14:22:13 +08:00
co-authored by Codex
parent d0dc7ef92b
commit 98652d4c68
17 changed files with 631 additions and 36 deletions
+1
View File
@@ -2628,6 +2628,7 @@ dependencies = [
"tokio-tungstenite 0.28.0",
"tokio-util",
"tracing",
"tracing-subscriber",
"tungstenite 0.27.0",
"url",
"uuid",
+33 -1
View File
@@ -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<LogGuard, LogError> {
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<LogGuard, LogError>
Ok(LogGuard { _writers: guards })
}
fn file_filter(directive: &str) -> Result<EnvFilter, LogError> {
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,
@@ -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]]
@@ -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(())
}
@@ -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::<Utc>::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::<Utc>::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<String>,
refresh_token: Option<String>,
) -> Result<(CodexOAuthSecret, Option<DateTime<Utc>>), 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::<Utc>::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,
@@ -8,6 +8,7 @@ mod cookie;
mod identity;
mod oauth;
mod quota;
mod recovery_log;
mod refresh;
mod repository;
mod security;
@@ -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::<Utc>::from)
@@ -661,6 +669,7 @@ impl CodexOAuthAdminService {
&self,
pending: &CodexPendingAuthorization,
callback_url: &str,
fallback_refresh_token: Option<SecretString>,
) -> 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<Option<SecretString>, CodexOAuthAdminError> {
let runtime = CodexCredentialCodec::decode(&current.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,
@@ -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"
);
}
@@ -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::<Utc>::from)
@@ -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<TokenPair, ()> {
}
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),
})
}
@@ -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)?,
@@ -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([
@@ -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<Mutex<Vec<u8>>>);
static OAUTH_RECOVERY_LOG_CAPTURE_LOCK: AsyncMutex<()> = AsyncMutex::const_new(());
pub(super) struct OAuthRecoveryLogWriter(Arc<Mutex<Vec<u8>>>);
impl Write for OAuthRecoveryLogWriter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
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;
@@ -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<AuthorizationTokenSet, AuthorizationCodeExchangeError> {
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<CodexAuthenticatedAccount, CodexIdentityVerificationError> {
Err(CodexIdentityVerificationError::Unavailable)
}
}
fn service() -> CodexOAuthAdminService {
service_with_store(Arc::new(MemoryAccountStore::default()))
}
@@ -209,10 +246,18 @@ fn service_with_store(store: Arc<MemoryAccountStore>) -> CodexOAuthAdminService
fn service_with_account_source(
store: Arc<MemoryAccountStore>,
account_source: Arc<dyn CodexAuthenticatedAccountSource>,
) -> CodexOAuthAdminService {
service_with_exchanger(store, account_source, Arc::new(Exchanger))
}
fn service_with_exchanger(
store: Arc<MemoryAccountStore>,
account_source: Arc<dyn CodexAuthenticatedAccountSource>,
exchanger: Arc<dyn AuthorizationCodeExchanger>,
) -> 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);
}
@@ -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;
@@ -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;
+9 -3
View File
@@ -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_<version>_<os>_<arch>.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`。