diff --git a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs index 1ff52f88ee..b0a32df15c 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs @@ -4,6 +4,7 @@ use app_test_support::TestAppServer; use axum::Json; use axum::Router; use axum::body::Bytes; +use axum::response::IntoResponse; use axum::routing::get; use axum::routing::post; use codex_app_server_protocol::CapabilityRootLocation; @@ -43,10 +44,14 @@ use rmcp::transport::streamable_http_server::session::local::LocalSessionManager use serde_json::json; use std::borrow::Cow; use std::collections::BTreeMap; +use std::process::Stdio; use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; +use tokio::io::AsyncBufReadExt; +use tokio::io::BufReader; use tokio::net::TcpListener; +use tokio::process::Command; use tokio::sync::mpsc; use tokio::time::timeout; use url::Url; @@ -62,12 +67,15 @@ const HOST_OAUTH_ACCESS_TOKEN: &str = "host-access-token"; const EXECUTOR_OAUTH_ACCESS_TOKEN: &str = "executor-access-token"; const EXECUTOR_ENV_NAME: &str = "MCP_EXECUTOR_MARKER"; const EXECUTOR_ENV_VALUE: &str = "executor-only"; +const EXECUTOR_HTTP_AUTH_ENV_NAME: &str = "NODE_REPL_AUTH_TOKEN"; +const EXECUTOR_HTTP_AUTH_ENV_VALUE: &str = "executor-only-http-token"; const EXECUTOR_ID: &str = "executor-1"; const REFRESH_PROBE_SERVER_NAME: &str = "refresh_probe"; const TOOL_CALL_ID: &str = "executor-mcp-call"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Result<()> { + assert!(std::env::var_os(EXECUTOR_HTTP_AUTH_ENV_NAME).is_none()); let responses_server = responses::start_mock_server().await; let http_listener = TcpListener::bind("127.0.0.1:0").await?; let http_addr = http_listener.local_addr()?; @@ -83,6 +91,24 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu Arc::new(LocalSessionManager::default()), http_server_config, ); + let http_mcp_router = + Router::new() + .nest_service("/mcp", http_mcp_service) + .layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| async move { + let authorized = request + .headers() + .get(axum::http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| { + value == format!("Bearer {EXECUTOR_HTTP_AUTH_ENV_VALUE}") + }); + if !authorized { + return axum::http::StatusCode::UNAUTHORIZED.into_response(); + } + next.run(request).await + }, + )); let (oauth_authorization_tx, mut oauth_authorization_rx) = mpsc::unbounded_channel(); let oauth_mcp_router = Router::new() .nest_service("/oauth-mcp", oauth_mcp_service) @@ -147,7 +173,7 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu } }), ) - .nest_service("/mcp", http_mcp_service) + .merge(http_mcp_router) .merge(oauth_mcp_router); let http_server_handle = tokio::spawn(async move { let _ = axum::serve(http_listener, http_router).await; @@ -183,12 +209,20 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu &oauth_credentials_path, serde_json::to_vec(&json!({"host": host_oauth_credential.clone()}))?, )?; - let codex_bin = toml::Value::String( - codex_utils_cargo_bin::cargo_bin("codex")? - .to_string_lossy() - .into_owned(), - ); - let http_proxy = toml::Value::String(format!("http://{http_addr}")); + let mut executor = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?) + .args(["exec-server", "--listen", "ws://127.0.0.1:0"]) + .stdout(Stdio::piped()) + .kill_on_drop(true) + .env("CODEX_HOME", codex_home.path()) + .env(EXECUTOR_ENV_NAME, EXECUTOR_ENV_VALUE) + .env(EXECUTOR_HTTP_AUTH_ENV_NAME, EXECUTOR_HTTP_AUTH_ENV_VALUE) + .env("HTTP_PROXY", format!("http://{http_addr}")) + .spawn()?; + let stdout = executor.stdout.take().expect("executor stdout is piped"); + let mut lines = BufReader::new(stdout).lines(); + let executor_url = timeout(DEFAULT_READ_TIMEOUT, lines.next_line()) + .await?? + .expect("executor emits its websocket URL"); std::fs::write( codex_home.path().join("environments.toml"), format!( @@ -197,11 +231,7 @@ include_local = true [[environments]] id = "{EXECUTOR_ID}" -program = {codex_bin} -args = ["exec-server", "--listen", "stdio"] -[environments.env] -{EXECUTOR_ENV_NAME} = "{EXECUTOR_ENV_VALUE}" -HTTP_PROXY = {http_proxy} +url = "{executor_url}" "# ), )?; @@ -224,6 +254,7 @@ HTTP_PROXY = {http_proxy} (HTTP_MCP_SERVER_NAME): { "url": EXECUTOR_HTTP_MCP_URL, "environment_id": "local", + "bearer_token_env_var": EXECUTOR_HTTP_AUTH_ENV_NAME, "startup_timeout_sec": 10, }, (OAUTH_MCP_SERVER_NAME): { diff --git a/codex-rs/codex-mcp/src/executor_environment_http_client.rs b/codex-rs/codex-mcp/src/executor_environment_http_client.rs new file mode 100644 index 0000000000..c2d50617c6 --- /dev/null +++ b/codex-rs/codex-mcp/src/executor_environment_http_client.rs @@ -0,0 +1,45 @@ +use std::sync::Arc; + +use codex_exec_server::ExecServerError; +use codex_exec_server::HttpClient; +use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRequestParams; +use codex_exec_server::HttpRequestResponse; +use codex_exec_server::HttpResponseBodyStream; +use futures::future::BoxFuture; + +pub(crate) struct ExecutorEnvironmentHttpClient { + pub(crate) bearer_token_env_var: String, + pub(crate) http_client: Arc, +} + +impl ExecutorEnvironmentHttpClient { + fn attach_authorization(&self, params: &mut HttpRequestParams) { + params + .headers + .retain(|header| !header.name.eq_ignore_ascii_case("authorization")); + params.headers.push(HttpHeader { + name: "authorization".to_string(), + value: "Bearer ".to_string(), + value_env_var: Some(self.bearer_token_env_var.clone()), + }); + } +} + +impl HttpClient for ExecutorEnvironmentHttpClient { + fn http_request( + &self, + mut params: HttpRequestParams, + ) -> BoxFuture<'_, Result> { + self.attach_authorization(&mut params); + self.http_client.http_request(params) + } + + fn http_request_stream( + &self, + mut params: HttpRequestParams, + ) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> { + self.attach_authorization(&mut params); + self.http_client.http_request_stream(params) + } +} diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 8a613671bc..6066839591 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -99,6 +99,7 @@ mod client_capabilities; pub(crate) mod codex_apps; pub(crate) mod connection_manager; pub(crate) mod elicitation; +mod executor_environment_http_client; pub(crate) mod mcp; mod openai_docs_source_attribution; mod pagination; diff --git a/codex-rs/codex-mcp/src/plugin_config.rs b/codex-rs/codex-mcp/src/plugin_config.rs index eadc12de54..e6fc333d86 100644 --- a/codex-rs/codex-mcp/src/plugin_config.rs +++ b/codex-rs/codex-mcp/src/plugin_config.rs @@ -183,22 +183,13 @@ fn bind_environment_env_vars(config: &mut McpServerConfig) -> Result<(), String> let is_local_environment = config.is_local_environment(); let env_vars = match &mut config.transport { McpServerTransportConfig::Stdio { env_vars, .. } => env_vars, - // Never resolve executor-owned environment references in the host process. - // Remove this rejection once the owning executor resolves these fields. + // Bearer credentials resolve on the executor; other header variables do not yet. McpServerTransportConfig::StreamableHttp { - bearer_token_env_var, - env_http_headers, - .. + env_http_headers, .. } => { if is_local_environment { return Ok(()); } - if bearer_token_env_var.is_some() { - return Err( - "`bearer_token_env_var` requires executor-side environment resolution for an executor-owned HTTP MCP" - .to_string(), - ); - } if env_http_headers .as_ref() .is_some_and(|headers| !headers.is_empty()) diff --git a/codex-rs/codex-mcp/src/plugin_config_tests.rs b/codex-rs/codex-mcp/src/plugin_config_tests.rs index c2c50b277f..57012c9249 100644 --- a/codex-rs/codex-mcp/src/plugin_config_tests.rs +++ b/codex-rs/codex-mcp/src/plugin_config_tests.rs @@ -786,7 +786,7 @@ fn environment_placement_rejects_orchestrator_env_vars() { } #[test] -fn remote_environment_placement_rejects_http_env_references() { +fn remote_environment_placement_preserves_bearer_and_rejects_header_env_references() { let plugin_root = plugin_root(); let outcome = parse_executor_plugin_mcp_config( &plugin_root_uri(&plugin_root), @@ -807,19 +807,20 @@ fn remote_environment_placement_rejects_http_env_references() { assert_eq!( outcome, PluginMcpConfigParseOutcome { - servers: BTreeMap::new(), - errors: vec![ - PluginMcpServerParseError { - name: "bearer".to_string(), - message: "`bearer_token_env_var` requires executor-side environment resolution for an executor-owned HTTP MCP" - .to_string(), - }, - PluginMcpServerParseError { - name: "headers".to_string(), - message: "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP" - .to_string(), - }, - ], + servers: BTreeMap::from([( + "bearer".to_string(), + serde_json::from_value(serde_json::json!({ + "url": "https://example.com/bearer", + "bearer_token_env_var": "TOKEN", + "environment_id": "executor-1", + })) + .expect("executor-owned bearer configuration"), + )]), + errors: vec![PluginMcpServerParseError { + name: "headers".to_string(), + message: "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP" + .to_string(), + }], } ); } diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 42e17bf6ce..157acf0c84 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -23,6 +23,7 @@ use crate::codex_apps::normalize_codex_apps_callable_namespace; use crate::codex_apps::normalize_codex_apps_tool_title; use crate::codex_apps::prepare_openai_file_params_for_model; use crate::elicitation::ElicitationRequestManager; +use crate::executor_environment_http_client::ExecutorEnvironmentHttpClient; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::ToolPluginProvenance; use crate::openai_docs_source_attribution::maybe_with_openai_docs_source_attribution; @@ -59,6 +60,7 @@ use codex_rmcp_client::LocalStdioServerLauncher; use codex_rmcp_client::McpProtocolMode; use codex_rmcp_client::RmcpClient; use codex_rmcp_client::StdioServerLauncher; +use codex_rmcp_client::StreamableHttpBearerToken; use codex_rmcp_client::StreamableHttpRedirectMode; use codex_rmcp_client::ToolWithConnectorId; use codex_rmcp_client::is_authentication_required_error; @@ -1137,10 +1139,20 @@ async fn make_rmcp_client( .http_client_for_server(server.config(), resolved_environment.as_ref()) .map_err(|error| StartupOutcomeError::from(anyhow!(error)))?; let http_client = maybe_with_openai_docs_source_attribution(&url, http_client); - let resolved_bearer_token = - match resolve_bearer_token(server_name, bearer_token_env_var.as_deref()) { - Ok(token) => token, - Err(error) => return Err(error.into()), + let (http_client, resolved_bearer_token) = + if !is_local_environment && let Some(env_var) = bearer_token_env_var.as_ref() { + ( + Arc::new(ExecutorEnvironmentHttpClient { + bearer_token_env_var: env_var.clone(), + http_client, + }) as Arc, + Some(StreamableHttpBearerToken::ProvidedByHttpClient), + ) + } else { + let token = resolve_bearer_token(server_name, bearer_token_env_var.as_deref()) + .map_err(StartupOutcomeError::from)? + .map(StreamableHttpBearerToken::Resolved); + (http_client, token) }; let redirect_mode = if server.is_agent_plugin() { StreamableHttpRedirectMode::AgentPluginV1 diff --git a/codex-rs/codex-mcp/src/server.rs b/codex-rs/codex-mcp/src/server.rs index 2416d7b6dc..f0e739a904 100644 --- a/codex-rs/codex-mcp/src/server.rs +++ b/codex-rs/codex-mcp/src/server.rs @@ -329,6 +329,9 @@ fn referenced_environment_variables(config: &McpServerConfig) -> Vec<(String, Op .filter(|env_var| !env_var.is_remote_source()) .map(|env_var| env_var.name().to_string()) .collect::>(), + McpServerTransportConfig::StreamableHttp { .. } if !config.is_local_environment() => { + Vec::new() + } McpServerTransportConfig::StreamableHttp { bearer_token_env_var, env_http_headers, diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index d55b29ef75..5c1f978d78 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -677,8 +677,11 @@ impl ExecutorCapabilityDiscoverySnapshot { pub struct HttpHeader { /// Header name as it appears on the HTTP wire. pub name: String, - /// Header value after UTF-8 conversion. + /// Literal header value, or prefix for an executor-local environment value. pub value: String, + /// Environment variable resolved by the process that sends the HTTP request. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value_env_var: Option, } /// Redirect behavior for an executor-side HTTP request. diff --git a/codex-rs/exec-server/src/client/route_aware_http_client.rs b/codex-rs/exec-server/src/client/route_aware_http_client.rs index 59037a0edd..81dd4607dd 100644 --- a/codex-rs/exec-server/src/client/route_aware_http_client.rs +++ b/codex-rs/exec-server/src/client/route_aware_http_client.rs @@ -12,6 +12,10 @@ use codex_http_client::ClientRouteClass; use codex_http_client::HttpClientFactory; use codex_http_client::RouteAwareClientPool; use codex_http_client::RouteAwareRequestError; +use codex_protocol::shell_environment::CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR; +use codex_protocol::shell_environment::OPENAI_FEDERATION_RULE_ID_ENV_VAR; +use codex_protocol::shell_environment::OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR; +use codex_protocol::shell_environment::OPENAI_WORKLOAD_IDENTITY_CONTEXT_ENV_VAR; use futures::FutureExt; use futures::StreamExt; use futures::future::BoxFuture; @@ -36,6 +40,23 @@ use crate::rpc::RpcNotificationSender; use crate::rpc::internal_error; use crate::rpc::invalid_params; +const HTTP_HEADER_ENV_DENYLIST: &[&str] = &[ + CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, + OPENAI_FEDERATION_RULE_ID_ENV_VAR, + OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR, + OPENAI_WORKLOAD_IDENTITY_CONTEXT_ENV_VAR, + "OPENAI_API_KEY", + "CODEX_API_KEY", + "CODEX_ACCESS_TOKEN", + "CODEX_CONNECTORS_TOKEN", + "AWS_ACCESS_KEY_ID", + "AWS_SECRET_ACCESS_KEY", + "AWS_SESSION_TOKEN", + "AZURE_CLIENT_SECRET", + "AZURE_FEDERATED_TOKEN_FILE", + "GOOGLE_APPLICATION_CREDENTIALS", +]; + /// HTTP capability implementation backed by the shared route-aware transport. #[derive(Clone)] pub struct RouteAwareHttpClient { @@ -288,7 +309,34 @@ impl RouteAwareHttpRequestRunner { let name = HeaderName::from_bytes(header.name.as_bytes()).map_err(|error| { invalid_params(format!("http/request header name is invalid: {error}")) })?; - let value = HeaderValue::from_str(&header.value).map_err(|error| { + let value = match header.value_env_var { + Some(env_var) => { + if HTTP_HEADER_ENV_DENYLIST + .iter() + .any(|denied| denied.eq_ignore_ascii_case(&env_var)) + { + return Err(invalid_params(format!( + "http/request header {} cannot use executor environment variable {env_var}", + header.name + ))); + } + let env_value = std::env::var(&env_var).map_err(|_| { + invalid_params(format!( + "http/request header {} requires executor environment variable {env_var}", + header.name + )) + })?; + if env_value.is_empty() { + return Err(invalid_params(format!( + "http/request header {} requires a non-empty executor environment variable {env_var}", + header.name + ))); + } + format!("{}{env_value}", header.value) + } + None => header.value, + }; + let value = HeaderValue::from_str(&value).map_err(|error| { invalid_params(format!( "http/request header value is invalid for {}: {error}", header.name @@ -306,6 +354,7 @@ impl RouteAwareHttpRequestRunner { Some(HttpHeader { name: name.as_str().to_string(), value: value.to_str().ok()?.to_string(), + value_env_var: None, }) }) .collect() diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs index 4c91a2232d..b10cda7e7d 100644 --- a/codex-rs/exec-server/tests/http_client.rs +++ b/codex-rs/exec-server/tests/http_client.rs @@ -136,6 +136,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: vec![HttpHeader { name: "accept".to_string(), value: "text/event-stream".to_string(), + value_env_var: None, }], body: None, timeout_ms: None, @@ -154,6 +155,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: vec![HttpHeader { name: "content-type".to_string(), value: "text/event-stream".to_string(), + value_env_var: None, }], body: Vec::new().into(), }, @@ -222,6 +224,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: vec![HttpHeader { name: "accept".to_string(), value: "text/event-stream".to_string(), + value_env_var: None, }], body: None, timeout_ms: None, @@ -239,6 +242,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: vec![HttpHeader { name: "content-type".to_string(), value: "text/event-stream".to_string(), + value_env_var: None, }], body: Vec::new().into(), } diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs index 4713f8a114..55144de11e 100644 --- a/codex-rs/exec-server/tests/http_request.rs +++ b/codex-rs/exec-server/tests/http_request.rs @@ -49,7 +49,14 @@ struct CapturedHttpRequest { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> { // Phase 1: start exec-server and complete the JSON-RPC handshake. - let mut server = exec_server().await?; + let mut server = exec_server_with_env( + [ + ("NODE_REPL_AUTH_TOKEN", "executor-token"), + ("LINEAR_API_KEY", "linear-token"), + ], + &[], + ) + .await?; initialize_exec_server(&mut server).await?; // Phase 2: start a local HTTP peer and ask exec-server to POST to it. @@ -61,10 +68,23 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> serde_json::to_value(HttpRequestParams { method: "POST".to_string(), url, - headers: vec![HttpHeader { - name: "x-codex-test".to_string(), - value: "buffered".to_string(), - }], + headers: vec![ + HttpHeader { + name: "x-codex-test".to_string(), + value: "buffered".to_string(), + value_env_var: None, + }, + HttpHeader { + name: "authorization".to_string(), + value: "Bearer ".to_string(), + value_env_var: Some("NODE_REPL_AUTH_TOKEN".to_string()), + }, + HttpHeader { + name: "x-linear-authorization".to_string(), + value: "Bearer ".to_string(), + value_env_var: Some("LINEAR_API_KEY".to_string()), + }, + ], body: Some(b"request-body".to_vec().into()), timeout_ms: Some(5_000), redirect_policy: HttpRedirectPolicy::Follow, @@ -81,11 +101,18 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> ( captured.request_line.as_str(), captured.headers.get("x-codex-test").map(String::as_str), + captured.headers.get("authorization").map(String::as_str), + captured + .headers + .get("x-linear-authorization") + .map(String::as_str), captured.body.as_slice(), ), ( "POST /mcp?case=buffered HTTP/1.1", Some("buffered"), + Some("Bearer executor-token"), + Some("Bearer linear-token"), b"request-body".as_slice(), ) ); @@ -113,6 +140,70 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_rejects_protected_environment_headers() -> anyhow::Result<()> { + let mut server = exec_server_with_env( + [( + "CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN", + "executor-internal-token", + )], + &[], + ) + .await?; + initialize_exec_server(&mut server).await?; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + for (index, env_var) in [ + "CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN", + "codex_exec_server_noise_auth_token", + "OPENAI_API_KEY", + "CODEX_ACCESS_TOKEN", + "CODEX_CONNECTORS_TOKEN", + "AWS_SECRET_ACCESS_KEY", + "AZURE_FEDERATED_TOKEN_FILE", + "OPENAI_IDENTITY_TOKEN_FILE", + ] + .into_iter() + .enumerate() + { + let request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: format!("http://{}/mcp", listener.local_addr()?), + headers: vec![HttpHeader { + name: "authorization".to_string(), + value: "Bearer ".to_string(), + value_env_var: Some(env_var.to_string()), + }], + body: None, + timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Follow, + request_id: format!("protected-header-request-{index}"), + stream_response: false, + })?, + ) + .await?; + let error = wait_for_error_response(&mut server, request_id).await?; + assert_eq!(error.code, -32602); + assert_eq!( + error.message, + format!( + "http/request header authorization cannot use executor environment variable {env_var}" + ) + ); + } + assert!( + timeout(Duration::from_millis(50), listener.accept()) + .await + .is_err() + ); + + server.shutdown().await?; + Ok(()) +} + /// What this tests: delegated HTTP accepts URLs with client-side fragments and /// does not include those fragments in the request sent over the network. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -417,6 +508,7 @@ async fn exec_server_http_request_streams_response_body_notifications() -> anyho headers: vec![HttpHeader { name: "accept".to_string(), value: "text/event-stream".to_string(), + value_env_var: None, }], body: None, timeout_ms: Some(5_000), diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index e9f07f9865..0d31522cc6 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -711,6 +711,7 @@ fn protocol_headers(headers: &HeaderMap) -> Vec { Some(HttpHeader { name: name.as_str().to_string(), value: std::str::from_utf8(value.as_bytes()).ok()?.to_string(), + value_env_var: None, }) }) .collect() diff --git a/codex-rs/rmcp-client/src/http_client_adapter/www_authenticate_tests.rs b/codex-rs/rmcp-client/src/http_client_adapter/www_authenticate_tests.rs index 6696655d88..7ecfae9560 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter/www_authenticate_tests.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter/www_authenticate_tests.rs @@ -107,10 +107,12 @@ fn selects_bearer_challenge_from_a_later_www_authenticate_field_value() { HttpHeader { name: "www-authenticate".to_string(), value: r#"Basic realm="example""#.to_string(), + value_env_var: None, }, HttpHeader { name: "WWW-Authenticate".to_string(), value: r#"Bearer error="insufficient_scope", scope="files:read""#.to_string(), + value_env_var: None, }, ]; diff --git a/codex-rs/rmcp-client/src/http_client_adapter_tests.rs b/codex-rs/rmcp-client/src/http_client_adapter_tests.rs index 7452f5b400..c3effd9983 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter_tests.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter_tests.rs @@ -25,6 +25,7 @@ fn protocol_headers_preserve_utf8_values() { vec![HttpHeader { name: "x-plugin-name".to_string(), value: "cafĂ©".to_string(), + value_env_var: None, }] ); } diff --git a/codex-rs/rmcp-client/src/http_client_redirect.rs b/codex-rs/rmcp-client/src/http_client_redirect.rs index fa73b66b36..5d8cfee05f 100644 --- a/codex-rs/rmcp-client/src/http_client_redirect.rs +++ b/codex-rs/rmcp-client/src/http_client_redirect.rs @@ -166,6 +166,7 @@ impl SameOriginRedirectHttpClient { params.headers.push(HttpHeader { name: "referer".to_string(), value: referer.to_string(), + value_env_var: None, }); } diff --git a/codex-rs/rmcp-client/src/http_client_redirect_tests.rs b/codex-rs/rmcp-client/src/http_client_redirect_tests.rs index f3bb42cfdb..966fcfcfcb 100644 --- a/codex-rs/rmcp-client/src/http_client_redirect_tests.rs +++ b/codex-rs/rmcp-client/src/http_client_redirect_tests.rs @@ -38,6 +38,7 @@ fn headers(headers: [(&str, &str); N]) -> Vec { .map(|(name, value)| HttpHeader { name: name.to_string(), value: value.to_string(), + value_env_var: None, }) .collect() } diff --git a/codex-rs/rmcp-client/src/http_headers.rs b/codex-rs/rmcp-client/src/http_headers.rs index 230443cd1d..12118e50f6 100644 --- a/codex-rs/rmcp-client/src/http_headers.rs +++ b/codex-rs/rmcp-client/src/http_headers.rs @@ -185,6 +185,7 @@ impl HttpHeadersClient { value: std::str::from_utf8(value.as_bytes()) .map_err(|error| ExecServerError::HttpRequest(error.to_string()))? .to_string(), + value_env_var: None, }); } if let Some(deadline) = deadline { diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index 32ccd8ef28..396e1e4853 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -52,6 +52,7 @@ pub use rmcp_client::ElicitationResponse; pub use rmcp_client::ListToolsWithConnectorIdResult; pub use rmcp_client::RmcpClient; pub use rmcp_client::SendElicitation; +pub use rmcp_client::StreamableHttpBearerToken; pub use rmcp_client::ToolWithConnectorId; pub use startup_error::is_authentication_required_error; pub use stdio_server_launcher::ExecutorStdioServerLauncher; diff --git a/codex-rs/rmcp-client/src/oauth_http_client.rs b/codex-rs/rmcp-client/src/oauth_http_client.rs index d4e15debda..6a6429529b 100644 --- a/codex-rs/rmcp-client/src/oauth_http_client.rs +++ b/codex-rs/rmcp-client/src/oauth_http_client.rs @@ -169,6 +169,7 @@ impl OAuthHttpClientAdapter { Ok(HttpHeader { name: name.as_str().to_string(), value: value.to_str().map_err(oauth_http_client_error)?.to_string(), + value_env_var: None, }) }) .collect::, OAuthHttpClientError>>()?; diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index e6ffc9e0f1..0dd10638df 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -130,6 +130,15 @@ enum ClientState { Closed, } +/// Bearer authentication applied directly or by the selected HTTP transport. +#[derive(Clone)] +pub enum StreamableHttpBearerToken { + /// A token already resolved in the current process. + Resolved(String), + /// The HTTP client attaches credentials when it sends each request. + ProvidedByHttpClient, +} + #[derive(Clone)] enum TransportRecipe { InProcess { @@ -142,7 +151,7 @@ enum TransportRecipe { StreamableHttp { server_name: String, url: String, - bearer_token: Option, + bearer_token: Option, http_headers: Option>, env_http_headers: Option>, store_mode: OAuthCredentialsStoreMode, @@ -514,7 +523,7 @@ impl RmcpClient { Self::new_streamable_http_client_with_protocol_mode_and_redirect_mode( server_name, url, - bearer_token, + bearer_token.map(StreamableHttpBearerToken::Resolved), http_headers, env_http_headers, store_mode, @@ -531,7 +540,7 @@ impl RmcpClient { pub async fn new_streamable_http_client_with_protocol_mode_and_redirect_mode( server_name: &str, url: &str, - bearer_token: Option, + bearer_token: Option, http_headers: Option>, env_http_headers: Option>, store_mode: OAuthCredentialsStoreMode, @@ -1012,7 +1021,10 @@ impl RmcpClient { redirect_mode, initialize_deadline, } => { - let has_configured_headers = http_headers + let has_configured_headers = matches!( + bearer_token, + Some(StreamableHttpBearerToken::ProvidedByHttpClient) + ) || http_headers .as_ref() .is_some_and(|headers| !headers.is_empty()) || env_http_headers @@ -1133,8 +1145,8 @@ impl RmcpClient { } else { let mut http_config = StreamableHttpClientTransportConfig::with_uri(url.clone()); - if let Some(bearer_token) = bearer_token.clone() { - http_config = http_config.auth_header(bearer_token); + if let Some(StreamableHttpBearerToken::Resolved(bearer_token)) = bearer_token { + http_config = http_config.auth_header(bearer_token.clone()); } let transport = StreamableHttpClientTransport::with_client( diff --git a/codex-rs/rmcp-client/tests/streamable_http_oauth_store_pinning.rs b/codex-rs/rmcp-client/tests/streamable_http_oauth_store_pinning.rs index d3f10d5d09..3846fe4f81 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_oauth_store_pinning.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_oauth_store_pinning.rs @@ -12,11 +12,15 @@ use codex_config::types::OAuthCredentialsStoreMode; use codex_exec_server::Environment; use codex_exec_server::ExecServerError; use codex_exec_server::HttpClient; +use codex_exec_server::HttpRedirectPolicy; use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpRequestResponse; use codex_exec_server::HttpResponseBodyStream; +use codex_rmcp_client::McpProtocolMode; use codex_rmcp_client::RmcpClient; use codex_rmcp_client::StoredOAuthTokens; +use codex_rmcp_client::StreamableHttpBearerToken; +use codex_rmcp_client::StreamableHttpRedirectMode; use codex_rmcp_client::WrappedOAuthTokenResponse; use codex_rmcp_client::save_oauth_tokens; use codex_rmcp_client::stored_oauth_credential_snapshot; @@ -48,6 +52,7 @@ const FILE_ACCESS_TOKEN: &str = "stale-file-access-token"; struct RecordingHttpClient { inner: Arc, bearer_tokens: Arc>>, + redirect_policies: Arc>>, } impl RecordingHttpClient { @@ -55,10 +60,15 @@ impl RecordingHttpClient { Self { inner, bearer_tokens: Arc::new(Mutex::new(Vec::new())), + redirect_policies: Arc::new(Mutex::new(Vec::new())), } } - fn record_bearer_token(&self, params: &HttpRequestParams) { + fn record_request(&self, params: &HttpRequestParams) { + self.redirect_policies + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(params.redirect_policy); let Some(header) = params .headers .iter() @@ -85,7 +95,7 @@ impl HttpClient for RecordingHttpClient { &self, params: HttpRequestParams, ) -> BoxFuture<'_, Result> { - self.record_bearer_token(¶ms); + self.record_request(¶ms); self.inner.http_request(params) } @@ -93,7 +103,7 @@ impl HttpClient for RecordingHttpClient { &self, params: HttpRequestParams, ) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> { - self.record_bearer_token(¶ms); + self.record_request(¶ms); self.inner.http_request_stream(params) } } @@ -176,6 +186,47 @@ impl CredentialBuilderApi for TestCredentialBuilder { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn transport_provided_bearer_token_avoids_placeholder_headers_and_redirects() +-> anyhow::Result<()> { + let (_server, base_url) = spawn_streamable_http_server().await?; + let http_client = RecordingHttpClient::new(Environment::default_for_tests().get_http_client()); + let client = RmcpClient::new_streamable_http_client_with_protocol_mode_and_redirect_mode( + "transport-provided-bearer-token", + &format!("{base_url}/mcp"), + Some(StreamableHttpBearerToken::ProvidedByHttpClient), + /*http_headers*/ None, + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + AuthKeyringBackendKind::Direct, + Arc::new(http_client.clone()), + /*auth_provider*/ None, + McpProtocolMode::Legacy, + StreamableHttpRedirectMode::AgentPluginV1, + ) + .await?; + + initialize_client(&client).await?; + assert_eq!( + call_echo_tool(&client, "transport-provided").await?, + expected_echo_result("transport-provided") + ); + assert_eq!(http_client.bearer_tokens(), Vec::::new()); + + let redirect_policies = http_client + .redirect_policies + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone(); + assert!(!redirect_policies.is_empty()); + assert_eq!( + redirect_policies, + vec![HttpRedirectPolicy::Stop; redirect_policies.len()] + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn auto_store_remains_pinned_across_session_recovery() -> anyhow::Result<()> { let (_server, base_url) = spawn_streamable_http_server().await?;