Serialize MCP OAuth login and logout

codex-rs/cli/src/mcp_cmd.rs

@@ -29,7 +29,7 @@ use codex_mcp::oauth_login_support;
 use codex_mcp::resolve_oauth_scopes;
 use codex_mcp::should_retry_without_scopes;
 use codex_protocol::protocol::McpAuthStatus;
-use codex_rmcp_client::delete_oauth_tokens;
+use codex_rmcp_client::delete_oauth_tokens_locked;
 use codex_rmcp_client::perform_oauth_login;
 use codex_utils_cli::CliConfigOverrides;
 use codex_utils_cli::format_env_display;
@@ -523,12 +523,14 @@ async fn run_logout(config_overrides: &CliConfigOverrides, logout_args: LogoutAr
         _ => bail!("OAuth logout is only supported for streamable_http transports."),
     };
 
-    match delete_oauth_tokens(
+    match delete_oauth_tokens_locked(
         &name,
         &url,
         config.mcp_oauth_credentials_store_mode,
         config.auth_keyring_backend_kind(),
-    ) {
+    )
+    .await
+    {
         Ok(true) => println!("Removed OAuth credentials for '{name}'."),
         Ok(false) => println!("No OAuth credentials stored for '{name}'."),
         Err(err) => return Err(anyhow!("failed to delete OAuth credentials: {err}")),

codex-rs/cli/tests/mcp_oauth_logout.rs

@@ -0,0 +1,80 @@
+use std::path::Path;
+
+use anyhow::Result;
+use codex_config::types::AuthKeyringBackendKind;
+use codex_config::types::OAuthCredentialsStoreMode;
+use codex_rmcp_client::StoredOAuthTokens;
+use codex_rmcp_client::save_oauth_tokens;
+use predicates::str::contains;
+use serde_json::json;
+use tempfile::TempDir;
+
+const SERVER_NAME: &str = "oauth-server";
+const SERVER_URL: &str = "https://example.com/mcp";
+
+fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
+    let mut cmd = assert_cmd::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?);
+    cmd.env("CODEX_HOME", codex_home);
+    Ok(cmd)
+}
+
+#[tokio::test]
+async fn mcp_logout_cli_removes_file_credentials() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let status = tokio::process::Command::new(std::env::current_exe()?)
+        .args([
+            "mcp_logout_cli_child",
+            "--exact",
+            "--ignored",
+            "--nocapture",
+        ])
+        .env("CODEX_HOME", codex_home.path())
+        .status()
+        .await?;
+    anyhow::ensure!(status.success(), "MCP logout child failed: {status}");
+    Ok(())
+}
+
+#[tokio::test]
+#[ignore = "spawned by mcp_logout_cli_removes_file_credentials"]
+async fn mcp_logout_cli_child() -> Result<()> {
+    let codex_home = std::env::var("CODEX_HOME")?;
+    let codex_home = Path::new(&codex_home);
+    std::fs::write(
+        codex_home.join("config.toml"),
+        format!(
+            "mcp_oauth_credentials_store = \"file\"\n\n[mcp_servers.{SERVER_NAME}]\nurl = \"{SERVER_URL}\"\n"
+        ),
+    )?;
+
+    let tokens: StoredOAuthTokens = serde_json::from_value(json!({
+        "server_name": SERVER_NAME,
+        "url": SERVER_URL,
+        "client_id": "test-client-id",
+        "token_response": {
+            "access_token": "access-token",
+            "token_type": "bearer",
+            "expires_in": 3600,
+            "refresh_token": "refresh-token",
+        },
+        "expires_at": null,
+    }))?;
+    save_oauth_tokens(
+        SERVER_NAME,
+        &tokens,
+        OAuthCredentialsStoreMode::File,
+        AuthKeyringBackendKind::default(),
+    )?;
+    assert!(codex_home.join(".credentials.json").exists());
+
+    let mut logout = codex_command(codex_home)?;
+    logout
+        .args(["mcp", "logout", SERVER_NAME])
+        .assert()
+        .success()
+        .stdout(contains(format!(
+            "Removed OAuth credentials for '{SERVER_NAME}'."
+        )));
+    assert!(!codex_home.join(".credentials.json").exists());
+    Ok(())
+}

