openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bind MCP OAuth refresh tokens to their issuer (#39615)

Browse Source 
## Why

An authorization server discovered for an MCP server can change. A stored refresh token must not be sent to a different issuer than the one that originally granted it.

## What changed

- Persist the discovered authorization server issuer with new OAuth credentials and preserve it across refreshes.
- Validate that issuer against a single pinned metadata snapshot before refreshing or adopting stored refresh credentials.
- Require reauthentication when refresh credentials have no issuer or the issuer changed. An unexpired access token can still be used without exposing its refresh token.

## Testing

Add coverage for issuer persistence, missing and changed issuers, metadata changes during startup, and refresh-token preservation.

GitOrigin-RevId: 07e48ba892f627428a28bbed5cca4ad703e2a5fd
This commit is contained in:
Jeremy Rose
2026-08-20 06:25:17 +00:00
committed by copyberry
parent 530c1aed58
commit 250b5ea2bf
16 changed files with 722 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/mcp_server_status.rs
+8 -37
View File
@@ -251,7 +251,6 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
});
let registrations = Arc::new(AtomicUsize::new(0));
let registration_count = Arc::clone(&registrations);
let token_count = Arc::new(AtomicUsize::new(0));
let (token_request_tx, mut token_request_rx) = mpsc::unbounded_channel();
let (mcp_authorization_tx, mut mcp_authorization_rx) = mpsc::unbounded_channel();
let tool_name = Arc::new("cimd".to_string());
@@ -304,7 +303,6 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
"/token",
post(move |headers: HeaderMap, body: Bytes| {
let token_request_tx = token_request_tx.clone();
let token_count = Arc::clone(&token_count);
async move {
let _ = token_request_tx.send((
String::from_utf8_lossy(&body).into_owned(),
@@ -313,21 +311,12 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
.and_then(|value| value.to_str().ok())
.map(str::to_string),
));
if token_count.fetch_add(1, Ordering::SeqCst) == 0 {
Json(json!({
"access_token": "expired-cimd-access-token",
"token_type": "Bearer",
"expires_in": 0,
"refresh_token": "test-refresh-token",
}))
} else {
Json(json!({
"access_token": "refreshed-cimd-access-token",
"token_type": "Bearer",
"expires_in": 3600,
"refresh_token": "test-refresh-token",
}))
}
Json(json!({
"access_token": "cimd-access-token",
"token_type": "Bearer",
"expires_in": 3600,
"refresh_token": "test-refresh-token",
}))
}
}),
)
@@ -431,29 +420,11 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
},
})
.await?;
let (refresh_request, refresh_authorization) =
timeout(DEFAULT_READ_TIMEOUT, token_request_rx.recv())
.await?
.expect("expired CIMD token should be refreshed");
let refresh_parameters = url::form_urlencoded::parse(refresh_request.as_bytes())
.into_owned()
.collect::<BTreeMap<String, String>>();
assert_eq!(
refresh_parameters.get("grant_type").map(String::as_str),
Some("refresh_token")
);
assert_eq!(
refresh_parameters.get("refresh_token").map(String::as_str),
Some("test-refresh-token")
);
assert_eq!(refresh_parameters.get("client_id"), Some(&client_id));
assert!(!refresh_parameters.contains_key("client_secret"));
assert_eq!(refresh_authorization, None);
assert_eq!(
timeout(DEFAULT_READ_TIMEOUT, mcp_authorization_rx.recv())
.await?
.expect("MCP startup should use the refreshed token"),
"Bearer refreshed-cimd-access-token"
.expect("MCP startup should use the access token"),
"Bearer cimd-access-token"
);
assert_eq!(registrations.load(Ordering::SeqCst), 0);
codex-rs/core/tests/suite/rmcp_client.rs
+1
View File
@@ -4108,6 +4108,7 @@ fn write_fallback_oauth_tokens(
"server_name": server_name,
"url": server_url,
"client_id": client_id,
"issuer": server_url,
"token_response": {
"access_token": access_token,
"token_type": "Bearer",
codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs
+1
View File
@@ -179,6 +179,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.header(CONTENT_TYPE, "application/json")
.body(Body::from(
serde_json::to_vec(&json!({
"issuer": format!("{metadata_base}/mcp"),
"authorization_endpoint": format!("{metadata_base}/oauth/authorize"),
"token_endpoint": format!("{metadata_base}/oauth/token"),
"scopes_supported": [""],
codex-rs/rmcp-client/src/oauth.rs
+43 -6
View File
@@ -16,6 +16,7 @@
//!
//! If the keyring is not available or fails, we fall back to CODEX_HOME/.credentials.json which is consistent with other coding CLI agents.
mod issuer_binding;
mod refresh_lock;
mod refresh_transaction;
mod resolved_store;
@@ -23,7 +24,7 @@ mod store_lock;
#[cfg(test)]
#[path = "oauth/test_support.rs"]
mod test_support;
pub(crate) mod test_support;
use anyhow::Context;
use anyhow::Error;
@@ -70,6 +71,8 @@ use tokio::sync::Mutex;
use codex_utils_home_dir::find_codex_home;
pub(crate) use self::issuer_binding::validate_refresh_token_issuer;
pub(crate) use self::refresh_transaction::install_tokens_in_manager;
pub(crate) use self::resolved_store::ResolvedOAuthCredentialStore;
pub(crate) use self::resolved_store::ResolvedOAuthTokens;
pub(crate) use self::resolved_store::resolve_oauth_tokens_from_store_policy;
@@ -83,12 +86,40 @@ const REFRESH_SKEW_MILLIS: u64 = 30_000;
pub struct StoredOAuthTokens {
pub server_name: String,
pub url: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub issuer: Option<String>,
pub client_id: String,
pub token_response: WrappedOAuthTokenResponse,
#[serde(default)]
pub expires_at: Option<u64>,
}
impl StoredOAuthTokens {
pub(crate) fn has_refresh_token(&self) -> bool {
self.token_response
.0
.refresh_token()
.is_some_and(|refresh_token| !refresh_token.secret().trim().is_empty())
}
pub(crate) fn bound_issuer(&self) -> Option<&str> {
self.issuer
.as_deref()
.filter(|issuer| !issuer.trim().is_empty())
}
pub(crate) fn access_token_is_usable_without_refresh(&self) -> bool {
!token_needs_refresh(self.expires_at)
&& !self
.token_response
.0
.access_token()
.secret()
.trim()
.is_empty()
}
}
/// OAuth credentials paired with the concrete store selected for their client lifecycle.
#[derive(Debug, Clone)]
pub struct StoredOAuthCredentialSnapshot {
@@ -281,14 +312,11 @@ fn oauth_tokens_are_usable(tokens: &StoredOAuthTokens) -> bool {
return false;
}
let token_response = &tokens.token_response.0;
if token_needs_refresh(tokens.expires_at) {
return token_response
.refresh_token()
.is_some_and(|token| !token.secret().trim().is_empty());
return tokens.bound_issuer().is_some() && tokens.has_refresh_token();
}
!token_response.access_token().secret().trim().is_empty()
tokens.access_token_is_usable_without_refresh()
}
fn refresh_expires_in_from_timestamp(tokens: &mut StoredOAuthTokens) {
@@ -702,6 +730,9 @@ impl OAuthPersistor {
let stored = StoredOAuthTokens {
server_name: self.inner.server_name.clone(),
url: self.inner.url.clone(),
issuer: last_credentials
.as_ref()
.and_then(|previous| previous.issuer.clone()),
client_id,
token_response: new_token_response,
expires_at,
@@ -746,6 +777,8 @@ type FallbackFile = BTreeMap<String, FallbackTokenEntry>;
struct FallbackTokenEntry {
server_name: String,
server_url: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
issuer: Option<String>,
client_id: String,
access_token: String,
#[serde(default)]
@@ -810,6 +843,7 @@ fn load_oauth_tokens_from_file_with_lock_held(
let mut stored = StoredOAuthTokens {
server_name: entry.server_name.clone(),
url: entry.server_url.clone(),
issuer: entry.issuer.clone(),
client_id: entry.client_id.clone(),
token_response: WrappedOAuthTokenResponse(token_response),
expires_at: entry.expires_at,
@@ -852,6 +886,7 @@ fn save_oauth_tokens_to_file_with_lock_held(tokens: &StoredOAuthTokens) -> Resul
let entry = FallbackTokenEntry {
server_name: tokens.server_name.clone(),
server_url: tokens.url.clone(),
issuer: tokens.issuer.clone(),
client_id: tokens.client_id.clone(),
access_token: token_response.access_token().secret().to_string(),
expires_at,
@@ -1673,6 +1708,7 @@ mod tests {
) {
assert_eq!(actual.server_name, expected.server_name);
assert_eq!(actual.url, expected.url);
assert_eq!(actual.issuer, expected.issuer);
assert_eq!(actual.client_id, expected.client_id);
assert_eq!(actual.expires_at, expected.expires_at);
assert_token_response_match_without_expiry(
@@ -1726,6 +1762,7 @@ mod tests {
StoredOAuthTokens {
server_name: "test-server".to_string(),
url: "https://example.test".to_string(),
issuer: Some("https://issuer.example.test".to_string()),
client_id: "client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at,
codex-rs/rmcp-client/src/oauth/issuer_binding.rs
+52
View File
@@ -0,0 +1,52 @@
use anyhow::Context;
use anyhow::Result;
use rmcp::transport::auth::AuthError;
use rmcp::transport::auth::AuthorizationMetadata;
use super::StoredOAuthTokens;
/// Verifies that a stored refresh token remains bound to its original issuer.
///
/// Call this with the same metadata snapshot that RMCP will use for the credentials. Missing or
/// changed issuers require a new login rather than risking sending a refresh token to a different
/// authorization server.
pub(crate) fn validate_refresh_token_issuer(
metadata: &AuthorizationMetadata,
tokens: &StoredOAuthTokens,
) -> Result<()> {
if !tokens.has_refresh_token() {
return Ok(());
}
let Some(stored_issuer) = tokens.bound_issuer() else {
return Err(AuthError::AuthorizationRequired).with_context(|| {
format!(
"OAuth refresh credentials for server {} are missing an authorization server issuer; authorization required",
tokens.server_name
)
});
};
let Some(current_issuer) = metadata
.issuer
.as_deref()
.filter(|issuer| !issuer.trim().is_empty())
else {
return Err(AuthError::AuthorizationRequired).with_context(|| {
format!(
"OAuth metadata for server {} did not include an authorization server issuer; authorization required",
tokens.server_name
)
});
};
if current_issuer != stored_issuer {
return Err(AuthError::AuthorizationRequired).with_context(|| {
format!(
"OAuth authorization server issuer changed for server {}; authorization required",
tokens.server_name
)
});
}
Ok(())
}
codex-rs/rmcp-client/src/oauth/refresh_transaction.rs
+38 -17
View File
@@ -26,6 +26,7 @@ use super::WrappedOAuthTokenResponse;
use super::compute_expires_at_millis;
use super::refresh_lock::RefreshCredentialLock;
use super::token_needs_refresh;
use super::validate_refresh_token_issuer;
const REFRESH_REQUEST_TIMEOUT: Duration = Duration::from_secs(45);
@@ -136,19 +137,27 @@ impl OAuthPersistor {
debug!("adopting newer MCP OAuth credentials without contacting the provider");
let manager = self.inner.authorization_manager.clone();
let mut guard = manager.lock().await;
install_tokens_in_manager_guard(&mut guard, &latest).await?;
if latest.has_refresh_token() {
let previous = self.inner.last_credentials.lock().await;
let expected_issuer = previous.as_ref().and_then(StoredOAuthTokens::bound_issuer);
let latest_issuer = latest.bound_issuer();
if latest_issuer.is_none() || latest_issuer != expected_issuer {
return Err(AuthError::AuthorizationRequired).with_context(|| {
format!(
"OAuth refresh credentials for server {} could not be bound to the previously validated issuer; authorization required",
self.inner.server_name
)
});
}
}
install_tokens_in_manager(&mut guard, &latest).await?;
*self.inner.last_credentials.lock().await = Some(latest);
return Ok(());
}
// Preserve RMCP's `AuthorizationRequired` marker only for credentials known to be
// unrefreshable. Network and provider failures below remain ordinary errors.
if latest
.token_response
.0
.refresh_token()
.is_none_or(|refresh_token| refresh_token.secret().trim().is_empty())
{
if !latest.has_refresh_token() {
return Err(AuthError::AuthorizationRequired).with_context(|| {
format!(
"OAuth tokens for server {} cannot be refreshed; authorization required",
@@ -161,7 +170,14 @@ impl OAuthPersistor {
// The provider uses a separate HTTP client and cannot re-enter `AuthClient`. Retain this
// async guard so requests cannot observe credentials while they are staged and committed.
let mut guard = manager.lock().await;
install_tokens_in_manager_guard(&mut guard, &latest)
let metadata = guard
.resolve_metadata()
.await
.context("failed to resolve OAuth metadata before using stored refresh credentials")?
.metadata;
validate_refresh_token_issuer(&metadata, &latest)?;
guard.set_metadata(metadata);
install_tokens_in_manager(&mut guard, &latest)
.await
.context("failed to stage OAuth credentials for refresh")?;
// The owned task prevents caller deadlines from canceling after possible token rotation;
@@ -230,7 +246,7 @@ impl OAuthPersistor {
error = %error,
"failed to persist refreshed MCP OAuth credentials; returning the error and restoring the previous in-process credentials"
);
install_tokens_in_manager_guard(&mut guard, &latest)
install_tokens_in_manager(&mut guard, &latest)
.await
.context(
"failed to restore previous OAuth credentials after refresh persistence failed",
@@ -241,7 +257,7 @@ impl OAuthPersistor {
// This layer retains RMCP's legacy persistence hook. Install the same merged response
// (including carried-forward refresh token/scopes) so that hook cannot overwrite durable
// credentials with the provider's partial response.
install_tokens_in_manager_guard(&mut guard, &refreshed)
install_tokens_in_manager(&mut guard, &refreshed)
.await
.context(
"refreshed OAuth tokens were persisted but could not be installed in the authorization manager",
@@ -253,7 +269,8 @@ impl OAuthPersistor {
}
}
async fn install_tokens_in_manager_guard(
/// Installs tokens without resolving metadata again, so callers can pin the validated snapshot.
pub(crate) async fn install_tokens_in_manager(
authorization_manager: &mut AuthorizationManager,
tokens: &StoredOAuthTokens,
) -> Result<()> {
@@ -268,12 +285,15 @@ async fn install_tokens_in_manager_guard(
.ok()
.map(|duration| duration.as_secs());
store
.save(StoredCredentials::new(
tokens.client_id.clone(),
Some(token_response),
granted_scopes,
token_received_at,
))
.save(
StoredCredentials::new(
tokens.client_id.clone(),
Some(token_response),
granted_scopes,
token_received_at,
)
.with_issuer(tokens.issuer.clone()),
)
.await
.context("failed to stage OAuth tokens for authorization manager")?;
@@ -301,6 +321,7 @@ fn refreshed_tokens(
StoredOAuthTokens {
server_name: inner.server_name.clone(),
url: inner.url.clone(),
issuer: previous.issuer.clone(),
client_id: previous.client_id.clone(),
expires_at: compute_expires_at_millis(&token_response),
token_response: WrappedOAuthTokenResponse(token_response),
codex-rs/rmcp-client/src/oauth/test_support.rs
+2 -2
View File
@@ -10,13 +10,13 @@ use tempfile::tempdir;
///
/// Keep OAuth tests on this one guard instead of defining per-module helpers; otherwise
/// concurrently running test modules can point File/Secrets storage at different homes.
pub(super) struct TempCodexHome {
pub(crate) struct TempCodexHome {
_guard: MutexGuard<'static, ()>,
_dir: tempfile::TempDir,
}
impl TempCodexHome {
pub(super) fn new() -> Self {
pub(crate) fn new() -> Self {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
let guard = LOCK
.get_or_init(Mutex::default)
codex-rs/rmcp-client/src/oauth/tests/persistor_tests.rs
+72
View File
@@ -143,6 +143,7 @@ async fn concurrent_refreshes_call_provider_once_and_carry_omitted_fields() -> R
first.persist_if_needed().await?;
let stored = load_oauth_tokens_from_file(&initial.server_name, &initial.url)?
.expect("refreshed credentials should be stored");
assert_eq!(stored.issuer, initial.issuer);
let live_credentials = first.stored_credentials().await;
let disk_credentials = stored_oauth_credentials(
&initial.server_name,
@@ -248,6 +249,76 @@ async fn rejected_refresh_token_requires_reauthorization() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn changed_issuer_requires_reauthorization_before_refresh() -> Result<()> {
let (_env, server, mut initial) = test_context().await?;
initial.issuer = Some("https://original-issuer.example.test".to_string());
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(ResponseTemplate::new(500))
.expect(0)
.mount(&server)
.await;
save_oauth_tokens_to_file(&initial)?;
let persistor = persistor_for(&initial).await?;
let error = persistor
.refresh_if_needed()
.await
.expect_err("a changed issuer must abort refresh");
assert!(is_authentication_required_error(&error));
server.verify().await;
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn missing_issuer_requires_reauthorization_before_refresh() -> Result<()> {
let (_env, server, mut initial) = test_context().await?;
initial.issuer = None;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(ResponseTemplate::new(500))
.expect(0)
.mount(&server)
.await;
save_oauth_tokens_to_file(&initial)?;
let persistor = persistor_for(&initial).await?;
let error = persistor
.refresh_if_needed()
.await
.expect_err("a missing issuer must abort refresh");
assert!(is_authentication_required_error(&error));
server.verify().await;
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn issuerless_newer_credentials_are_not_adopted_before_refresh() -> Result<()> {
let (_env, server, initial) = test_context().await?;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(ResponseTemplate::new(500))
.expect(0)
.mount(&server)
.await;
save_oauth_tokens_to_file(&initial)?;
let persistor = persistor_for(&initial).await?;
let mut latest = initial.clone();
latest.issuer = None;
latest.expires_at = Some(u64::MAX);
save_oauth_tokens_to_file(&latest)?;
let error = persistor
.refresh_if_needed()
.await
.expect_err("an issuer-less refresh token must not be adopted");
assert!(is_authentication_required_error(&error));
server.verify().await;
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn transient_refresh_failure_does_not_require_reauthorization() -> Result<()> {
let (_env, server, initial) = test_context().await?;
@@ -470,6 +541,7 @@ async fn wait_for_lock_contention(rx: mpsc::Receiver<()>, expected_count: usize)
fn expired_tokens(url: &str) -> StoredOAuthTokens {
let mut tokens = sample_tokens();
tokens.url = url.to_string();
tokens.issuer = Some(url.to_string());
tokens.expires_at = Some(0);
tokens
.token_response
codex-rs/rmcp-client/src/oauth/tests/store_lock_tests.rs
+1
View File
@@ -103,6 +103,7 @@ fn sample_tokens() -> StoredOAuthTokens {
StoredOAuthTokens {
server_name: "test-server".to_string(),
url: "https://example.test".to_string(),
issuer: Some("https://issuer.example.test".to_string()),
client_id: "client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at,
codex-rs/rmcp-client/src/oauth_client_registration.rs
+12 -2
View File
@@ -21,6 +21,12 @@ pub enum McpOAuthClientRegistration {
Dcr,
}
/// OAuth state prepared from one authorization-server metadata resolution.
pub(crate) struct PreparedOAuthLogin {
pub(crate) oauth_state: OAuthState,
pub(crate) authorization_server_issuer: Option<String>,
}
pub(crate) async fn start_authorization(
server_url: &str,
http_client: Arc<dyn OAuthHttpClient>,
@@ -28,11 +34,12 @@ pub(crate) async fn start_authorization(
redirect_uri: &str,
callback_id: &str,
client_registration: McpOAuthClientRegistration,
) -> Result<OAuthState> {
) -> Result<PreparedOAuthLogin> {
let mut auth_manager =
AuthorizationManager::new_with_oauth_http_client(server_url, http_client).await?;
auth_manager.set_allow_missing_issuer(true);
let metadata = auth_manager.resolve_metadata().await?.metadata;
let authorization_server_issuer = metadata.issuer.clone();
let cimd_advertised = metadata
.additional_fields
@@ -96,7 +103,10 @@ pub(crate) async fn start_authorization(
.await
.map_err(|(_auth_manager, error)| error)?;
Ok(OAuthState::Session(session))
Ok(PreparedOAuthLogin {
oauth_state: OAuthState::Session(session),
authorization_server_issuer,
})
}
#[cfg(test)]
codex-rs/rmcp-client/src/oauth_client_registration_tests.rs
+8 -2
View File
@@ -101,7 +101,7 @@ async fn authorization(
redirect_uri: &str,
registration: McpOAuthClientRegistration,
) -> Result<(OAuthState, HashMap<String, String>)> {
let state = start_authorization(
let prepared = start_authorization(
&format!("{}/mcp", server.uri()),
Arc::new(OAuthHttpClientAdapter::new(
Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
@@ -116,6 +116,7 @@ async fn authorization(
registration,
)
.await?;
let state = prepared.oauth_state;
let query = Url::parse(&state.get_authorization_url().await?)?
.query_pairs()
.into_owned()
@@ -306,7 +307,7 @@ async fn resource_headers_follow_same_origin_registration_redirect_and_sdk_auth_
])),
/*env_http_headers*/ None,
)?;
let mut state = start_authorization(
let prepared = start_authorization(
&resource_url,
Arc::new(OAuthHttpClientAdapter::new(
Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
@@ -321,6 +322,11 @@ async fn resource_headers_follow_same_origin_registration_redirect_and_sdk_auth_
McpOAuthClientRegistration::Dcr,
)
.await?;
assert_eq!(
prepared.authorization_server_issuer.as_deref(),
Some(authorization_server.uri().as_str())
);
let mut state = prepared.oauth_state;
let csrf_state = Url::parse(&state.get_authorization_url().await?)?
.query_pairs()
.find(|(name, _)| name == "state")
codex-rs/rmcp-client/src/perform_oauth_login.rs
+123 -7
View File
@@ -29,6 +29,7 @@ use crate::WrappedOAuthTokenResponse;
use crate::http_client_adapter::StreamableHttpRedirectMode;
use crate::oauth::compute_expires_at_millis;
use crate::oauth_client_registration::McpOAuthClientRegistration;
use crate::oauth_client_registration::PreparedOAuthLogin;
use crate::oauth_client_registration::start_authorization as start_client_registration;
use crate::oauth_http_client::OAuthHttpClientAdapter;
use crate::save_oauth_tokens;
@@ -396,6 +397,7 @@ impl OauthLoginHandle {
struct OauthLoginFlow {
auth_url: String,
oauth_state: OAuthState,
authorization_server_issuer: Option<String>,
rx: oneshot::Receiver<CallbackResult>,
guard: CallbackServerGuard,
server_name: String,
@@ -554,7 +556,10 @@ impl OauthLoginFlow {
)?);
let scope_refs: Vec<&str> = scopes.iter().map(String::as_str).collect();
let oauth_state = if let Some(oauth_client_id) =
let PreparedOAuthLogin {
oauth_state,
authorization_server_issuer,
} = if let Some(oauth_client_id) =
oauth_client_id.filter(|client_id| !client_id.trim().is_empty())
{
start_authorization(
@@ -587,6 +592,7 @@ impl OauthLoginFlow {
Ok(Self {
auth_url,
oauth_state,
authorization_server_issuer,
rx,
guard,
server_name: server_name.to_string(),
@@ -653,6 +659,7 @@ impl OauthLoginFlow {
let stored = StoredOAuthTokens {
server_name: self.server_name.clone(),
url: self.server_url.clone(),
issuer: self.authorization_server_issuer.clone(),
client_id,
token_response: WrappedOAuthTokenResponse(credentials),
expires_at,
@@ -698,11 +705,12 @@ async fn start_authorization(
scopes: &[&str],
redirect_uri: &str,
oauth_client_id: &str,
) -> Result<OAuthState> {
) -> Result<PreparedOAuthLogin> {
let mut auth_manager =
AuthorizationManager::new_with_oauth_http_client(server_url, http_client).await?;
auth_manager.set_allow_missing_issuer(true);
let metadata = auth_manager.resolve_metadata().await?.metadata;
let authorization_server_issuer = metadata.issuer.clone();
auth_manager.set_metadata(metadata);
auth_manager.configure_client(
OAuthClientConfig::new(oauth_client_id, redirect_uri)
@@ -710,9 +718,14 @@ async fn start_authorization(
)?;
let auth_url = auth_manager.get_authorization_url(scopes).await?;
Ok(OAuthState::Session(
AuthorizationSession::for_scope_upgrade(auth_manager, auth_url, redirect_uri),
))
Ok(PreparedOAuthLogin {
oauth_state: OAuthState::Session(AuthorizationSession::for_scope_upgrade(
auth_manager,
auth_url,
redirect_uri,
)),
authorization_server_issuer,
})
}
fn append_query_param(url: &str, key: &str, value: Option<&str>) -> String {
@@ -735,6 +748,9 @@ fn append_query_param(url: &str, key: &str, value: Option<&str>) -> String {
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::io::Read;
use std::io::Write;
use std::net::TcpStream;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
@@ -763,7 +779,9 @@ mod tests {
use super::CallbackOutcome;
use super::McpOAuthClientRegistration;
use super::OAuthHttpClientAdapter;
use super::OAuthHttpContext;
use super::OAuthProviderError;
use super::OauthLoginFlow;
use super::StreamableHttpRedirectMode;
use super::append_callback_id_to_redirect_uri;
use super::append_query_param;
@@ -773,6 +791,8 @@ mod tests {
use super::perform_oauth_login;
use super::perform_oauth_login_silent;
use super::start_authorization;
use crate::oauth::stored_oauth_credentials;
use crate::oauth::test_support::TempCodexHome;
#[derive(Default)]
struct RecordingHttpClient {
@@ -845,6 +865,16 @@ mod tests {
Json(json!({"client_id": "unexpected-dynamic-client"}))
}
}),
)
.route(
"/oauth/token",
post(|| async {
Json(json!({
"access_token": "test-access-token",
"token_type": "Bearer",
"refresh_token": "test-refresh-token",
}))
}),
);
tokio::spawn(async move {
@@ -856,12 +886,96 @@ mod tests {
(base_url, registration_requests)
}
async fn send_oauth_callback(callback_url: Url) -> anyhow::Result<()> {
tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
let host = callback_url
.host_str()
.ok_or_else(|| anyhow::anyhow!("callback URL should include a host"))?;
let port = callback_url
.port()
.ok_or_else(|| anyhow::anyhow!("callback URL should include a port"))?;
let mut stream = TcpStream::connect((host, port))?;
let mut path = callback_url.path().to_string();
if let Some(query) = callback_url.query() {
path.push('?');
path.push_str(query);
}
write!(
stream,
"GET {path} HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n"
)?;
let mut response = String::new();
stream.read_to_string(&mut response)?;
anyhow::ensure!(
response.starts_with("HTTP/1.1 200"),
"OAuth callback failed: {response}"
);
Ok(())
})
.await?
}
#[tokio::test]
async fn oauth_login_persists_discovered_issuer() -> anyhow::Result<()> {
let _env = TempCodexHome::new();
let (base_url, _registration_requests) = spawn_oauth_metadata_server().await;
let server_url = format!("{base_url}/mcp");
let flow = OauthLoginFlow::new(
"issuer-persistence-test",
&server_url,
OAuthCredentialsStoreMode::File,
AuthKeyringBackendKind::Direct,
OAuthHttpContext {
http_headers: None,
env_http_headers: None,
http_client: Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
OutboundProxyPolicy::ReqwestDefault,
))),
redirect_mode: StreamableHttpRedirectMode::Legacy,
},
&[],
Some("test-client"),
McpOAuthClientRegistration::Auto,
/*oauth_resource*/ None,
/*launch_browser*/ false,
/*callback_port*/ None,
/*callback_url*/ None,
Some(/*timeout_secs*/ 5),
)
.await?;
let authorization_url = Url::parse(&flow.authorization_url())?;
let query = authorization_url.query_pairs().collect::<HashMap<_, _>>();
let redirect_uri = query
.get("redirect_uri")
.ok_or_else(|| anyhow::anyhow!("authorization URL should include redirect_uri"))?;
let state = query
.get("state")
.ok_or_else(|| anyhow::anyhow!("authorization URL should include state"))?;
let mut callback_url = Url::parse(redirect_uri)?;
callback_url
.query_pairs_mut()
.append_pair("code", "test-code")
.append_pair("state", state);
send_oauth_callback(callback_url).await?;
flow.finish(/*emit_browser_url*/ false).await?;
let stored = stored_oauth_credentials(
"issuer-persistence-test",
&server_url,
OAuthCredentialsStoreMode::File,
AuthKeyringBackendKind::Direct,
)?
.expect("OAuth login should persist credentials");
assert_eq!(stored.issuer.as_deref(), Some(server_url.as_str()));
Ok(())
}
#[tokio::test]
async fn configured_client_preserves_exact_scopes_and_redirect_without_registration() {
for (scopes, expected_scope) in [(&[][..], None), (&["read"][..], Some("read"))] {
let (base_url, registration_requests) = spawn_oauth_metadata_server().await;
let redirect_uri = "http://127.0.0.1:43123/callback/configured-client";
let oauth_state = start_authorization(
let prepared = start_authorization(
&format!("{base_url}/mcp"),
Arc::new(OAuthHttpClientAdapter::new(
Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
@@ -876,6 +990,7 @@ mod tests {
)
.await
.expect("start pre-registered OAuth authorization");
let oauth_state = prepared.oauth_state;
let authorization_url = oauth_state
.get_authorization_url()
@@ -955,7 +1070,7 @@ mod tests {
.await
.expect("serve authorization metadata fixture");
});
let mut state = start_authorization(
let prepared = start_authorization(
&format!("{issuer}/mcp"),
Arc::new(OAuthHttpClientAdapter::new(
Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
@@ -970,6 +1085,7 @@ mod tests {
)
.await
.expect("start issuer-aware authorization");
let mut state = prepared.oauth_state;
let csrf_state = Url::parse(
&state
.get_authorization_url()
codex-rs/rmcp-client/src/rmcp_client.rs
+45 -29
View File
@@ -12,6 +12,7 @@ use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use codex_api::SharedAuthProvider;
@@ -57,7 +58,6 @@ use rmcp::transport::AuthorizationManager;
use rmcp::transport::StreamableHttpClientTransport;
use rmcp::transport::auth::AuthClient;
use rmcp::transport::auth::AuthError;
use rmcp::transport::auth::OAuthState;
use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
use rmcp::transport::streamable_http_client::StreamableHttpError;
use serde::Deserialize;
@@ -81,7 +81,9 @@ use crate::oauth::OAuthPersistor;
use crate::oauth::ResolvedOAuthCredentialStore;
use crate::oauth::ResolvedOAuthTokens;
use crate::oauth::StoredOAuthTokens;
use crate::oauth::install_tokens_in_manager;
use crate::oauth::resolve_oauth_tokens_from_store_policy;
use crate::oauth::validate_refresh_token_issuer;
use crate::oauth_http_client::OAuthHttpClientAdapter;
use crate::protocol_mode::McpProtocolMode;
use crate::stdio_server_launcher::StdioServerCommand;
@@ -112,6 +114,9 @@ enum PendingTransport {
transport: StreamableHttpClientTransport<AuthClient<StreamableHttpClientAdapter>>,
oauth_persistor: OAuthPersistor,
},
StreamableHttpWithAccessTokenOnly {
transport: StreamableHttpClientTransport<AuthClient<StreamableHttpClientAdapter>>,
},
}
enum ClientState {
@@ -448,7 +453,8 @@ impl RmcpClient {
PendingTransport::Stdio { transport } => Some(transport.process_handle()),
PendingTransport::InProcess { .. }
| PendingTransport::StreamableHttp { .. }
| PendingTransport::StreamableHttpWithOAuth { .. } => None,
| PendingTransport::StreamableHttpWithOAuth { .. }
| PendingTransport::StreamableHttpWithAccessTokenOnly { .. } => None,
};
Ok(Self {
@@ -1091,12 +1097,7 @@ impl RmcpClient {
)
.await
{
Ok((transport, oauth_persistor)) => {
Ok(PendingTransport::StreamableHttpWithOAuth {
transport,
oauth_persistor,
})
}
Ok(pending_transport) => Ok(pending_transport),
Err(err)
if err.downcast_ref::<AuthError>().is_some_and(|auth_err| {
matches!(auth_err, AuthError::NoAuthorizationSupport)
@@ -1206,6 +1207,12 @@ impl RmcpClient {
.boxed(),
Some(oauth_persistor),
),
PendingTransport::StreamableHttpWithAccessTokenOnly { transport } => (
client_service
.serve_with_lifecycle(transport, lifecycle)
.boxed(),
None,
),
};
let service_result = match timeout {
@@ -1477,10 +1484,7 @@ async fn create_oauth_transport_and_runtime(
has_configured_headers: bool,
redirect_mode: StreamableHttpRedirectMode,
initialize_deadline: Arc<StdMutex<Option<Instant>>>,
) -> Result<(
StreamableHttpClientTransport<AuthClient<StreamableHttpClientAdapter>>,
OAuthPersistor,
)> {
) -> Result<PendingTransport> {
let oauth_http_client = Arc::new(OAuthHttpClientAdapter::new_with_redirect_mode(
http_client.clone(),
default_headers.clone(),
@@ -1492,22 +1496,24 @@ async fn create_oauth_transport_and_runtime(
AuthorizationManager::new_with_oauth_http_client(url.to_string(), oauth_http_client)
.await?;
manager.set_allow_missing_issuer(true);
let mut oauth_state = OAuthState::Unauthorized(manager);
oauth_state
.set_credentials(
&initial_tokens.client_id,
initial_tokens.token_response.0.clone(),
)
.await?;
let manager = match oauth_state {
OAuthState::Authorized(manager) => manager,
OAuthState::Unauthorized(manager) => manager,
_ => {
return Err(anyhow!("unexpected OAuth state during client setup"));
}
};
let metadata = manager
.resolve_metadata()
.await
.context("failed to resolve OAuth metadata before using stored credentials")?
.metadata;
let use_stored_access_token_only =
match validate_refresh_token_issuer(&metadata, &initial_tokens) {
Ok(()) => false,
Err(_error) if initial_tokens.access_token_is_usable_without_refresh() => true,
Err(error) => return Err(error),
};
manager.set_metadata(metadata);
let mut runtime_tokens = initial_tokens.clone();
if use_stored_access_token_only {
runtime_tokens.token_response.0.set_refresh_token(None);
runtime_tokens.issuer = None;
}
install_tokens_in_manager(&mut manager, &runtime_tokens).await?;
let auth_client = AuthClient::new(
StreamableHttpClientAdapter::new(
@@ -1527,6 +1533,13 @@ async fn create_oauth_transport_and_runtime(
StreamableHttpClientTransportConfig::with_uri(url.to_string()),
);
if use_stored_access_token_only {
warn!(
"stored OAuth refresh credentials could not be bound to their issuer for MCP server `{server_name}`; using the stored access token without refresh"
);
return Ok(PendingTransport::StreamableHttpWithAccessTokenOnly { transport });
}
let runtime = OAuthPersistor::new(
server_name.to_string(),
url.to_string(),
@@ -1535,7 +1548,10 @@ async fn create_oauth_transport_and_runtime(
Some(initial_tokens),
);
Ok((transport, runtime))
Ok(PendingTransport::StreamableHttpWithOAuth {
transport,
oauth_persistor: runtime,
})
}
#[cfg(test)]
codex-rs/rmcp-client/src/streamable_http_retry.rs
+2 -1
View File
@@ -35,7 +35,8 @@ impl RmcpClient {
let should_retry = match &initial_transport {
PendingTransport::InProcess { .. } | PendingTransport::Stdio { .. } => false,
PendingTransport::StreamableHttp { .. }
| PendingTransport::StreamableHttpWithOAuth { .. } => true,
| PendingTransport::StreamableHttpWithOAuth { .. }
| PendingTransport::StreamableHttpWithAccessTokenOnly { .. } => true,
};
let mut retry_deadline = timeout.map(|duration| Instant::now() + duration);
let mut pending_transport = Some(initial_transport);
codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs
+313 -7
View File
@@ -1,6 +1,9 @@
mod streamable_http_test_support;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
@@ -51,7 +54,9 @@ const MCP_USER_AGENT: &str = concat!("codex-mcp-client/", env!("CARGO_PKG_VERSIO
const CHILD_SERVER_URL_ENV: &str = "MCP_TEST_OAUTH_STARTUP_SERVER_URL";
const CHILD_HELPER_COMMAND_ENV: &str = "MCP_TEST_OAUTH_STARTUP_HELPER_COMMAND";
const CHILD_RESOURCE_API_KEY_ENV: &str = "MCP_TEST_OAUTH_STARTUP_RESOURCE_API_KEY";
const UNREFRESHABLE_SERVER_URL: &str = "https://unrefreshable.example/mcp";
const CHILD_STORED_ISSUER_ENV: &str = "MCP_TEST_OAUTH_STARTUP_STORED_ISSUER";
const CHILD_ACCESS_TOKEN_EXPIRY_ENV: &str = "MCP_TEST_OAUTH_STARTUP_ACCESS_TOKEN_EXPIRY";
const LEGACY_REFRESHABLE_SERVER_URL: &str = "https://legacy-refreshable.example/mcp";
const UNEXPIRED_SERVER_URL: &str = "https://unexpired.example/mcp";
const REFRESHABLE_SERVER_URL: &str = "https://refreshable.example/mcp";
@@ -62,6 +67,12 @@ enum OAuthStartupScenario {
ProtectedResourceMetadata,
}
#[derive(Clone, Copy)]
enum IssuerMismatchAccessToken {
Expired,
Unexpired,
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refreshes_expired_persisted_token_before_initialize() -> anyhow::Result<()> {
assert_expired_token_refresh(OAuthStartupScenario::DirectAuthorizationMetadata).await
@@ -103,7 +114,7 @@ async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow:
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(1)
.expect(2)
.mount(&server)
.await;
Mock::given(method("GET"))
@@ -112,12 +123,13 @@ async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow:
"resource": resource_url,
"authorization_servers": [server.uri()],
})))
.expect(1)
.expect(2)
.mount(&server)
.await;
}
let mut authorization_metadata = json!({
"issuer": server_url.clone(),
"authorization_endpoint": format!("{}/oauth/authorize", authorization_server.uri()),
"token_endpoint": format!("{}/oauth/token", authorization_server.uri()),
"scopes_supported": [""],
@@ -125,13 +137,17 @@ async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow:
if matches!(scenario, OAuthStartupScenario::ProtectedResourceMetadata) {
authorization_metadata["issuer"] = json!(server.uri());
}
let authorization_server_issuer = authorization_metadata["issuer"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("authorization metadata should include issuer"))?
.to_string();
Mock::given(method("GET"))
.and(path(authorization_metadata_path))
.and(header("user-agent", RESOURCE_USER_AGENT))
.and(header("x-api-key", RESOURCE_API_KEY))
.respond_with(ResponseTemplate::new(200).set_body_json(authorization_metadata))
.expect(1)
.expect(2)
.mount(&server)
.await;
Mock::given(method("POST"))
@@ -208,6 +224,7 @@ async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow:
.args(["oauth_startup_child", "--exact", "--ignored", "--nocapture"])
.env("CODEX_HOME", codex_home.path())
.env(CHILD_SERVER_URL_ENV, server_url)
.env(CHILD_STORED_ISSUER_ENV, authorization_server_issuer)
.env(CHILD_RESOURCE_API_KEY_ENV, RESOURCE_API_KEY)
.env("MCP_TEST_AMBIENT_SECRET", "must-not-reach-helper");
if with_headers_helper {
@@ -252,6 +269,226 @@ async fn assert_expired_token_refresh(scenario: OAuthStartupScenario) -> anyhow:
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn rejects_refresh_when_authorization_server_issuer_changes_before_startup()
-> anyhow::Result<()> {
assert_issuer_mismatch_startup(IssuerMismatchAccessToken::Expired).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn does_not_refresh_unexpired_token_after_issuer_change_401() -> anyhow::Result<()> {
assert_issuer_mismatch_startup(IssuerMismatchAccessToken::Unexpired).await
}
async fn assert_issuer_mismatch_startup(
access_token: IssuerMismatchAccessToken,
) -> anyhow::Result<()> {
let issuer_a = MockServer::start().await;
let issuer_b = MockServer::start().await;
let mcp_server = MockServer::start().await;
let server_url = format!("{}/mcp", mcp_server.uri());
let resource_metadata_url = format!("{}/resource-metadata", mcp_server.uri());
Mock::given(method("GET"))
.and(path("/mcp"))
.respond_with(ResponseTemplate::new(401).insert_header(
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(1)
.mount(&mcp_server)
.await;
Mock::given(method("GET"))
.and(path("/resource-metadata"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"resource": server_url,
"authorization_servers": [issuer_b.uri()],
})))
.expect(1)
.mount(&mcp_server)
.await;
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"issuer": issuer_b.uri(),
"authorization_endpoint": format!("{}/oauth/authorize", issuer_b.uri()),
"token_endpoint": format!("{}/oauth/token", issuer_b.uri()),
"scopes_supported": [""],
})))
.expect(1)
.mount(&issuer_b)
.await;
Mock::given(method("POST"))
.and(path("/mcp"))
.and(header(
"authorization",
format!("Bearer {EXPIRED_ACCESS_TOKEN}"),
))
.respond_with(ResponseTemplate::new(401).insert_header(
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(match access_token {
IssuerMismatchAccessToken::Expired => 0,
IssuerMismatchAccessToken::Unexpired => 1,
})
.mount(&mcp_server)
.await;
let stored_issuer = issuer_a.uri();
let status = run_issuer_startup_child(
&server_url,
&stored_issuer,
match access_token {
IssuerMismatchAccessToken::Expired => "expired",
IssuerMismatchAccessToken::Unexpired => "unexpired",
},
)
.await?;
assert!(
status.success(),
"issuer mismatch startup child failed: {status}"
);
let issuer_b_requests = issuer_b.received_requests().await.unwrap_or_default();
assert!(
issuer_b_requests
.iter()
.all(|request| request.url.path() != "/oauth/token"),
"stored refresh token must not be posted to the replacement issuer"
);
mcp_server.verify().await;
issuer_b.verify().await;
Ok(())
}
async fn run_issuer_startup_child(
server_url: &str,
stored_issuer: &str,
access_token_expiry: &str,
) -> anyhow::Result<std::process::ExitStatus> {
let codex_home = TempDir::new()?;
Ok(Command::new(std::env::current_exe()?)
.args([
"issuer_startup_child",
"--exact",
"--ignored",
"--nocapture",
])
.env("CODEX_HOME", codex_home.path())
.env(CHILD_SERVER_URL_ENV, server_url)
.env(CHILD_STORED_ISSUER_ENV, stored_issuer)
.env(CHILD_ACCESS_TOKEN_EXPIRY_ENV, access_token_expiry)
.status()
.await?)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn does_not_refresh_against_metadata_from_second_discovery() -> anyhow::Result<()> {
let issuer_a = MockServer::start().await;
let issuer_b = MockServer::start().await;
let mcp_server = MockServer::start().await;
let server_url = format!("{}/mcp", mcp_server.uri());
let resource_metadata_url = format!("{}/resource-metadata", mcp_server.uri());
let discovery_count = Arc::new(AtomicUsize::new(0));
let issuer_a_url = issuer_a.uri();
let issuer_b_url = issuer_b.uri();
Mock::given(method("GET"))
.and(path("/mcp"))
.respond_with(ResponseTemplate::new(401).insert_header(
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(1)
.mount(&mcp_server)
.await;
Mock::given(method("GET"))
.and(path("/resource-metadata"))
.respond_with({
let discovery_count = Arc::clone(&discovery_count);
let issuer_a_url = issuer_a_url.clone();
let issuer_b_url = issuer_b_url.clone();
let metadata_resource_url = server_url.clone();
move |_request: &Request| {
let issuer = if discovery_count.fetch_add(1, Ordering::SeqCst) == 0 {
&issuer_a_url
} else {
&issuer_b_url
};
ResponseTemplate::new(200).set_body_json(json!({
"resource": metadata_resource_url,
"authorization_servers": [issuer],
}))
}
})
.expect(1)
.mount(&mcp_server)
.await;
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"issuer": issuer_a.uri(),
"authorization_endpoint": format!("{}/oauth/authorize", issuer_a.uri()),
"token_endpoint": format!("{}/oauth/token", issuer_a.uri()),
"scopes_supported": [""],
})))
.expect(1)
.mount(&issuer_a)
.await;
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"issuer": issuer_b.uri(),
"authorization_endpoint": format!("{}/oauth/authorize", issuer_b.uri()),
"token_endpoint": format!("{}/oauth/token", issuer_b.uri()),
"scopes_supported": [""],
})))
.mount(&issuer_b)
.await;
Mock::given(method("POST"))
.and(path("/mcp"))
.and(header(
"authorization",
format!("Bearer {EXPIRED_ACCESS_TOKEN}"),
))
.respond_with(ResponseTemplate::new(401).insert_header(
"www-authenticate",
format!("Bearer resource_metadata=\"{resource_metadata_url}\""),
))
.expect(1)
.mount(&mcp_server)
.await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.and(body_string_contains(format!(
"refresh_token={REFRESH_TOKEN}"
)))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({
"error": "invalid_grant",
})))
.expect(1)
.mount(&issuer_a)
.await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(ResponseTemplate::new(500))
.expect(0)
.mount(&issuer_b)
.await;
let status = run_issuer_startup_child(&server_url, &issuer_a_url, "unexpired").await?;
assert!(
status.success(),
"metadata swap startup child failed: {status}"
);
mcp_server.verify().await;
issuer_a.verify().await;
issuer_b.verify().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn reports_auth_status_for_persisted_credentials() -> anyhow::Result<()> {
let codex_home = TempDir::new()?;
@@ -326,14 +563,16 @@ async fn persisted_credentials_auth_status_child() -> anyhow::Result<()> {
assert_eq!(status, McpAuthState::LoggedOut(McpLoginRequirement::Login));
first_login_server.verify().await;
let response = OAuthTokenResponse::new(
let mut response = OAuthTokenResponse::new(
AccessToken::new(EXPIRED_ACCESS_TOKEN.to_string()),
BasicTokenType::Bearer,
VendorExtraTokenFields::default(),
);
response.set_refresh_token(Some(RefreshToken::new(REFRESH_TOKEN.to_string())));
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: UNREFRESHABLE_SERVER_URL.to_string(),
url: LEGACY_REFRESHABLE_SERVER_URL.to_string(),
issuer: None,
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at: Some(0),
@@ -345,7 +584,7 @@ async fn persisted_credentials_auth_status_child() -> anyhow::Result<()> {
AuthKeyringBackendKind::default(),
)?;
let status = auth_status(UNREFRESHABLE_SERVER_URL).await?;
let status = auth_status(LEGACY_REFRESHABLE_SERVER_URL).await?;
assert_eq!(
status,
McpAuthState::LoggedOut(McpLoginRequirement::Reauthentication)
@@ -363,6 +602,7 @@ async fn persisted_credentials_auth_status_child() -> anyhow::Result<()> {
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: UNEXPIRED_SERVER_URL.to_string(),
issuer: None,
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at: Some(now.saturating_add(/*rhs*/ 60_000)),
@@ -386,6 +626,7 @@ async fn persisted_credentials_auth_status_child() -> anyhow::Result<()> {
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: REFRESHABLE_SERVER_URL.to_string(),
issuer: Some("https://issuer.example.test".to_string()),
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at: Some(0),
@@ -435,6 +676,7 @@ async fn oauth_startup_child() -> anyhow::Result<()> {
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: server_url.clone(),
issuer: Some(std::env::var(CHILD_STORED_ISSUER_ENV)?),
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at: Some(0),
@@ -493,6 +735,7 @@ async fn expired_unrefreshable_startup_child() -> anyhow::Result<()> {
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: server_url.clone(),
issuer: None,
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at: Some(0),
@@ -523,3 +766,66 @@ async fn expired_unrefreshable_startup_child() -> anyhow::Result<()> {
assert!(is_authentication_required_error(&error));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[ignore = "spawned by issuer startup tests"]
async fn issuer_startup_child() -> anyhow::Result<()> {
let server_url = std::env::var(CHILD_SERVER_URL_ENV)?;
let mut response = OAuthTokenResponse::new(
AccessToken::new(EXPIRED_ACCESS_TOKEN.to_string()),
BasicTokenType::Bearer,
VendorExtraTokenFields::default(),
);
response.set_refresh_token(Some(RefreshToken::new(REFRESH_TOKEN.to_string())));
response.set_expires_in(Some(&Duration::from_secs(7200)));
let expires_at = match std::env::var(CHILD_ACCESS_TOKEN_EXPIRY_ENV).as_deref() {
Ok("expired") | Err(_) => Some(0),
Ok("unexpired") => {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_else(|_| Duration::from_secs(0))
.as_millis() as u64;
Some(now.saturating_add(/*rhs*/ 60_000))
}
Ok(value) => anyhow::bail!("unexpected access token expiry fixture: {value}"),
};
let tokens = StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: server_url.clone(),
issuer: Some(std::env::var(CHILD_STORED_ISSUER_ENV)?),
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(response),
expires_at,
};
save_oauth_tokens(
SERVER_NAME,
&tokens,
OAuthCredentialsStoreMode::File,
AuthKeyringBackendKind::default(),
)?;
let client = RmcpClient::new_streamable_http_client(
SERVER_NAME,
&server_url,
/*bearer_token*/ None,
/*http_headers*/ None,
/*env_http_headers*/ None,
OAuthCredentialsStoreMode::File,
AuthKeyringBackendKind::default(),
Environment::default_for_tests().get_http_client(),
/*auth_provider*/ None,
)
.await;
let error = match client {
Ok(client) => initialize_client(&client)
.await
.expect_err("stored refresh token must fail when startup cannot refresh safely"),
Err(error) => error,
};
assert!(
is_authentication_required_error(&error),
"unexpected issuer startup error: {error:#}"
);
Ok(())
}
codex-rs/rmcp-client/tests/streamable_http_oauth_store_pinning.rs
+1
View File
@@ -308,6 +308,7 @@ fn stored_tokens(server_url: &str, access_token: &str) -> StoredOAuthTokens {
StoredOAuthTokens {
server_name: SERVER_NAME.to_string(),
url: server_url.to_string(),
issuer: None,
client_id: "test-client-id".to_string(),
token_response: WrappedOAuthTokenResponse(OAuthTokenResponse::new(
AccessToken::new(access_token.to_string()),