Add scoped leases for persisted account OAuth grants (#55546)

## What changed

- Expose `AccountOAuthGrantLease` to prevent this auth manager's token refreshes from rotating account credentials during an identity-provider exchange. Releasing the lease returns an `AccountOAuthGrant` that callers can recheck before accepting asynchronous results.
- Validate grants against cached login ownership and persisted credentials, requiring OAuth provenance and complete account identity. Reject stale grants and propagate storage failures, including unavailable keyrings in `Auto` mode.
- Track `login_generation` separately from account ownership and use it to invalidate enterprise MCP authentication. Losing login metadata invalidates the login while preserving same-account network policy; a new login to the same account advances both generations.

## Testing

Extend auth refresh coverage to verify refresh blocking and wakeup on lease release, invalidation after token rotation or persisted login replacement, metadata loss and restoration, and logout while a lease is held.

GitOrigin-RevId: f51b3cd526d7ee94d5dffa16681f8ca332245ce3
This commit is contained in:
jif
2026-10-20 11:48:54 +00:00
committed by copyberry
parent 4be9cfdf30
commit b2aa07db8a
8 changed files with 368 additions and 15 deletions
@@ -51,6 +51,7 @@ async fn queued_notifications_follow_auth_owner_changes(owner_generation: u64) {
generation: 2,
owner_generation,
source_generation: owner_generation,
login_generation: owner_generation,
});
rx.recv().await.unwrap();
send.await;
+4 -4
View File
@@ -73,10 +73,10 @@ impl EmaAuth {
let scope = ema_auth_scope(auth).ok_or(EmaAuthFailure::ReauthenticationRequired)?;
let manager = auth_manager.context("enterprise MCP requires an account owner")?;
let mut changes = manager.auth_change_state_receiver();
let owner_generation = changes.borrow().owner_generation;
let login_generation = changes.borrow().login_generation;
// Never hold the watch guard while reading cached auth: writers take the auth lock first.
if ema_auth_scope(manager.auth_cached().as_ref()).as_ref() != Some(&scope)
|| changes.borrow().owner_generation != owner_generation
|| changes.borrow().login_generation != login_generation
{
bail!(EmaAuthFailure::ReauthenticationRequired);
}
@@ -86,7 +86,7 @@ impl EmaAuth {
let changes = changes.clone();
let policy = policy.clone();
move || {
changes.borrow().owner_generation == owner_generation
changes.borrow().login_generation == login_generation
&& policy.as_ref().is_none_or(|binding| {
binding.read().status == CloudConfigBundleBindingStatus::Current
})
@@ -105,7 +105,7 @@ impl EmaAuth {
} => {}
() = async {
while changes.changed().await.is_ok() {
if changes.borrow().owner_generation != owner_generation {
if changes.borrow().login_generation != login_generation {
break;
}
}
+87
View File
@@ -0,0 +1,87 @@
//! Scoped access to a persisted login grant. Refresh exclusion lasts only through the IdP
//! exchange; callers must recheck ownership before publishing its result. Storage reads
//! observe other processes at these boundaries, not continuously.
use std::fmt;
use std::io;
use std::sync::Arc;
use tokio::sync::OwnedSemaphorePermit;
use super::AuthManager;
use crate::token_data::TokenData;
/// A snapshot of one account grant, checked against both cached ownership and storage.
/// This does not authorize a destination; consumers must validate the recorded issuer/client.
pub struct AccountOAuthGrant {
pub(super) manager: Arc<AuthManager>,
pub(super) tokens: TokenData,
pub(super) login_generation: u64,
}
/// Keeps ordinary refresh from rotating the account token during an IdP exchange.
/// Login and logout remain available and invalidate the grant independently of this lease.
pub struct AccountOAuthGrantLease {
pub(super) grant: AccountOAuthGrant,
pub(super) permit: OwnedSemaphorePermit,
}
impl fmt::Debug for AccountOAuthGrant {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AccountOAuthGrant").finish_non_exhaustive()
}
}
impl fmt::Debug for AccountOAuthGrantLease {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("AccountOAuthGrantLease")
.finish_non_exhaustive()
}
}
impl AccountOAuthGrantLease {
pub fn tokens(&self) -> &TokenData {
&self.grant.tokens
}
/// Release refresh exclusion before contacting a resource authorization server.
pub fn release(self) -> AccountOAuthGrant {
drop(self.permit);
self.grant
}
}
impl AccountOAuthGrant {
pub(super) fn owner_is_current(&self) -> bool {
let Ok(cached) = self.manager.inner.read() else {
return false;
};
self.manager.auth_change_state_tx.borrow().login_generation == self.login_generation
&& cached
.auth
.as_ref()
.filter(|auth| matches!(auth, super::CodexAuth::Chatgpt(_)))
.and_then(|auth| auth.get_token_data().ok())
.is_some_and(|tokens| same_account_grant(&self.tokens, &tokens))
}
/// Check after asynchronous work, before accepting a token or publishing a connection.
/// A storage failure is an error, never proof that the saved grant is still valid.
pub async fn check_current(&self) -> io::Result<bool> {
if !self.owner_is_current() {
return Ok(false);
}
let stored = self.manager.read_account_grant_tokens().await?;
Ok(self.owner_is_current()
&& stored.as_ref().is_some_and(|tokens| {
same_account_grant(&self.tokens, tokens)
&& self.tokens.refresh_token == tokens.refresh_token
}))
}
}
pub(super) fn same_account_grant(a: &TokenData, b: &TokenData) -> bool {
a.oauth_provenance == b.oauth_provenance
&& a.login_owner_id == b.login_owner_id
&& a.account_id == b.account_id
&& a.id_token.chatgpt_user_id == b.id_token.chatgpt_user_id
}
+15 -1
View File
@@ -10,10 +10,13 @@ pub struct AuthChangeState {
/// Advances on login, logout, or changes of user, workspace, or auth mode.
/// A verified legacy-to-OAuth credential migration preserves the same owner.
/// Credential changes with incomplete owner identity also advance this revision.
/// Losing login metadata alone preserves the account owner.
pub owner_generation: u64,
/// Advances on an underlying actor credential/source change, including a new OAuth login.
/// Ordinary refresh within an identifiable OAuth session preserves this revision.
pub source_generation: u64,
/// Preserves strict login ownership for EMA, including loss of login metadata.
pub login_generation: u64,
}
pub(super) fn same_owner(previous: Option<&CodexAuth>, current: Option<&CodexAuth>) -> bool {
@@ -27,8 +30,19 @@ pub(super) fn same_owner(previous: Option<&CodexAuth>, current: Option<&CodexAut
else {
return false;
};
!user.trim().is_empty()
let current_login = login_owner_id(Some(current));
(current_login.is_none() || login_owner_id(Some(previous)) == current_login)
&& !user.trim().is_empty()
&& !workspace.trim().is_empty()
&& current.get_chatgpt_user_id().as_ref() == Some(&user)
&& current.get_account_id().as_ref() == Some(&workspace)
}
pub(super) fn same_login(previous: Option<&CodexAuth>, current: Option<&CodexAuth>) -> bool {
same_owner(previous, current) && login_owner_id(previous) == login_owner_id(current)
}
fn login_owner_id(auth: Option<&CodexAuth>) -> Option<String> {
auth.and_then(|auth| auth.get_token_data().ok())
.and_then(|tokens| tokens.login_owner_id)
}
@@ -59,6 +59,7 @@ fn auth_owner_generation_distinguishes_refreshes_from_identity_changes() {
generation,
owner_generation,
source_generation: generation,
login_generation: owner_generation,
},
);
assert_eq!(*legacy_changes.borrow_and_update(), generation);
@@ -81,6 +82,7 @@ async fn auth_owner_generation_preserves_logout_and_switches_when_coalesced() {
generation: 2,
owner_generation: 2,
source_generation: 2,
login_generation: 2,
},
);
@@ -92,6 +94,7 @@ async fn auth_owner_generation_preserves_logout_and_switches_when_coalesced() {
generation: 4,
owner_generation: 4,
source_generation: 4,
login_generation: 4,
},
);
}
+105 -6
View File
@@ -1,5 +1,11 @@
mod workspace_routing;
#[path = "account_grant.rs"]
mod account_grant;
pub use account_grant::AccountOAuthGrant;
pub use account_grant::AccountOAuthGrantLease;
use account_grant::same_account_grant;
use chrono::Utc;
use http::StatusCode;
use serde::Deserialize;
@@ -40,6 +46,7 @@ use super::agent_identity::register_managed_chatgpt_agent_identity;
use super::agent_identity::require_agent_identity_authapi_base_url;
use super::agent_identity::verified_record_from_jwt;
use super::change_state::AuthChangeState;
use super::change_state::same_login;
use super::change_state::same_owner;
use super::external_bearer::BearerTokenRefresher;
use super::revoke::revoke_auth_tokens;
@@ -2079,7 +2086,7 @@ pub struct AuthManager {
managed_auth_policy: ManagedAuthPolicy,
chatgpt_base_url: Option<String>,
agent_identity_authapi_base_url: Option<String>,
refresh_lock: Semaphore,
refresh_lock: Arc<Semaphore>,
agent_identity_lock: Semaphore,
agent_identity_bootstrap_cooldown: Mutex<AgentIdentityBootstrapCooldown>,
external_auth: RwLock<Option<Arc<dyn ExternalAuth>>>,
@@ -2236,7 +2243,7 @@ impl AuthManager {
managed_auth_policy,
chatgpt_base_url,
agent_identity_authapi_base_url,
refresh_lock: Semaphore::new(/*permits*/ 1),
refresh_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
agent_identity_lock: Semaphore::new(/*permits*/ 1),
agent_identity_bootstrap_cooldown: Mutex::default(),
external_auth: RwLock::new(None),
@@ -2272,7 +2279,7 @@ impl AuthManager {
managed_auth_policy: ManagedAuthPolicy::default(),
chatgpt_base_url: None,
agent_identity_authapi_base_url: default_agent_identity_authapi_base_url(),
refresh_lock: Semaphore::new(/*permits*/ 1),
refresh_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
agent_identity_lock: Semaphore::new(/*permits*/ 1),
agent_identity_bootstrap_cooldown: Mutex::default(),
external_auth: RwLock::new(None),
@@ -2302,7 +2309,7 @@ impl AuthManager {
managed_auth_policy: ManagedAuthPolicy::default(),
chatgpt_base_url: None,
agent_identity_authapi_base_url: default_agent_identity_authapi_base_url(),
refresh_lock: Semaphore::new(/*permits*/ 1),
refresh_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
agent_identity_lock: Semaphore::new(/*permits*/ 1),
agent_identity_bootstrap_cooldown: Mutex::default(),
external_auth: RwLock::new(None),
@@ -2340,7 +2347,7 @@ impl AuthManager {
.trim_end_matches('/')
.to_string(),
),
refresh_lock: Semaphore::new(/*permits*/ 1),
refresh_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
agent_identity_lock: Semaphore::new(/*permits*/ 1),
agent_identity_bootstrap_cooldown: Mutex::default(),
external_auth: RwLock::new(None),
@@ -2368,7 +2375,7 @@ impl AuthManager {
managed_auth_policy: ManagedAuthPolicy::default(),
chatgpt_base_url: None,
agent_identity_authapi_base_url: default_agent_identity_authapi_base_url(),
refresh_lock: Semaphore::new(/*permits*/ 1),
refresh_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
agent_identity_lock: Semaphore::new(/*permits*/ 1),
agent_identity_bootstrap_cooldown: Mutex::default(),
external_auth: RwLock::new(Some(Arc::new(BearerTokenRefresher::new(config)))),
@@ -2509,6 +2516,93 @@ impl AuthManager {
.await
}
/// Borrows a persisted account grant while excluding this manager's token refreshes.
/// Release the lease after the IdP exchange, then check the grant before using its result.
pub async fn lease_account_oauth_grant(
self: &Arc<Self>,
) -> std::io::Result<Option<AccountOAuthGrantLease>> {
let permit = Arc::clone(&self.refresh_lock)
.acquire_owned()
.await
.map_err(|_| std::io::Error::other("Account refresh coordination is unavailable."))?;
let login_generation = self.auth_change_state_tx.borrow().login_generation;
let Some(tokens) = self.read_account_grant_tokens().await? else {
return Ok(None);
};
let grant = AccountOAuthGrant {
manager: Arc::clone(self),
tokens,
login_generation,
};
if !grant.owner_is_current() {
return Ok(None);
}
Ok(Some(AccountOAuthGrantLease { grant, permit }))
}
async fn read_account_grant_tokens(&self) -> std::io::Result<Option<TokenData>> {
if self.has_external_auth()
|| self.auth_credentials_store_mode == AuthCredentialsStoreMode::Ephemeral
{
return Ok(None);
}
let Some(auth @ CodexAuth::Chatgpt(_)) = self.auth_cached() else {
return Ok(None);
};
let allowed = self.allowed_login_methods();
if validate_auth_restrictions(
Some(&allowed),
self.effective_chatgpt_workspaces().as_deref(),
&auth,
)
.is_err()
{
return Ok(None);
}
let cached = auth.get_token_data()?;
let home = self.codex_home.clone();
let mode = self.auth_credentials_store_mode;
let backend = self.keyring_backend_kind;
let stored = tokio::task::spawn_blocking(move || {
// Unlike ordinary login loading, an unavailable keyring must not authorize
// account exchange using an older file fallback.
if mode == AuthCredentialsStoreMode::Auto {
let keyring =
create_auth_storage(home.clone(), AuthCredentialsStoreMode::Keyring, backend);
if let Some(auth) = keyring.load()? {
return Ok(Some(auth));
}
create_auth_storage(home, AuthCredentialsStoreMode::File, backend).load()
} else {
create_auth_storage(home, mode, backend).load()
}
})
.await
.map_err(|_| std::io::Error::other("Account credential read failed."))?
.map_err(|_: std::io::Error| {
std::io::Error::other("Account credential storage is unavailable.")
})?;
let Some(stored) = stored.filter(|auth| auth.resolved_mode() == AuthMode::Chatgpt) else {
return Ok(None);
};
Ok(stored.tokens.filter(|tokens| {
tokens.oauth_provenance.as_ref().is_some_and(|provenance| {
!provenance.issuer.trim().is_empty() && !provenance.client_id.trim().is_empty()
}) && tokens
.login_owner_id
.as_ref()
.is_some_and(|id| !id.is_empty())
&& !tokens.refresh_token.is_empty()
&& tokens
.id_token
.chatgpt_user_id
.as_ref()
.is_some_and(|id| !id.is_empty())
&& tokens.account_id.as_ref().is_some_and(|id| !id.is_empty())
&& same_account_grant(tokens, &cached)
}))
}
/// Reloads auth from the active source. Returns whether the auth value changed.
pub async fn reload(&self) -> bool {
tracing::info!("Reloading auth");
@@ -2668,6 +2762,8 @@ impl AuthManager {
auth_changed_for_refresh && !same_owner(previous, new_auth.as_ref());
let source_changed = auth_changed_for_refresh
&& !super::source::same_source(previous, new_auth.as_ref());
let login_changed =
auth_changed_for_refresh && !same_login(previous, new_auth.as_ref());
if owner_changed {
self.auth_route_config
.application_network_policy()
@@ -2687,6 +2783,9 @@ impl AuthManager {
if source_changed {
state.source_generation += 1;
}
if login_changed {
state.login_generation += 1;
}
});
self.auth_change_tx.send_modify(|revision| *revision += 1);
}
+2
View File
@@ -32,6 +32,8 @@ pub use success_page::CODEX_OPEN_APP_URL;
pub use success_page::LoginSuccessPage;
pub use success_page::LoginSuccessPageBrand;
pub use auth::AccountOAuthGrant;
pub use auth::AccountOAuthGrantLease;
pub use auth::AgentIdentityAuthPolicy;
pub use auth::AuthChangeState;
pub use auth::AuthConfig;
+151 -4
View File
@@ -4,7 +4,9 @@ use base64::Engine;
use chrono::Duration;
use chrono::Utc;
use codex_config::types::AuthCredentialsStoreMode;
use codex_http_client::DestinationPolicy;
use codex_http_client::HttpClientFactory;
use codex_http_client::NetworkPolicyController;
use codex_http_client::OutboundProxyPolicy;
use codex_http_client::cache_system_proxy_route_for_test;
use codex_login::AuthDotJson;
@@ -28,6 +30,10 @@ use std::ffi::OsString;
use std::net::TcpListener;
use std::process::Command;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::task::Wake;
use std::task::Waker;
use std::time::Duration as StdDuration;
use tempfile::TempDir;
use wiremock::Mock;
@@ -200,7 +206,19 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server).await?;
let mut ctx = RefreshTokenTestContext::new(&server).await?;
let policy_controller = NetworkPolicyController::default();
ctx.auth_manager = AuthManager::shared(
ctx.codex_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*forced_chatgpt_workspace_id*/ None,
/*chatgpt_base_url*/ None,
AuthKeyringBackendKind::default(),
codex_login::test_support::transport_default_auth_route_config()
.with_application_network_policy(policy_controller.policy()),
)
.await;
// Exercise the recorded issuer instead of the test endpoint override.
// SAFETY: auth_env serializes these tests; ctx restores the environment on drop.
unsafe {
@@ -209,6 +227,12 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
let initial_last_refresh = Utc::now() - Duration::days(1);
let mut initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
initial_tokens.login_owner_id = Some("completed-login".into());
initial_tokens.id_token = codex_login::token_data::parse_chatgpt_jwt_claims(
&jwt_with_payload(json!({
"sub": "user-123",
"https://api.openai.com/auth": { "chatgpt_user_id": "user-123", "chatgpt_account_id": "account-id" }
})),
)?;
initial_tokens.oauth_provenance = Some(AccountOAuthProvenance {
issuer: server.uri(),
client_id: "recorded-client".into(),
@@ -225,10 +249,56 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
};
ctx.write_auth(&initial_auth).await?;
ctx.auth_manager
.refresh_token_from_authority()
let changes = ctx.auth_manager.auth_change_state_receiver();
let before = *changes.borrow();
let lease = ctx
.auth_manager
.lease_account_oauth_grant()
.await?
.context("eligible account")?;
assert_eq!(lease.tokens(), &initial_tokens);
let refresh = ctx.auth_manager.refresh_token_from_authority();
tokio::pin!(refresh);
struct RefreshWake(AtomicBool);
impl Wake for RefreshWake {
fn wake(self: Arc<Self>) {
self.0.store(true, Ordering::SeqCst);
}
}
let refresh_wake = Arc::new(RefreshWake(AtomicBool::new(false)));
let waker = Waker::from(Arc::clone(&refresh_wake));
// Register the refresh waiter so lease release must wake it, not just return Pending.
assert!(
refresh
.as_mut()
.poll(&mut std::task::Context::from_waker(&waker))
.is_pending()
);
assert!(
server
.received_requests()
.await
.unwrap_or_default()
.is_empty()
);
// No await between clearing the flag and release: this wake must come from the lease.
refresh_wake.0.store(false, Ordering::SeqCst);
let grant = lease.release();
assert!(
refresh_wake.0.load(Ordering::SeqCst),
"releasing the account grant must wake the blocked refresh"
);
assert!(grant.check_current().await?);
refresh
.await
.context("refresh should succeed")?;
.context("refresh should succeed once the IdP lease is released")?;
assert!(
!grant.check_current().await?,
"the old credential was rotated"
);
let after = *changes.borrow();
assert_eq!(after.owner_generation, before.owner_generation);
assert_eq!(after.login_generation, before.login_generation);
let requests = server.received_requests().await.unwrap_or_default();
assert_eq!(
@@ -268,6 +338,83 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
.context("token data should be cached")?;
assert_eq!(cached, refreshed_tokens);
let lease = ctx
.auth_manager
.lease_account_oauth_grant()
.await?
.context("rotated account grant")?;
assert_eq!(lease.tokens(), &refreshed_tokens);
let grant = lease.release();
let mut replacement = stored.clone();
replacement
.tokens
.as_mut()
.context("replacement tokens")?
.login_owner_id = Some("new-login-same-account".into());
// A persisted replacement is visible even before the ordinary auth cache reloads.
save_auth(
ctx.codex_home.path(),
&replacement,
AuthCredentialsStoreMode::File,
AuthKeyringBackendKind::default(),
)?;
assert!(!grant.check_current().await?);
ctx.auth_manager.reload().await;
let after = *changes.borrow();
assert_eq!(after.owner_generation, before.owner_generation + 1);
assert_eq!(after.login_generation, before.login_generation + 1);
let lease = ctx
.auth_manager
.lease_account_oauth_grant()
.await?
.context("new login grant")?;
let grant = lease.release();
let before = *changes.borrow();
let policy = ctx.auth_manager.application_network_policy();
assert!(policy_controller.publish(policy.revision(), DestinationPolicy::Unrestricted));
let policy = policy.for_current_account();
let endpoint = server.uri().parse()?;
assert!(policy.acquire(&endpoint).is_ok());
// Older clients drop unknown login metadata when saving a same-account refresh.
let mut legacy = replacement.clone();
let tokens = legacy.tokens.as_mut().context("legacy tokens")?;
tokens.login_owner_id = None;
tokens.oauth_provenance = None;
ctx.write_auth(&legacy).await?;
let after = *changes.borrow();
assert_eq!(after.owner_generation, before.owner_generation);
assert!(policy.acquire(&endpoint).is_ok());
assert_eq!(after.login_generation, before.login_generation + 1);
assert!(!grant.check_current().await?);
assert!(
ctx.auth_manager
.lease_account_oauth_grant()
.await?
.is_none()
);
// Restoring identical credentials must not revive the old EMA grant.
ctx.write_auth(&replacement).await?;
assert_eq!(
changes.borrow().login_generation,
before.login_generation + 2
);
assert!(!grant.check_current().await?);
let lease = ctx
.auth_manager
.lease_account_oauth_grant()
.await?
.context("restored login grant")?;
// Logout must complete even while an IdP exchange holds the refresh lease.
ctx.auth_manager.logout().await?;
assert!(!lease.release().check_current().await?);
assert!(
ctx.auth_manager
.lease_account_oauth_grant()
.await?
.is_none()
);
server.verify().await;
Ok(())
}