codex-rs/config/src/types.rs

@@ -103,7 +103,9 @@ pub enum AuthCredentialsStoreMode {
 #[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
 #[serde(rename_all = "lowercase")]
 pub enum OAuthCredentialsStoreMode {
-    /// `Keyring` when available; otherwise, `File`.
+    /// Prefer `Keyring` and use `File` when keyring storage is unavailable.
+    /// Once an MCP client loads credentials from one store, that client keeps the resolved store
+    /// for its lifetime so refreshes cannot switch to a possibly stale credential source.
     /// Credentials stored in the keyring will only be readable by Codex unless the user explicitly grants access via OS-level keyring access.
     #[default]
     Auto,

codex-rs/core/config.schema.json

@@ -2129,7 +2129,7 @@
       "description": "Determine where Codex should store and read MCP credentials.",
       "oneOf": [
         {
-          "description": "`Keyring` when available; otherwise, `File`. Credentials stored in the keyring will only be readable by Codex unless the user explicitly grants access via OS-level keyring access.",
+          "description": "Prefer `Keyring` and use `File` when keyring storage is unavailable. Once an MCP client loads credentials from one store, that client keeps the resolved store for its lifetime so refreshes cannot switch to a possibly stale credential source. Credentials stored in the keyring will only be readable by Codex unless the user explicitly grants access via OS-level keyring access.",
           "enum": [
             "auto"
           ],

codex-rs/rmcp-client/Cargo.toml

@@ -84,5 +84,10 @@ keyring = { workspace = true, features = ["windows-native"] }
 [target.'cfg(any(target_os = "freebsd", target_os = "openbsd"))'.dependencies]
 keyring = { workspace = true, features = ["sync-secret-service"] }
 
+# This test is compiled through `#[path]` inside the inline `oauth::tests` module. Cargo-shear
+# cannot resolve that nested module path and otherwise reports the linked file as unlinked.
+[package.metadata.cargo-shear]
+ignored-paths = ["src/oauth/tests/persistor_tests.rs"]
+
 [lib]
 doctest = false

codex-rs/rmcp-client/src/http_client_adapter.rs

@@ -22,6 +22,7 @@ use codex_exec_server::HttpResponseBodyStream;
 use futures::StreamExt;
 use futures::stream;
 use futures::stream::BoxStream;
+use oauth2::AccessToken;
 use reqwest::StatusCode;
 use reqwest::header::ACCEPT;
 use reqwest::header::AUTHORIZATION;
@@ -55,12 +56,17 @@ pub(crate) struct StreamableHttpClientAdapter {
     http_client: Arc<dyn HttpClient>,
     default_headers: HeaderMap,
     auth_provider: Option<SharedAuthProvider>,
+    attribute_rejected_access_token: bool,
 }
 
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum StreamableHttpClientAdapterError {
     #[error("streamable HTTP session expired with 404 Not Found")]
     SessionExpired404,
+    #[error("MCP server rejected the access token with HTTP 401 Unauthorized")]
+    AccessTokenRejected { rejected_access_token: AccessToken },
+    #[error("MCP OAuth operation failed: {0:#}")]
+    OAuth(#[source] anyhow::Error),
     #[error(transparent)]
     HttpRequest(#[from] ExecServerError),
     #[error("invalid HTTP header: {0}")]
@@ -77,8 +83,15 @@ impl StreamableHttpClientAdapter {
             http_client,
             default_headers,
             auth_provider,
+            attribute_rejected_access_token: false,
         }
     }
+
+    /// Preserves the access token associated with a 401 for Codex-owned OAuth recovery.
+    pub(crate) fn with_rejected_token_attribution(mut self) -> Self {
+        self.attribute_rejected_access_token = true;
+        self
+    }
 }
 
 impl StreamableHttpClient for StreamableHttpClientAdapter {
@@ -109,7 +122,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
             JSON_MIME_TYPE.to_string(),
             StreamableHttpClientAdapterError::Header,
         )?;
-        if let Some(auth_token) = auth_token {
+        if let Some(auth_token) = auth_token.as_deref() {
             insert_header(
                 &mut headers,
                 AUTHORIZATION,
@@ -162,6 +175,12 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
                 StreamableHttpClientAdapterError::SessionExpired404,
             ));
         }
+        if self.attribute_rejected_access_token
+            && response.status == StatusCode::UNAUTHORIZED.as_u16()
+            && let Some(error) = access_token_rejected(auth_token.as_deref())
+        {
+            return Err(error);
+        }
         if response.status == StatusCode::UNAUTHORIZED.as_u16()
             && let Some(header) =
                 response_header(&response.headers, reqwest::header::WWW_AUTHENTICATE)
@@ -240,7 +259,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
         let mut headers = self.default_headers.clone();
         headers.extend(custom_headers);
         self.add_auth_headers(&mut headers);
-        if let Some(auth_token) = auth_token {
+        if let Some(auth_token) = auth_token.as_deref() {
             insert_header(
                 &mut headers,
                 AUTHORIZATION,
@@ -274,6 +293,12 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
         if response.status == StatusCode::METHOD_NOT_ALLOWED.as_u16() {
             return Ok(());
         }
+        if self.attribute_rejected_access_token
+            && response.status == StatusCode::UNAUTHORIZED.as_u16()
+            && let Some(error) = access_token_rejected(auth_token.as_deref())
+        {
+            return Err(error);
+        }
         if !status_is_success(response.status) {
             return Err(StreamableHttpError::UnexpectedServerResponse(
                 format!("DELETE returned HTTP {}", response.status).into(),
@@ -316,7 +341,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
                 StreamableHttpClientAdapterError::Header,
             )?;
         }
-        if let Some(auth_token) = auth_token {
+        if let Some(auth_token) = auth_token.as_deref() {
             insert_header(
                 &mut headers,
                 AUTHORIZATION,
@@ -349,6 +374,12 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
                 StreamableHttpClientAdapterError::SessionExpired404,
             ));
         }
+        if self.attribute_rejected_access_token
+            && response.status == StatusCode::UNAUTHORIZED.as_u16()
+            && let Some(error) = access_token_rejected(auth_token.as_deref())
+        {
+            return Err(error);
+        }
         if !status_is_success(response.status) {
             return Err(StreamableHttpError::UnexpectedServerResponse(
                 format!("GET returned HTTP {}", response.status).into(),
@@ -371,6 +402,20 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
     }
 }
 
+fn access_token_rejected(
+    auth_token: Option<&str>,
+) -> Option<StreamableHttpError<StreamableHttpClientAdapterError>> {
+    // Preserve the token associated with this response. Reading the current credential after a
+    // delayed 401 is racy: another concurrent request may already have refreshed A to B, in which
+    // case recovery must retry B rather than refresh B a second time. AccessToken's Debug
+    // implementation redacts the secret if this error is logged.
+    auth_token.map(|rejected_access_token| {
+        StreamableHttpError::Client(StreamableHttpClientAdapterError::AccessTokenRejected {
+            rejected_access_token: AccessToken::new(rejected_access_token.to_string()),
+        })
+    })
+}
+
 impl StreamableHttpClientAdapter {
     fn add_auth_headers(&self, headers: &mut HeaderMap) {
         if let Some(auth_provider) = &self.auth_provider {

codex-rs/rmcp-client/src/lib.rs

@@ -6,6 +6,7 @@ mod in_process_transport;
 mod logging_client_handler;
 mod oauth;
 mod oauth_http_client;
+mod oauth_transport;
 mod perform_oauth_login;
 mod program_resolver;
 mod rmcp_client;
@@ -26,7 +27,7 @@ pub use in_process_transport::InProcessTransportFactory;
 pub use oauth::StoredOAuthTokens;
 pub use oauth::WrappedOAuthTokenResponse;
 pub use oauth::delete_oauth_tokens;
-pub(crate) use oauth::load_oauth_tokens;
+pub use oauth::delete_oauth_tokens_locked;
 pub use oauth::save_oauth_tokens;
 pub use perform_oauth_login::OAuthProviderError;
 pub use perform_oauth_login::OauthLoginHandle;