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

Add dynamic HTTP header helpers for MCP servers (#38245)

Browse Source 
## What changed

- Add `http_headers_helper` configuration for local streamable HTTP MCP servers. The configured shell command runs once per connection and returns a JSON object of headers that is cached across requests.
- Apply helper headers to MCP startup and OAuth flows while restricting them to the server origin, stopping redirects, rejecting reserved or duplicate headers, and enforcing output and execution limits.
- Reject helpers for remote or managed-disabled servers, use the local environment working directory, and redact helper commands from `codex mcp list` and `codex mcp get` output.

## Testing

- Cover configuration validation, helper lifecycle and output parsing, origin isolation, OAuth discovery and token refresh, managed requirements, environment selection, and CLI redaction.

GitOrigin-RevId: 84e0e26ce75520b0869d37c72b1678e033bd6818
This commit is contained in:
xl-openai
2026-08-12 20:36:32 +00:00
committed by copyberry
parent 8d4d57387a
commit 379cb68444
38 changed files with 1285 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/mcp_server_status.rs
+136
View File
@@ -18,6 +18,7 @@ use axum::http::HeaderMap;
use axum::routing::get;
use axum::routing::post;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::ListMcpServerStatusParams;
use codex_app_server_protocol::ListMcpServerStatusResponse;
use codex_app_server_protocol::McpServerOauthLoginCompletedNotification;
@@ -52,9 +53,144 @@ use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::sleep;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
#[tokio::test]
async fn oauth_login_uses_http_headers_helper() -> Result<()> {
let oauth = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server/mcp"))
.and(header("x-gateway", "gateway-token"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"issuer": format!("{}/mcp", oauth.uri()),
"authorization_endpoint": format!("{}/oauth/authorize", oauth.uri()),
"token_endpoint": format!("{}/oauth/token", oauth.uri()),
})))
.mount(&oauth)
.await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.and(header("x-gateway", "gateway-token"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"access_token": "oauth-token",
"token_type": "Bearer",
})))
.expect(1)
.mount(&oauth)
.await;
let codex_home = TempDir::new()?;
let helper_command = if cfg!(windows) {
r#"echo {"X-Gateway":"gateway-token"}"#
} else {
r#"printf '{"X-Gateway":"gateway-token"}'"#
};
std::fs::write(
codex_home.path().join("config.toml"),
format!(
"mcp_oauth_credentials_store = \"file\"\n\
[mcp_servers.gateway]\n\
url = \"{}/mcp\"\n\
http_headers_helper = {}\n\
[mcp_servers.gateway.oauth]\n\
client_id = \"test-client\"\n",
oauth.uri(),
toml::Value::String(helper_command.to_string()),
),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
.await?;
let request_id = mcp
.send_raw_request(
"mcpServer/oauth/login",
Some(json!({"name": "gateway", "timeoutSecs": 10})),
)
.await?;
let response: McpServerOauthLoginResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??;
let authorization_url = reqwest::Url::parse(&response.authorization_url)?;
let query: BTreeMap<_, _> = authorization_url.query_pairs().into_owned().collect();
let mut callback_url = reqwest::Url::parse(&query["redirect_uri"])?;
callback_url
.query_pairs_mut()
.append_pair("code", "test-code")
.append_pair("state", &query["state"]);
reqwest::Client::builder()
.no_proxy()
.build()?
.get(callback_url)
.send()
.await?
.error_for_status()?;
let completed: McpServerOauthLoginCompletedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("mcpServer/oauthLogin/completed"),
)
.await??;
assert_eq!(
completed,
McpServerOauthLoginCompletedNotification {
name: "gateway".to_string(),
thread_id: None,
success: true,
error: None,
}
);
oauth.verify().await;
Ok(())
}
#[tokio::test]
async fn oauth_login_does_not_run_helper_disabled_by_managed_requirements() -> Result<()> {
let codex_home = TempDir::new()?;
let marker = codex_home.path().join("helper-ran");
let helper = toml::Value::String(format!("echo invoked > \"{}\"", marker.display()));
std::fs::write(
codex_home.path().join("config.toml"),
format!(
"[mcp_servers.blocked]\nurl = \"https://example.com/mcp\"\nhttp_headers_helper = {helper}\n"
),
)?;
std::fs::write(
codex_home.path().join("requirements.toml"),
"[mcp_servers.blocked.identity]\nurl = \"https://allowed.example.com/mcp\"\n",
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized_with_timeout(DEFAULT_READ_TIMEOUT)
.await?;
let request_id = mcp
.send_raw_request(
"mcpServer/oauth/login",
Some(json!({"name": "blocked", "timeoutSecs": 10})),
)
.await?;
let error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert!(
error
.error
.message
.contains("disabled by managed requirements")
);
assert!(!marker.exists());
Ok(())
}
async fn wait_for_new_pid(path: &Path, previous_pid: Option<&str>) -> Result<String> {
Ok(timeout(DEFAULT_READ_TIMEOUT, async {
loop {
codex-rs/cli/src/mcp_cmd.rs
+15
View File
@@ -25,6 +25,7 @@ use codex_login::AuthManager;
use codex_mcp::McpOAuthLoginSupport;
use codex_mcp::McpRuntimeContext;
use codex_mcp::ResolvedMcpOAuthScopes;
use codex_mcp::apply_http_headers_helper;
use codex_mcp::compute_auth_statuses;
use codex_mcp::discover_supported_scopes;
use codex_mcp::oauth_login_support;
@@ -398,6 +399,7 @@ async fn run_add(config_overrides: &CliConfigOverrides, add_args: AddArgs) -> Re
bearer_token_env_var,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
oauth_client_id,
oauth_client_registration
@@ -558,6 +560,8 @@ async fn run_login(config: &Config, login_args: LoginArgs) -> Result<()> {
// environment routing belongs to app-server and session MCP flows.
let http_client: Arc<dyn HttpClient> =
Arc::new(RouteAwareHttpClient::new(config.http_client_factory()));
let http_client = apply_http_headers_helper(http_client, server, config.cwd.to_path_buf())
.map_err(anyhow::Error::msg)?;
let explicit_scopes = (!scopes.is_empty()).then_some(scopes);
let discovered_scopes = if explicit_scopes.is_none() && server.scopes.is_none() {
discover_supported_scopes(
@@ -679,6 +683,7 @@ async fn run_list(config: &Config, list_args: ListArgs) -> Result<()> {
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} => {
serde_json::json!({
"type": "streamable_http",
@@ -686,6 +691,9 @@ async fn run_list(config: &Config, list_args: ListArgs) -> Result<()> {
"bearer_token_env_var": bearer_token_env_var,
"http_headers": http_headers,
"env_http_headers": env_http_headers,
"http_headers_helper": http_headers_helper
.as_ref()
.map(|_| "<redacted>"),
})
}
};
@@ -914,12 +922,16 @@ async fn run_get(config: &Config, get_args: GetArgs) -> Result<()> {
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} => serde_json::json!({
"type": "streamable_http",
"url": url,
"bearer_token_env_var": bearer_token_env_var,
"http_headers": http_headers,
"env_http_headers": env_http_headers,
"http_headers_helper": http_headers_helper
.as_ref()
.map(|_| "<redacted>"),
}),
};
let output = serde_json::to_string_pretty(&serde_json::json!({
@@ -996,6 +1008,7 @@ async fn run_get(config: &Config, get_args: GetArgs) -> Result<()> {
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} => {
println!(" transport: streamable_http");
println!(" url: {url}");
@@ -1027,6 +1040,8 @@ async fn run_get(config: &Config, get_args: GetArgs) -> Result<()> {
_ => "-".to_string(),
};
println!(" env_http_headers: {env_headers_display}");
let helper_display = http_headers_helper.as_ref().map_or("-", |_| "<redacted>");
println!(" http_headers_helper: {helper_display}");
}
}
if let Some(timeout) = server.startup_timeout_sec {
codex-rs/cli/tests/mcp_add_remove.rs
+26 -4
View File
@@ -137,6 +137,18 @@ async fn add_and_login_discover_oauth_through_configured_http_proxy() -> Result<
.await?
.contains_key("oauth")
);
let helper_command = if cfg!(windows) {
r#"echo {"X-Gateway":"gateway-token"}"#
} else {
r#"printf '{"X-Gateway":"gateway-token"}'"#
};
let config_path = codex_home.path().join("config.toml");
let mut config = std::fs::read_to_string(&config_path)?;
config.push_str(&format!(
"http_headers_helper = {}\n",
toml::Value::String(helper_command.to_string())
));
std::fs::write(config_path, config)?;
// Local OAuth login does not require the execution-environment registry.
std::fs::write(codex_home.path().join("environments.toml"), "invalid = [")?;
@@ -164,14 +176,22 @@ async fn add_and_login_discover_oauth_through_configured_http_proxy() -> Result<
"mock OAuth registration should terminate the explicit login"
);
let registrations = proxy
let requests = proxy
.received_requests()
.await
.expect("mock proxy should record OAuth requests")
.expect("mock proxy should record OAuth requests");
let registrations: Vec<_> = requests
.iter()
.filter(|request| request.method == "POST" && request.url.path() == "/oauth/register")
.count();
assert_eq!(registrations, 2);
.collect();
assert_eq!(registrations.len(), 2);
assert_eq!(
registrations
.iter()
.filter(|request| request.headers.get("x-gateway").is_some())
.count(),
1
);
Ok(())
}
@@ -259,6 +279,7 @@ async fn add_streamable_http_without_manual_token() -> Result<()> {
bearer_token_env_var,
http_headers,
env_http_headers,
..
} => {
assert_eq!(url, "https://example.com/mcp");
assert!(bearer_token_env_var.is_none());
@@ -305,6 +326,7 @@ async fn add_streamable_http_with_custom_env_var() -> Result<()> {
bearer_token_env_var,
http_headers,
env_http_headers,
..
} => {
assert_eq!(url, "https://example.com/issues");
assert_eq!(bearer_token_env_var.as_deref(), Some("GITHUB_TOKEN"));
codex-rs/cli/tests/mcp_list.rs
+64
View File
@@ -1,3 +1,4 @@
use std::collections::BTreeMap;
use std::io::Read;
use std::io::Write;
use std::net::TcpListener;
@@ -419,6 +420,69 @@ async fn list_and_get_render_expected_output() -> Result<()> {
Ok(())
}
#[test]
fn list_and_get_redact_http_headers_helper() -> Result<()> {
let codex_home = TempDir::new()?;
let marker = codex_home.path().join("helper-ran");
let helper = toml::Value::String(format!("echo invoked > \"{}\"", marker.display()));
std::fs::write(
codex_home.path().join("config.toml"),
format!(
"[mcp_servers.docs]\n\
url = \"https://example.com/mcp\"\n\
http_headers_helper = {helper}\n\
[mcp_servers.authenticated]\n\
url = \"https://example.com/mcp\"\n\
http_headers = {{ Authorization = \"Bearer static\" }}\n\
http_headers_helper = {helper}\n"
),
)?;
let list_output = codex_command(codex_home.path())?
.args(["mcp", "list", "--json"])
.output()?;
assert!(list_output.status.success());
let stdout = String::from_utf8(list_output.stdout)?;
assert!(stdout.contains("http_headers_helper"));
assert!(stdout.contains("<redacted>"));
let entries: Vec<JsonValue> = serde_json::from_str(&stdout)?;
let auth_statuses = entries
.into_iter()
.map(|entry| {
(
entry["name"].as_str().expect("server name").to_string(),
entry["auth_status"]
.as_str()
.expect("auth status")
.to_string(),
)
})
.collect::<BTreeMap<_, _>>();
assert_eq!(
auth_statuses,
BTreeMap::from([
("authenticated".to_string(), "bearer_token".to_string()),
("docs".to_string(), "unknown".to_string()),
])
);
assert!(!marker.exists());
for args in [
&["mcp", "get", "docs", "--json"][..],
&["mcp", "get", "docs"][..],
] {
let output = codex_command(codex_home.path())?.args(args).output()?;
assert!(output.status.success());
let stdout = String::from_utf8(output.stdout)?;
assert!(stdout.contains("http_headers_helper"));
assert!(stdout.contains("<redacted>"));
assert!(!stdout.contains("helper-ran"));
assert!(!marker.exists());
}
Ok(())
}
#[tokio::test]
async fn get_disabled_server_shows_single_line() -> Result<()> {
let codex_home = TempDir::new()?;
codex-rs/codex-mcp/src/catalog_tests.rs
+1
View File
@@ -23,6 +23,7 @@ fn server(url: &str) -> McpServerConfig {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/codex-mcp/src/connection_manager.rs
+1
View File
@@ -564,6 +564,7 @@ impl McpConnectionSet {
bearer_token_env_var,
http_headers,
env_http_headers,
..
} => {
match determine_streamable_http_auth_status_from_credentials(
configured_config
codex-rs/codex-mcp/src/connection_manager_tests.rs
+42
View File
@@ -2069,6 +2069,43 @@ fn tool_catalog_cache_bypasses_remote_sourced_environment_variables() {
);
}
#[test]
fn tool_catalog_cache_bypasses_http_headers_helpers() {
let cache = McpToolCatalogCache::default();
let runtime_context = reusable_server_runtime_context();
let mut config = reusable_server_config("https://example.com/mcp");
let identity = reusable_server_identity(&config, &runtime_context);
let context = |config: &McpServerConfig, identity: &McpServerConnectionIdentity| {
cache.context(
"docs",
config,
&runtime_context,
/*resolved_environment*/ None,
(
&ElicitationCapability::default(),
&ClientMcpExtensions::default(),
),
Some((
identity,
crate::McpProtocolMode::Legacy,
/*agent_plugin*/ false,
)),
)
};
assert!(context(&config, &identity).is_some());
let McpServerTransportConfig::StreamableHttp {
http_headers_helper,
..
} = &mut config.transport
else {
unreachable!("expected HTTP transport");
};
*http_headers_helper = Some("auth-cli headers".to_string());
let identity = reusable_server_identity(&config, &runtime_context);
assert!(context(&config, &identity).is_none());
}
#[tokio::test]
async fn list_available_server_infos_uses_cache_while_client_is_pending() {
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
@@ -3532,6 +3569,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -3642,6 +3680,7 @@ fn mcp_init_error_display_prompts_for_github_pat() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -3757,6 +3796,7 @@ fn mcp_init_error_display_reports_generic_errors() {
bearer_token_env_var: Some("TOKEN".to_string()),
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -3809,6 +3849,7 @@ fn reusable_server_config(url: &str) -> McpServerConfig {
bearer_token_env_var: Some("CODEX_MCP_REUSE_TEST_TOKEN".to_string()),
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -4119,6 +4160,7 @@ fn connection_identity_uses_effective_authorization_headers() {
.map(|value| HashMap::from([("aUtHoRiZaTiOn".to_string(), value.to_string())])),
env_http_headers: environment_header
.map(|value| HashMap::from([("aUtHoRiZaTiOn".to_string(), value.to_string())])),
http_headers_helper: None,
};
let server = EffectiveMcpServer::configured(config);
let identity = |keyring_backend_kind| {
codex-rs/codex-mcp/src/lib.rs
+1
View File
@@ -21,6 +21,7 @@ pub use runtime::McpRuntimeContext;
pub use runtime::McpRuntimeInput;
pub use runtime::McpStartupPolicy;
pub use runtime::SandboxState;
pub use runtime::apply_http_headers_helper;
pub use tool_catalog_cache::McpToolCatalogCache;
pub use tools::ToolInfo;
codex-rs/codex-mcp/src/mcp/auth.rs
+17
View File
@@ -14,6 +14,7 @@ use codex_rmcp_client::OAuthDiscoveryTimeout;
use codex_rmcp_client::OAuthProviderError;
use codex_rmcp_client::StreamableHttpRedirectMode;
use codex_rmcp_client::determine_streamable_http_auth_status;
use codex_rmcp_client::determine_streamable_http_auth_status_from_credentials;
use codex_rmcp_client::discover_streamable_http_oauth;
use futures::FutureExt;
use futures::future::join_all;
@@ -92,6 +93,7 @@ fn oauth_login_candidate(transport: &McpServerTransportConfig) -> Option<McpOAut
bearer_token_env_var,
http_headers,
env_http_headers,
..
} = transport
else {
return None;
@@ -247,7 +249,22 @@ async fn compute_auth_status(
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} => {
if http_headers_helper.is_some() {
// Status inspection must not execute an arbitrary local helper. Existing
// credentials remain reportable; otherwise discovery waits for startup/login.
return Ok(determine_streamable_http_auth_status_from_credentials(
config.oauth_credential_name(server_name).as_ref(),
url,
bearer_token_env_var.as_deref(),
http_headers.clone(),
env_http_headers.clone(),
store_mode,
keyring_backend_kind,
)?
.unwrap_or(McpAuthState::Unknown));
}
let http_client = runtime_context
.resolve_http_client(server_name, config)
.map_err(anyhow::Error::msg)?;
codex-rs/codex-mcp/src/mcp/mod.rs
+1
View File
@@ -565,6 +565,7 @@ fn mcp_server_config_for_url(
bearer_token_env_var: codex_apps_mcp_bearer_token_env_var(),
http_headers: Some(http_headers),
env_http_headers,
http_headers_helper: None,
},
auth: auth_mode,
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
codex-rs/codex-mcp/src/mcp/mod_tests.rs
+2
View File
@@ -401,6 +401,7 @@ async fn effective_mcp_servers_preserve_runtime_servers() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -428,6 +429,7 @@ async fn effective_mcp_servers_preserve_runtime_servers() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/codex-mcp/src/plugin_config.rs
+3
View File
@@ -64,6 +64,9 @@ impl PluginMcpFile {
/// Parses the two supported plugin MCP file shapes and normalizes each server.
///
/// Native plugin HTTP servers share the regular MCP transport configuration;
/// relative helper commands therefore use the session's local process cwd.
///
/// Invalid individual servers are returned as errors without discarding valid
/// siblings. A malformed top-level document fails the whole parse.
pub fn parse_plugin_mcp_config(
codex-rs/codex-mcp/src/plugin_config_tests.rs
+17 -1
View File
@@ -566,6 +566,7 @@ fn declared_placement_preserves_local_plugin_normalization() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -585,6 +586,18 @@ fn declared_placement_preserves_local_plugin_normalization() {
oauth_resource: None,
tools: HashMap::new(),
};
let mut expected_helper = McpServerConfig {
oauth: None,
..expected_http.clone()
};
let McpServerTransportConfig::StreamableHttp {
http_headers_helper,
..
} = &mut expected_helper.transport
else {
unreachable!("expected HTTP transport");
};
*http_headers_helper = Some("./auth.sh".to_string());
let outcome = parse_plugin_mcp_config(
&plugin_root,
@@ -598,7 +611,8 @@ fn declared_placement_preserves_local_plugin_normalization() {
"type": "http",
"url": "https://example.com/mcp",
"oauth": {"clientId": "client-id", "callbackPort": 9876}
}
},
"helper": {"type":"http","url":"https://example.com/mcp","http_headers_helper":"./auth.sh"}
}"#,
)
.expect("parse plugin MCP config");
@@ -608,6 +622,7 @@ fn declared_placement_preserves_local_plugin_normalization() {
PluginMcpConfigParseOutcome {
servers: BTreeMap::from([
("demo".to_string(), expected_stdio),
("helper".to_string(), expected_helper),
("hosted".to_string(), expected_http),
]),
errors: Vec::new(),
@@ -839,6 +854,7 @@ fn local_environment_placement_preserves_http_env_references() {
"X-Account".to_string(),
"ACCOUNT_ID".to_string(),
)])),
http_headers_helper: None,
},
environment_id: DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/codex-mcp/src/rmcp_client.rs
+5 -5
View File
@@ -1061,7 +1061,7 @@ async fn make_rmcp_client(
// `ExecutorStdioServerLauncher` once the executor-backed path
// preserves `LocalStdioServerLauncher` semantics.
Arc::new(LocalStdioServerLauncher::new(
runtime_context.local_stdio_fallback_cwd(),
runtime_context.local_process_cwd(),
)) as Arc<dyn StdioServerLauncher>
} else {
let Some(environment) = resolved_environment.as_ref() else {
@@ -1092,11 +1092,11 @@ async fn make_rmcp_client(
http_headers,
env_http_headers,
bearer_token_env_var,
http_headers_helper: _,
} => {
let http_client = resolved_environment.as_ref().map_or_else(
|| runtime_context.local_http_client(),
|environment| environment.get_http_client(),
);
let http_client = runtime_context
.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()) {
codex-rs/codex-mcp/src/runtime.rs
+73 -17
View File
@@ -15,6 +15,7 @@ use std::time::Duration;
use arc_swap::ArcSwap;
use async_channel::Sender;
use codex_config::types::McpServerDisabledReason;
use codex_connectors::ConnectorRuntimeContextKey;
use codex_connectors::ConnectorRuntimeManager;
use codex_exec_server::Environment;
@@ -29,6 +30,7 @@ use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::Event;
use codex_rmcp_client::ElicitationResponse;
use codex_rmcp_client::with_http_headers_helper;
use codex_utils_path_uri::PathUri;
use rmcp::model::ReadResourceRequestParams;
use rmcp::model::ReadResourceResult;
@@ -453,30 +455,57 @@ pub struct SandboxState {
/// Runtime context used when resolving per-server MCP environments.
///
/// `McpConfig` describes what servers exist. This value carries the canonical
/// environment registry plus the local stdio fallback cwd used when a local
/// stdio server omits its own working directory.
/// environment registry plus the host-local cwd used by local MCP processes.
#[derive(Clone)]
pub struct McpRuntimeContext {
environment_manager: Arc<EnvironmentManager>,
local_stdio_fallback_cwd: PathBuf,
local_process_cwd: PathBuf,
}
/// Applies the local HTTP headers helper configured for an MCP server.
///
/// Callers retain ownership of selecting the underlying HTTP transport. This
/// function centralizes the helper-specific policy checks and decoration used
/// by both MCP runtime startup and standalone OAuth login.
pub fn apply_http_headers_helper(
client: Arc<dyn HttpClient>,
config: &codex_config::McpServerConfig,
local_process_cwd: PathBuf,
) -> Result<Arc<dyn HttpClient>, String> {
let codex_config::McpServerTransportConfig::StreamableHttp {
url,
http_headers_helper: Some(command),
..
} = &config.transport
else {
return Ok(client);
};
if matches!(
config.disabled_reason,
Some(McpServerDisabledReason::Requirements { .. })
) {
return Err("the MCP server is disabled by managed requirements".to_string());
}
if !config.is_local_environment() {
return Err("HTTP headers helpers can only run in the local environment".to_string());
}
with_http_headers_helper(client, url, command, local_process_cwd)
.map_err(|error| error.to_string())
}
impl McpRuntimeContext {
pub fn new(
environment_manager: Arc<EnvironmentManager>,
local_stdio_fallback_cwd: PathBuf,
) -> Self {
pub fn new(environment_manager: Arc<EnvironmentManager>, local_process_cwd: PathBuf) -> Self {
Self {
environment_manager,
local_stdio_fallback_cwd,
local_process_cwd,
}
}
pub(crate) fn local_stdio_fallback_cwd(&self) -> PathBuf {
self.local_stdio_fallback_cwd.clone()
pub(crate) fn local_process_cwd(&self) -> PathBuf {
self.local_process_cwd.clone()
}
pub(crate) fn local_http_client(&self) -> Arc<dyn HttpClient> {
fn local_http_client(&self) -> Arc<dyn HttpClient> {
Arc::new(RouteAwareHttpClient::new(
self.environment_manager.http_client_factory().clone(),
))
@@ -518,12 +547,20 @@ impl McpRuntimeContext {
server_name: &str,
config: &codex_config::McpServerConfig,
) -> Result<Arc<dyn HttpClient>, String> {
Ok(self
.resolve_server_environment(server_name, config)?
.map_or_else(
|| self.local_http_client(),
|environment| environment.get_http_client(),
))
let environment = self.resolve_server_environment(server_name, config)?;
self.http_client_for_server(config, environment.as_ref())
}
pub(crate) fn http_client_for_server(
&self,
config: &codex_config::McpServerConfig,
environment: Option<&Arc<Environment>>,
) -> Result<Arc<dyn HttpClient>, String> {
let client = environment.map_or_else(
|| self.local_http_client(),
|environment| environment.get_http_client(),
);
apply_http_headers_helper(client, config, self.local_process_cwd())
}
}
@@ -638,6 +675,7 @@ mod tests {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: environment_id.to_string(),
..stdio_server(environment_id)
@@ -781,6 +819,24 @@ mod tests {
};
assert!(resolved_runtime.is_some());
}
let mut remote_http_with_helper = http_server("remote");
let McpServerTransportConfig::StreamableHttp {
http_headers_helper,
..
} = &mut remote_http_with_helper.transport
else {
unreachable!("HTTP helper should build streamable HTTP transport");
};
*http_headers_helper = Some("helper-that-must-not-run".to_string());
let error = match runtime_context.resolve_http_client("http", &remote_http_with_helper) {
Ok(_) => panic!("remote HTTP helper should be rejected"),
Err(error) => error,
};
assert_eq!(
error,
"HTTP headers helpers can only run in the local environment"
);
}
#[tokio::test]
codex-rs/codex-mcp/src/server.rs
+6 -1
View File
@@ -140,6 +140,7 @@ impl McpServerConnectionIdentity {
bearer_token_env_var: None,
http_headers,
env_http_headers,
http_headers_helper: _,
} if !http_headers.as_ref().is_some_and(|headers| {
headers.iter().any(|(name, value)| {
name.eq_ignore_ascii_case("authorization") && valid_http_header_value(value)
@@ -189,8 +190,12 @@ impl McpServerConnectionIdentity {
&& matches!(
config.transport,
McpServerTransportConfig::Stdio { cwd: None, .. }
| McpServerTransportConfig::StreamableHttp {
http_headers_helper: Some(_),
..
}
))
.then(|| runtime_context.local_stdio_fallback_cwd());
.then(|| runtime_context.local_process_cwd());
let referenced_environment_variables = referenced_environment_variables(config);
let runtime_auth = runtime_auth_provider.and(auth).cloned();
let runtime_auth_token = runtime_auth.as_ref().and_then(|auth| auth.get_token().ok());
codex-rs/codex-mcp/src/tool_catalog_cache.rs
+6 -1
View File
@@ -235,7 +235,7 @@ impl ToolCatalogIdentity {
&config.transport,
McpServerTransportConfig::Stdio { cwd: None, .. }
)
.then(|| runtime_context.local_stdio_fallback_cwd()),
.then(|| runtime_context.local_process_cwd()),
})
}
}
@@ -258,8 +258,13 @@ impl ToolCatalogTransportIdentity {
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} = &config.transport
{
// Helper output is a dynamic credential identity that cannot be represented by config.
if http_headers_helper.is_some() {
return None;
}
let (connection_identity, protocol_mode, agent_plugin) = connection_identity?;
if config.oauth.is_some()
|| config.scopes.is_some()
codex-rs/config/src/mcp_requirements.rs
+1
View File
@@ -122,6 +122,7 @@ impl McpServerRequirement {
}
pub fn matches(&self, server: &McpServerConfig) -> bool {
// HTTP requirements intentionally authorize the complete server configuration by URL.
match (self, &server.transport) {
(
Self::Identity {
codex-rs/config/src/mcp_types.rs
+23
View File
@@ -305,6 +305,7 @@ pub struct RawMcpServerConfig {
#[schemars(skip)]
pub bearer_token: Option<String>,
pub bearer_token_env_var: Option<String>,
pub http_headers_helper: Option<String>,
// shared
#[serde(default)]
@@ -360,6 +361,7 @@ impl TryFrom<RawMcpServerConfig> for McpServerConfig {
url,
bearer_token,
bearer_token_env_var,
http_headers_helper,
environment_id,
auth,
startup_timeout_sec,
@@ -402,6 +404,7 @@ impl TryFrom<RawMcpServerConfig> for McpServerConfig {
bearer_token_env_var.as_ref(),
)?;
throw_if_set("stdio", "bearer_token", bearer_token.as_ref())?;
throw_if_set("stdio", "http_headers_helper", http_headers_helper.as_ref())?;
throw_if_set("stdio", "http_headers", http_headers.as_ref())?;
throw_if_set("stdio", "env_http_headers", env_http_headers.as_ref())?;
throw_if_set("stdio", "oauth", oauth.as_ref())?;
@@ -424,11 +427,27 @@ impl TryFrom<RawMcpServerConfig> for McpServerConfig {
throw_if_set("streamable_http", "env_vars", env_vars.as_ref())?;
throw_if_set("streamable_http", "cwd", cwd.as_ref())?;
throw_if_set("streamable_http", "bearer_token", bearer_token.as_ref())?;
if http_headers_helper
.as_deref()
.is_some_and(|command| command.trim().is_empty())
{
return Err("http_headers_helper must not be empty".to_string());
}
if environment_id
.as_deref()
.is_some_and(|environment_id| environment_id != DEFAULT_MCP_SERVER_ENVIRONMENT_ID)
&& http_headers_helper.is_some()
{
return Err(
"http_headers_helper is only supported for local MCP servers".to_string(),
);
}
McpServerTransportConfig::StreamableHttp {
url,
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
}
} else {
return Err("invalid transport".to_string());
@@ -503,6 +522,10 @@ pub enum McpServerTransportConfig {
/// HTTP headers where the value is sourced from an environment variable.
#[serde(default, skip_serializing_if = "Option::is_none")]
env_http_headers: Option<HashMap<String, String>>,
/// Local-only shell command that prints a JSON object of dynamic HTTP headers.
/// The command may be visible to local process inspection; do not embed credentials.
#[serde(default, skip_serializing_if = "Option::is_none")]
http_headers_helper: Option<String>,
},
}
codex-rs/config/src/mcp_types_tests.rs
+15
View File
@@ -273,6 +273,7 @@ fn deserialize_streamable_http_server_config() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
}
);
assert!(cfg.enabled);
@@ -295,6 +296,7 @@ fn deserialize_streamable_http_server_config_with_env_var() {
bearer_token_env_var: Some("GITHUB_TOKEN".to_string()),
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
}
);
assert!(cfg.enabled);
@@ -307,6 +309,7 @@ fn deserialize_streamable_http_server_config_with_headers() {
url = "https://example.com/mcp"
http_headers = { "X-Foo" = "bar" }
env_http_headers = { "X-Token" = "TOKEN_ENV" }
http_headers_helper = "auth-cli headers"
"#,
)
.expect("should deserialize http config with headers");
@@ -321,10 +324,22 @@ fn deserialize_streamable_http_server_config_with_headers() {
"X-Token".to_string(),
"TOKEN_ENV".to_string()
)])),
http_headers_helper: Some("auth-cli headers".to_string()),
}
);
}
#[test]
fn rejects_http_headers_helper_outside_local_http_servers() {
for contents in [
"command = \"server\"\nhttp_headers_helper = \"auth-cli headers\"",
"url = \"https://example.com/mcp\"\nhttp_headers_helper = \" \"",
"url = \"https://example.com/mcp\"\nenvironment_id = \"remote\"\nhttp_headers_helper = \"auth-cli headers\"",
] {
toml::from_str::<McpServerConfig>(contents).expect_err("invalid helper placement");
}
}
#[test]
fn deserialize_streamable_http_server_config_with_oauth_resource() {
let cfg: McpServerConfig = toml::from_str(
codex-rs/core-plugins/src/manager_tests.rs
+5
View File
@@ -952,6 +952,7 @@ async fn load_plugins_loads_default_skills_and_mcp_servers() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "local".to_string(),
enabled: true,
@@ -1050,6 +1051,7 @@ enabled = true
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "local".to_string(),
enabled: true,
@@ -2146,6 +2148,7 @@ async fn load_plugins_uses_manifest_configured_component_paths() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "local".to_string(),
enabled: true,
@@ -2483,6 +2486,7 @@ async fn load_plugins_ignores_manifest_component_paths_without_dot_slash() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "local".to_string(),
enabled: true,
@@ -2737,6 +2741,7 @@ fn capability_index_filters_inactive_and_zero_capability_plugins() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "local".to_string(),
enabled: true,
codex-rs/core/config.schema.json
+3
View File
@@ -2721,6 +2721,9 @@
},
"type": "object"
},
"http_headers_helper": {
"type": "string"
},
"name": {
"default": null,
"description": "Legacy display-name field accepted for backward compatibility.",
codex-rs/core/src/config/config_tests.rs
+9
View File
@@ -159,6 +159,7 @@ fn http_mcp(url: &str) -> McpServerConfig {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -6882,6 +6883,7 @@ async fn replace_mcp_servers_streamable_http_serializes_bearer_token() -> anyhow
bearer_token_env_var: Some("MCP_TOKEN".to_string()),
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -6925,6 +6927,7 @@ startup_timeout_sec = 2.0
bearer_token_env_var,
http_headers,
env_http_headers,
..
} => {
assert_eq!(url, "https://example.com/mcp");
assert_eq!(bearer_token_env_var.as_deref(), Some("MCP_TOKEN"));
@@ -6954,6 +6957,7 @@ async fn replace_mcp_servers_streamable_http_serializes_custom_headers() -> anyh
"X-Auth".to_string(),
"DOCS_AUTH".to_string(),
)])),
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -7038,6 +7042,7 @@ async fn replace_mcp_servers_streamable_http_removes_optional_sections() -> anyh
"X-Auth".to_string(),
"DOCS_AUTH".to_string(),
)])),
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -7075,6 +7080,7 @@ async fn replace_mcp_servers_streamable_http_removes_optional_sections() -> anyh
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -7114,6 +7120,7 @@ url = "https://example.com/mcp"
bearer_token_env_var,
http_headers,
env_http_headers,
..
} => {
assert_eq!(url, "https://example.com/mcp");
assert!(bearer_token_env_var.is_none());
@@ -7147,6 +7154,7 @@ async fn replace_mcp_servers_streamable_http_isolates_headers_between_servers()
"X-Auth".to_string(),
"DOCS_AUTH".to_string(),
)])),
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
@@ -7425,6 +7433,7 @@ async fn replace_mcp_servers_streamable_http_serializes_oauth_resource() -> anyh
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/core/src/config/edit/document_helpers.rs
+4
View File
@@ -78,6 +78,7 @@ fn serialize_mcp_server_table(config: &McpServerConfig) -> TomlTable {
bearer_token_env_var,
http_headers,
env_http_headers,
http_headers_helper,
} => {
entry["url"] = value(url.clone());
if let Some(env_var) = bearer_token_env_var {
@@ -93,6 +94,9 @@ fn serialize_mcp_server_table(config: &McpServerConfig) -> TomlTable {
{
entry["env_http_headers"] = table_from_pairs(headers.iter());
}
if let Some(command) = http_headers_helper {
entry["http_headers_helper"] = value(command.clone());
}
}
}
codex-rs/core/src/config/edit_tests.rs
+2
View File
@@ -1040,6 +1040,7 @@ fn blocking_replace_mcp_servers_round_trips() {
.collect(),
),
env_http_headers: None,
http_headers_helper: Some("auth-cli headers".to_string()),
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: false,
@@ -1072,6 +1073,7 @@ fn blocking_replace_mcp_servers_round_trips() {
[mcp_servers.http]
url = \"https://example.com\"
bearer_token_env_var = \"TOKEN\"
http_headers_helper = \"auth-cli headers\"
enabled = false
startup_timeout_sec = 5.0
disabled_tools = [\"forbidden\"]
codex-rs/core/src/environment_selection.rs
+46 -4
View File
@@ -566,6 +566,26 @@ impl TurnEnvironmentSnapshot {
.find(|environment| !environment.environment.is_remote())
}
pub(crate) fn local_environment_cwd(&self) -> Option<AbsolutePathBuf> {
self.environments
.iter()
.find_map(|environment| match environment {
TurnEnvironmentState::Ready(environment)
if !environment.environment.is_remote() =>
{
environment.cwd().to_abs_path().ok()
}
TurnEnvironmentState::Ready(_) => None,
TurnEnvironmentState::Starting(environment)
if environment.selection.environment_id
== codex_exec_server::LOCAL_ENVIRONMENT_ID =>
{
environment.selection.cwd.to_abs_path().ok()
}
TurnEnvironmentState::Starting(_) => None,
})
}
#[cfg(test)]
pub(crate) fn primary_environment(&self) -> Option<Arc<codex_exec_server::Environment>> {
self.primary()
@@ -1425,6 +1445,7 @@ url = "ws://127.0.0.1:8765"
async fn single_local_environment_cwd_requires_exactly_one_local_environment() {
let cwd = AbsolutePathBuf::current_dir().expect("cwd");
let cwd_uri = PathUri::from_abs_path(&cwd);
let remote_cwd_uri = PathUri::from_abs_path(&cwd.join("remote-cwd"));
let local_manager = Arc::new(EnvironmentManager::default_for_tests());
let local = resolve_turn_environments(
Arc::clone(&local_manager),
@@ -1444,7 +1465,7 @@ url = "ws://127.0.0.1:8765"
environments: vec![TurnEnvironmentState::Ready(TurnEnvironment::new(
REMOTE_ENVIRONMENT_ID.to_string(),
remote_environment.clone(),
cwd_uri.clone(),
remote_cwd_uri.clone(),
Vec::new(),
/*shell*/ None,
test_environment_config(),
@@ -1452,20 +1473,41 @@ url = "ws://127.0.0.1:8765"
};
let multiple = TurnEnvironmentSnapshot {
environments: vec![
TurnEnvironmentState::Ready(local.primary().expect("local environment").clone()),
TurnEnvironmentState::Ready(TurnEnvironment::new(
REMOTE_ENVIRONMENT_ID.to_string(),
remote_environment,
cwd_uri,
remote_cwd_uri,
Vec::new(),
/*shell*/ None,
test_environment_config(),
)),
TurnEnvironmentState::Ready(local.primary().expect("local environment").clone()),
],
};
assert_eq!(local.single_local_environment_cwd(), Some(cwd));
assert_eq!(local.single_local_environment_cwd(), Some(cwd.clone()));
assert_eq!(remote.single_local_environment_cwd(), None);
assert_eq!(multiple.single_local_environment_cwd(), None);
assert_eq!(multiple.local_environment_cwd(), Some(cwd));
}
#[test]
fn local_environment_cwd_uses_starting_local_selection() {
let cwd = AbsolutePathBuf::current_dir()
.expect("cwd")
.join("starting-local");
let snapshot = TurnEnvironmentSnapshot {
environments: vec![TurnEnvironmentState::Starting(StartingTurnEnvironment {
selection: TurnEnvironmentSelection {
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
cwd: PathUri::from_abs_path(&cwd),
workspace_roots: Vec::new(),
},
config: test_environment_config(),
resolution: futures::future::pending().boxed().shared(),
})],
};
assert_eq!(snapshot.local_environment_cwd(), Some(cwd));
}
}
codex-rs/core/src/mcp_skill_dependencies.rs
+1
View File
@@ -391,6 +391,7 @@ fn mcp_dependency_to_server_config(
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/core/src/session/mcp.rs
+6 -6
View File
@@ -96,11 +96,12 @@ impl Session {
config: &Config,
) -> (McpConfig, McpRuntimeContext) {
let originator = self.originator().await;
let (windows_sandbox_level, session_source) = {
let (windows_sandbox_level, session_source, host_fallback_cwd) = {
let state = self.state.lock().await;
(
state.session_configuration.windows_sandbox_level,
state.session_configuration.session_source.clone(),
state.session_configuration.cwd().clone(),
)
};
let environments = self.services.turn_environments.snapshot().await;
@@ -133,14 +134,13 @@ impl Session {
)
.await
.config;
let local_stdio_fallback_cwd = environments
.primary()
.and_then(|environment| environment.cwd().to_abs_path().ok())
let local_process_cwd = environments
.local_environment_cwd()
.map(|cwd| cwd.to_path_buf())
.unwrap_or_else(|| config.cwd.to_path_buf());
.unwrap_or_else(|| host_fallback_cwd.to_path_buf());
let runtime_context = McpRuntimeContext::new(
self.services.turn_environments.environment_manager(),
local_stdio_fallback_cwd,
local_process_cwd,
);
(mcp_config, runtime_context)
}
codex-rs/core/src/session/mcp_runtime.rs
+14 -14
View File
@@ -19,19 +19,10 @@ pub(super) struct McpDesiredState {
pub(super) originator: String,
pub(super) session_source: SessionSource,
pub(super) environments: TurnEnvironmentSnapshot,
pub(super) local_process_cwd: PathBuf,
pub(super) windows_sandbox_level: WindowsSandboxLevel,
}
impl McpDesiredState {
pub(super) fn local_stdio_fallback_cwd(&self) -> PathBuf {
self.environments
.primary()
.and_then(|environment| environment.cwd().to_abs_path().ok())
.map(|cwd| cwd.to_path_buf())
.unwrap_or_else(|| self.config.cwd.to_path_buf())
}
}
impl Session {
/// Waits on this session's refreshed server before tool execution is admitted.
pub(crate) async fn wait_for_mcp_server(self: &Arc<Self>, server: &str) {
@@ -70,6 +61,10 @@ impl Session {
.and_then(|environment| environment.cwd().to_abs_path().ok())
.unwrap_or_else(|| session_configuration.cwd().clone());
let config = Self::build_per_turn_config(&session_configuration, cwd);
let local_process_cwd = environments
.local_environment_cwd()
.unwrap_or_else(|| session_configuration.cwd().clone())
.to_path_buf();
McpDesiredState {
config: Arc::new(config),
@@ -78,6 +73,7 @@ impl Session {
originator: session_configuration.originator.clone(),
session_source: session_configuration.session_source.clone(),
environments,
local_process_cwd,
windows_sandbox_level: session_configuration.windows_sandbox_level,
}
}
@@ -88,11 +84,15 @@ impl Session {
auth: Option<CodexAuth>,
mcp_projection: McpRuntimeProjection,
resolved_environments: &TurnEnvironmentSnapshot,
local_stdio_fallback_cwd: PathBuf,
mcp_runtime_cwd: PathBuf,
) -> anyhow::Result<()> {
let cwd = AbsolutePathBuf::from_absolute_path(local_stdio_fallback_cwd)
let cwd = AbsolutePathBuf::from_absolute_path(mcp_runtime_cwd)
.unwrap_or_else(|_| session_configuration.cwd().clone());
let config = Self::build_per_turn_config(session_configuration, cwd);
let local_process_cwd = resolved_environments
.local_environment_cwd()
.unwrap_or_else(|| session_configuration.cwd().clone())
.to_path_buf();
let desired = McpDesiredState {
config: Arc::new(config),
auth,
@@ -100,6 +100,7 @@ impl Session {
originator: session_configuration.originator.clone(),
session_source: session_configuration.session_source.clone(),
environments: resolved_environments.clone(),
local_process_cwd,
windows_sandbox_level: session_configuration.windows_sandbox_level,
};
self.publish_mcp_runtime(
@@ -165,10 +166,9 @@ impl Session {
.or_insert_with(|| PathUri::from_abs_path(&desired.config.cwd));
let mcp_config = Arc::new(config);
let mcp_servers = effective_mcp_servers(&mcp_config, auth.as_ref());
let local_stdio_fallback_cwd = desired.local_stdio_fallback_cwd();
let runtime_context = McpRuntimeContext::new(
self.services.turn_environments.environment_manager(),
local_stdio_fallback_cwd,
desired.local_process_cwd.clone(),
);
let codex_apps_auth_manager =
codex_mcp::host_owned_codex_apps_enabled(&mcp_config, auth.as_ref())
codex-rs/core/tests/suite/mcp_startup_refresh_http_proxy.rs
+1
View File
@@ -113,6 +113,7 @@ async fn local_mcp_startup_and_refresh_use_configured_http_client() -> Result<()
"Bearer initial".to_string(),
)])),
env_http_headers: None,
http_headers_helper: None,
},
environment_id: DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
enabled: true,
codex-rs/core/tests/suite/mcp_tool_cache.rs
+39 -5
View File
@@ -4,6 +4,7 @@ use std::time::Duration;
use anyhow::Context;
use codex_config::Constrained;
use codex_config::types::McpServerConfig;
use codex_config::types::McpServerTransportConfig;
use codex_core::NewThread;
use codex_core::StartThreadOptions;
use codex_exec_server::ExecutorFileSystem;
@@ -19,6 +20,7 @@ use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use codex_utils_path_uri::PathUri;
use core_test_support::apps_test_server::AppsTestServer;
use core_test_support::is_remote_test_environment;
use core_test_support::responses;
use core_test_support::responses::ResponseMock;
use core_test_support::responses::mount_sse_once;
@@ -247,11 +249,19 @@ async fn mcp_calls_stay_bound_to_each_thread() -> anyhow::Result<()> {
Ok(())
}
#[test_case(false, false, 1; "optional server uses cache")]
#[test_case(true, false, 1; "required server uses cache")]
#[test_case(false, true, 2; "headers helper bypasses cache")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[test_case(false; "optional server")]
#[test_case(true; "required server")]
async fn cached_http_mcp_starts_lazily_for_subagents(required: bool) -> anyhow::Result<()> {
async fn cached_http_mcp_starts_lazily_for_subagents(
required: bool,
with_headers_helper: bool,
expected_startup_attempts: usize,
) -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
if with_headers_helper && is_remote_test_environment() {
return Ok(());
}
let responses_server = responses::start_mock_server().await;
let (http_server, startup_control) =
@@ -287,6 +297,24 @@ async fn cached_http_mcp_starts_lazily_for_subagents(required: bool) -> anyhow::
wait_for_mcp_server(&fixture.codex, SERVER_NAME).await?;
assert_eq!(startup_control.initialize_attempts(), 1);
let mut subagent_config = fixture.config.clone();
if with_headers_helper {
let mut servers = subagent_config.mcp_servers.get().clone();
let server = servers.get_mut(SERVER_NAME).expect("cached HTTP server");
let McpServerTransportConfig::StreamableHttp {
http_headers_helper,
..
} = &mut server.transport
else {
unreachable!("expected HTTP transport");
};
*http_headers_helper = Some(if cfg!(windows) {
r#"echo {"X-Cache-Test":"helper"}"#.to_string()
} else {
r#"printf '{"X-Cache-Test":"helper"}'"#.to_string()
});
subagent_config.mcp_servers.set(servers)?;
}
let NewThread {
thread: subagent, ..
} = fixture
@@ -299,10 +327,16 @@ async fn cached_http_mcp_starts_lazily_for_subagents(required: bool) -> anyhow::
agent_nickname: None,
agent_role: None,
})),
..StartThreadOptions::new(fixture.config.clone())
..StartThreadOptions::new(subagent_config)
})
.await?;
assert_eq!(startup_control.initialize_attempts(), 1);
if with_headers_helper {
wait_for_mcp_server(&subagent, SERVER_NAME).await?;
}
assert_eq!(
startup_control.initialize_attempts(),
expected_startup_attempts
);
let call_id = "http-call";
let call_response = mount_sse_once(
codex-rs/core/tests/suite/rmcp_client.rs
+42 -9
View File
@@ -1010,6 +1010,7 @@ async fn apps_enabled_turn_skips_pending_optional_mcp_without_cached_tools() ->
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
TestMcpServerOptions::default(),
);
@@ -1080,6 +1081,7 @@ async fn shutdown_cancels_startup_prewarm_waiting_for_mcp_startup() -> anyhow::R
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
TestMcpServerOptions::default(),
);
@@ -1142,6 +1144,7 @@ async fn interrupt_during_mcp_startup_preserves_user_input_in_history(
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
TestMcpServerOptions::default(),
);
@@ -2790,9 +2793,14 @@ impl StreamableHttpTestServer {
/// What this tests: Codex can discover and call a Streamable HTTP MCP tool in
/// both local and remote-aware placements, and the tool observes the expected
/// environment value from the server process that actually handled the request.
#[test_case(false; "plain")]
#[test_case(true; "headers helper")]
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
async fn streamable_http_tool_call_round_trip(with_headers_helper: bool) -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
if with_headers_helper && is_remote_test_environment() {
return Ok(());
}
// Phase 1: script the model responses so Codex will call the MCP echo tool
// and then complete the turn after the tool result is returned.
@@ -2832,12 +2840,23 @@ async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
// placement. In full CI this may be the remote environment container; locally
// it is a host process.
let expected_env_value = "propagated-env-http";
let Some(http_server) =
start_streamable_http_test_server(expected_env_value, /*expected_token*/ None).await?
let Some(http_server) = start_streamable_http_test_server(
expected_env_value,
/*expected_token*/ None,
with_headers_helper.then_some("gateway-token"),
)
.await?
else {
return Ok(());
};
let server_url = http_server.url().to_string();
let http_headers_helper = with_headers_helper.then(|| {
if cfg!(windows) {
r#"echo {"Proxy-Authorization":"Bearer gateway-token"}"#.to_string()
} else {
r#"printf '{"Proxy-Authorization":"Bearer gateway-token"}'"#.to_string()
}
});
// Phase 3: configure Codex with the Streamable HTTP MCP server and build a
// fixture that selects remote MCP placement only when the remote test
@@ -2852,6 +2871,7 @@ async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper,
},
TestMcpServerOptions {
environment_id: remote_aware_environment_id(),
@@ -2921,7 +2941,6 @@ async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
.and_then(Value::as_str)
.expect("env snapshot inserted");
assert_eq!(env_value, expected_env_value);
// Phase 7: verify the scripted model calls were consumed and clean up the
// placement-aware MCP server.
wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
@@ -2938,8 +2957,12 @@ async fn streamable_http_configured_auth_precedes_chatgpt_auth() -> anyhow::Resu
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let Some(configured_auth_server) =
start_streamable_http_test_server("configured-auth", Some("configured-token")).await?
let Some(configured_auth_server) = start_streamable_http_test_server(
"configured-auth",
Some("configured-token"),
/*expected_gateway_token*/ None,
)
.await?
else {
return Ok(());
};
@@ -2959,6 +2982,7 @@ async fn streamable_http_configured_auth_precedes_chatgpt_auth() -> anyhow::Resu
"Bearer configured-token".to_string(),
)])),
env_http_headers: None,
http_headers_helper: None,
},
TestMcpServerOptions {
environment_id: remote_aware_environment_id(),
@@ -2999,6 +3023,7 @@ async fn streamable_http_chatgpt_auth_is_not_sent_to_configured_origin() -> anyh
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
TestMcpServerOptions {
auth: McpServerAuth::ChatGpt,
@@ -3190,8 +3215,12 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
let expected_token = "initial-access-token";
let client_id = "test-client-id";
let refresh_token = "initial-refresh-token";
let Some(http_server) =
start_streamable_http_test_server(expected_env_value, Some(expected_token)).await?
let Some(http_server) = start_streamable_http_test_server(
expected_env_value,
Some(expected_token),
/*expected_gateway_token*/ None,
)
.await?
else {
return Ok(());
};
@@ -3249,6 +3278,7 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
"Authorization".to_string(),
unset_authorization_env_var,
)])),
http_headers_helper: None,
},
TestMcpServerOptions {
environment_id,
@@ -3518,6 +3548,7 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
async fn start_streamable_http_test_server(
expected_env_value: &str,
expected_token: Option<&str>,
expected_gateway_token: Option<&str>,
) -> anyhow::Result<Option<StreamableHttpTestServer>> {
let rmcp_http_server_bin = match cargo_bin("test_streamable_http_server") {
Ok(path) => path,
@@ -3553,6 +3584,9 @@ async fn start_streamable_http_test_server(
if let Some(expected_token) = expected_token {
command.env("MCP_EXPECT_BEARER", expected_token);
}
if let Some(expected_gateway_token) = expected_gateway_token {
command.env("MCP_EXPECT_GATEWAY_BEARER", expected_gateway_token);
}
let mut child = command.spawn()?;
wait_for_local_streamable_http_server(&mut child, &server_url, Duration::from_secs(5)).await?;
@@ -3593,7 +3627,6 @@ async fn start_remote_streamable_http_test_server(
sh_single_quote(expected_token)
));
}
let script = format!(
"{} nohup {} > {} 2>&1 < /dev/null & echo $!",
env_assignments.join(" "),
codex-rs/ext/mcp/src/executor_plugin/provider_tests.rs
+1
View File
@@ -201,6 +201,7 @@ async fn reads_declared_config_only_through_executor_file_system() {
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
http_headers_helper: None,
},
environment_id: "executor-test".to_string(),
enabled: true,
codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs
+39
View File
@@ -112,6 +112,17 @@ struct EchoArgs {
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut args = std::env::args_os().skip(1);
match args.next().as_deref() {
Some(value) if value == std::ffi::OsStr::new("--http-headers-helper") => {
if std::env::var_os("MCP_TEST_AMBIENT_SECRET").is_some() {
return Err("helper inherited ambient secret".into());
}
println!(r#"{{"Proxy-Authorization":"Bearer gateway-token"}}"#);
return Ok(());
}
_ => {}
}
let bind_addr = parse_bind_addr()?;
let post_failure_state = PostFailureState::default();
const MAX_BIND_RETRIES: u32 = 20;
@@ -209,6 +220,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
} else {
router
};
let router = if let Ok(token) = std::env::var("MCP_EXPECT_GATEWAY_BEARER") {
let expected = Arc::new(format!("Bearer {token}"));
router.layer(middleware::from_fn_with_state(
expected,
require_gateway_bearer,
))
} else {
router
};
axum::serve(listener, router).await?;
task::yield_now().await;
@@ -412,6 +432,25 @@ async fn require_bearer(
}
}
async fn require_gateway_bearer(
State(expected): State<Arc<String>>,
request: Request<Body>,
next: Next,
) -> Result<Response, StatusCode> {
if !request.uri().path().starts_with("/mcp") {
return Ok(next.run(request).await);
}
if request
.headers()
.get("proxy-authorization")
.is_some_and(|value| value.as_bytes() == expected.as_bytes())
{
Ok(next.run(request).await)
} else {
Err(StatusCode::UNAUTHORIZED)
}
}
async fn arm_session_post_failure(
State(state): State<PostFailureState>,
Json(request): Json<ArmSessionPostFailureRequest>,
codex-rs/rmcp-client/src/http_headers.rs
+366
View File
@@ -0,0 +1,366 @@
use std::collections::HashSet;
use std::fmt;
#[cfg(windows)]
use std::os::windows::process::CommandExt;
use std::path::Path;
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use anyhow::anyhow;
use codex_exec_server::ExecServerError;
use codex_exec_server::HttpClient;
use codex_exec_server::HttpHeader;
use codex_exec_server::HttpRedirectPolicy;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::HttpRequestResponse;
use codex_exec_server::HttpResponseBodyStream;
#[cfg(all(unix, not(target_os = "macos")))]
use codex_utils_pty::process_group::kill_process_group;
#[cfg(target_os = "macos")]
use codex_utils_pty::process_group::kill_process_group_with_member_fallback as kill_process_group;
use futures::FutureExt;
use futures::future::BoxFuture;
use futures::future::Shared;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
use serde::Deserialize;
use serde::de::MapAccess;
use serde::de::Visitor;
use tokio::io::AsyncReadExt;
use tokio::process::Child;
use tokio::process::Command;
use tokio::time::Instant;
use url::Origin;
use url::Url;
use crate::utils::create_env_for_mcp_server;
const HELPER_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_HELPER_OUTPUT_BYTES: usize = 64 * 1024;
type CachedHeaders = Shared<BoxFuture<'static, std::result::Result<Arc<HeaderMap>, Arc<str>>>>;
struct HttpHeadersProvider {
server_origin: Origin,
cached: CachedHeaders,
}
struct HttpHeadersClient {
inner: Arc<dyn HttpClient>,
provider: HttpHeadersProvider,
}
struct HelperProcess {
child: Child,
#[cfg(unix)]
process_group_id: u32,
#[cfg(windows)]
job: codex_utils_pty::JobObject,
}
struct RawHeaderEntries {
// Keep raw entries because ordinary map deserialization collapses exact duplicate keys.
entries: Vec<(String, String)>,
has_exact_duplicate: bool,
}
struct RawHeaderEntriesVisitor;
impl<'de> Visitor<'de> for RawHeaderEntriesVisitor {
type Value = RawHeaderEntries;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("a JSON object of string header names and values")
}
fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
where
A: MapAccess<'de>,
{
let mut entries = Vec::with_capacity(map.size_hint().unwrap_or_default());
let mut names = HashSet::new();
let mut has_exact_duplicate = false;
while let Some((name, value)) = map.next_entry::<String, String>()? {
has_exact_duplicate |= !names.insert(name.clone());
entries.push((name, value));
}
Ok(RawHeaderEntries {
entries,
has_exact_duplicate,
})
}
}
impl<'de> Deserialize<'de> for RawHeaderEntries {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
deserializer.deserialize_map(RawHeaderEntriesVisitor)
}
}
impl Drop for HelperProcess {
fn drop(&mut self) {
#[cfg(unix)]
let _ = kill_process_group(self.process_group_id);
#[cfg(windows)]
let _ = self.job.terminate();
let _ = self.child.start_kill();
}
}
impl HttpHeadersProvider {
fn new(server_url: &str, command: &str, cwd: PathBuf) -> Result<Self> {
let command = command.to_string();
let cached = async move {
run_helper(&command, &cwd)
.await
.map(Arc::new)
.map_err(|error| Arc::<str>::from(error.to_string()))
}
.boxed()
.shared();
Ok(Self {
server_origin: Url::parse(server_url)?.origin(),
cached,
})
}
async fn headers(&self) -> Result<Arc<HeaderMap>, ExecServerError> {
self.cached
.clone()
.await
.map_err(|error| ExecServerError::HttpRequest(error.to_string()))
}
}
/// No rejection-driven refresh: 401/403 may be OAuth challenges, and reconnecting loses sessions.
pub fn with_http_headers_helper(
inner: Arc<dyn HttpClient>,
server_url: &str,
command: &str,
cwd: PathBuf,
) -> Result<Arc<dyn HttpClient>> {
let provider = HttpHeadersProvider::new(server_url, command, cwd)?;
Ok(Arc::new(HttpHeadersClient { inner, provider }))
}
impl HttpHeadersClient {
async fn prepare_request(
&self,
mut params: HttpRequestParams,
) -> Result<HttpRequestParams, ExecServerError> {
let Ok(url) = Url::parse(&params.url) else {
return Ok(params);
};
if self.provider.server_origin != url.origin() {
return Ok(params);
}
// TODO: Follow same-origin redirects once later hops cannot leak helper headers.
params.redirect_policy = HttpRedirectPolicy::Stop;
let deadline = params
.timeout_ms
.map(|timeout_ms| Instant::now() + Duration::from_millis(timeout_ms));
let headers = match deadline {
Some(deadline) => tokio::time::timeout_at(deadline, self.provider.headers())
.await
.map_err(|_| {
ExecServerError::HttpRequest("HTTP request timed out".to_string())
})??,
None => self.provider.headers().await?,
};
for (name, value) in headers.iter() {
params
.headers
.retain(|header| !header.name.eq_ignore_ascii_case(name.as_str()));
params.headers.push(HttpHeader {
name: name.to_string(),
value: std::str::from_utf8(value.as_bytes())
.map_err(|error| ExecServerError::HttpRequest(error.to_string()))?
.to_string(),
});
}
if let Some(deadline) = deadline {
let remaining = deadline.saturating_duration_since(Instant::now());
params.timeout_ms = Some(
u64::try_from(remaining.as_millis())
.unwrap_or(u64::MAX)
.max(1),
);
}
Ok(params)
}
}
impl HttpClient for HttpHeadersClient {
fn http_request(
&self,
params: HttpRequestParams,
) -> BoxFuture<'_, Result<HttpRequestResponse, ExecServerError>> {
async move {
let params = self.prepare_request(params).await?;
self.inner.http_request(params).await
}
.boxed()
}
fn http_request_stream(
&self,
params: HttpRequestParams,
) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> {
async move {
let params = self.prepare_request(params).await?;
self.inner.http_request_stream(params).await
}
.boxed()
}
}
async fn run_helper(command: &str, cwd: &Path) -> Result<HeaderMap> {
#[cfg(windows)]
let shell = std::env::var_os("COMSPEC").unwrap_or_else(|| "cmd.exe".into());
#[cfg(not(windows))]
let shell = "sh";
// Match the repository's existing shell-command convention. The command is ordinary
// configuration and may be visible in local process metadata; credentials belong in the
// JSON output rather than in the command text.
let mut process = Command::new(shell);
#[cfg(windows)]
{
process.args(["/Q", "/D", "/C"]);
process.as_std_mut().raw_arg(format!(r#""{command}""#));
}
#[cfg(not(windows))]
process.args(["-c", command]);
#[cfg(unix)]
process.process_group(0);
process
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.current_dir(cwd)
// Match local MCP subprocess policy; arbitrary ambient variables are not inherited.
.env_clear()
.envs(create_env_for_mcp_server(/*extra_env*/ None, &[])?)
.kill_on_drop(true);
#[cfg(windows)]
let (child, job) = {
let job = codex_utils_pty::JobObject::create_without_breakaway()
.map_err(|error| anyhow!("MCP HTTP headers helper containment failed: {error}"))?;
let child = job
.spawn_contained(&mut process)
.map_err(|error| anyhow!("MCP HTTP headers helper failed to start: {error}"))?;
(child, job)
};
#[cfg(not(windows))]
let child = process
.spawn()
.map_err(|error| anyhow!("MCP HTTP headers helper failed to start: {error}"))?;
let mut process = HelperProcess {
#[cfg(unix)]
process_group_id: child
.id()
.ok_or_else(|| anyhow!("MCP HTTP headers helper process id was unavailable"))?,
child,
#[cfg(windows)]
job,
};
let output = tokio::time::timeout(HELPER_TIMEOUT, async {
let stdout = process
.child
.stdout
.take()
.ok_or_else(|| anyhow!("MCP HTTP headers helper stdout was unavailable"))?;
let mut output = Vec::new();
stdout
.take((MAX_HELPER_OUTPUT_BYTES + 1) as u64)
.read_to_end(&mut output)
.await?;
if output.len() > MAX_HELPER_OUTPUT_BYTES {
return Err(anyhow!("MCP HTTP headers helper output exceeds 64 KiB"));
}
let status = process.child.wait().await?;
if !status.success() {
return Err(anyhow!(
"MCP HTTP headers helper exited with status {status}"
));
}
Ok(output)
})
.await
.map_err(|_| anyhow!("MCP HTTP headers helper timed out after 10 seconds"))??;
parse_helper_output(output)
}
fn parse_helper_output(stdout: Vec<u8>) -> Result<HeaderMap> {
let stdout = String::from_utf8(stdout)
.map_err(|_| anyhow!("MCP HTTP headers helper wrote non-UTF-8 data"))?;
let mut deserializer = serde_json::Deserializer::from_str(stdout.trim());
let headers = RawHeaderEntries::deserialize(&mut deserializer)
.and_then(|headers| {
deserializer.end()?;
Ok(headers)
})
.map_err(|_| anyhow!("MCP HTTP headers helper must output a JSON object of strings"))?;
if headers.has_exact_duplicate {
return Err(anyhow!(
"MCP HTTP headers helper returned duplicate header names"
));
}
let mut parsed = HeaderMap::with_capacity(headers.entries.len());
for (name, value) in headers.entries {
let name = HeaderName::from_bytes(name.as_bytes())
.map_err(|_| anyhow!("MCP HTTP headers helper returned an invalid header name"))?;
// Helper values replace same-name configured headers; bearer/OAuth owns Authorization.
// Google IAP uses Proxy-Authorization alongside application Authorization. For HTTPS MCP
// URLs it is sent through the forward-proxy tunnel to IAP, not used as CONNECT auth.
if matches!(
name.as_str(),
"accept"
| "authorization"
| "connection"
| "content-encoding"
| "content-length"
| "content-type"
| "host"
| "keep-alive"
| "last-event-id"
| "mcp-protocol-version"
| "mcp-session-id"
| "origin"
| "proxy-connection"
| "referer"
| "te"
| "trailer"
| "transfer-encoding"
| "upgrade"
) {
return Err(anyhow!(
"MCP HTTP headers helper returned a reserved header"
));
}
if parsed.contains_key(&name) {
return Err(anyhow!(
"MCP HTTP headers helper returned duplicate header names"
));
}
let value = HeaderValue::from_str(&value)
.map_err(|_| anyhow!("MCP HTTP headers helper returned an invalid header value"))?;
parsed.insert(name, value);
}
Ok(parsed)
}
#[cfg(test)]
#[path = "http_headers_tests.rs"]
mod tests;
codex-rs/rmcp-client/src/http_headers_tests.rs
+200
View File
@@ -0,0 +1,200 @@
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn helper_output_errors_do_not_echo_secrets() {
for output in [
br#"{"Authorization":"secret"}"#.as_slice(),
br#"{"secret":"secret","secret":"secret"}"#.as_slice(),
br#"{"secret":"secret","Secret":"secret"}"#.as_slice(),
] {
let error = parse_helper_output(output.to_vec()).expect_err("invalid helper output");
assert!(!error.to_string().contains("secret"));
}
}
#[cfg(unix)]
#[tokio::test]
async fn helper_attempt_is_shared_after_cancellation() {
use tempfile::tempdir;
let temp = tempdir().expect("temporary helper directory");
let cwd = temp
.path()
.canonicalize()
.expect("canonical helper directory");
let cancelled_invocations = cwd.join("cancelled-invocations");
let cancelled = HttpHeadersProvider::new(
"https://example.com",
&format!(
"test \"$(pwd)\" = '{0}'; test -n \"$HOME\"; test -n \"$PATH\"; \
printf x >> '{1}'; sleep 0.2; printf '{{\"X-Gateway\":\"token\"}}'",
cwd.display(),
cancelled_invocations.display(),
),
cwd.clone(),
)
.expect("cancelled provider");
assert!(
tokio::time::timeout(Duration::from_millis(20), cancelled.headers())
.await
.is_err()
);
assert!(cancelled.headers().await.is_ok());
assert_eq!(
std::fs::read_to_string(cancelled_invocations).expect("cancelled invocation count"),
"x"
);
let dropped_started = cwd.join("dropped-helper-started");
let dropped_finished = cwd.join("dropped-helper-finished");
let dropped = HttpHeadersProvider::new(
"https://example.com",
&format!(
"printf x > '{}'; sleep 1; printf x > '{}'",
dropped_started.display(),
dropped_finished.display(),
),
cwd,
)
.expect("dropped provider");
assert!(
tokio::time::timeout(Duration::from_millis(500), dropped.headers())
.await
.is_err()
);
assert!(dropped_started.exists());
drop(dropped);
tokio::time::sleep(Duration::from_millis(1_100)).await;
assert!(!dropped_finished.exists());
}
#[tokio::test]
async fn nonzero_helper_exit_is_cached() {
let temp = tempfile::tempdir().expect("temporary helper directory");
let failed_invocations = temp.path().join("failed-invocations");
let command = if cfg!(windows) {
format!(
r#"echo x>>"{0}" & echo {{"X-Gateway":"valid"}} & exit /b 23"#,
failed_invocations.display()
)
} else {
format!(
"echo x >> '{0}'; printf '{{\"X-Gateway\":\"valid\"}}'; exit 23",
failed_invocations.display()
)
};
let failed = HttpHeadersProvider::new(
"https://example.com/mcp",
&command,
temp.path().to_path_buf(),
)
.expect("failed provider");
let first = failed.headers().await.expect_err("failed helper");
let second = failed.headers().await.expect_err("cached failure");
assert_eq!(first.to_string(), second.to_string());
let invocations = std::fs::read_to_string(failed_invocations).expect("failed invocation count");
assert_eq!(invocations.lines().count(), 1);
}
#[cfg(unix)]
#[tokio::test]
async fn connection_headers_are_cached_and_origin_bound() {
use axum::Router;
use axum::http::StatusCode;
use axum::response::Redirect;
use axum::routing::get;
use axum::routing::post;
use codex_exec_server::RouteAwareHttpClient;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use std::sync::Arc;
use tempfile::tempdir;
use tokio::net::TcpListener;
async fn handle(headers: axum::http::HeaderMap) -> StatusCode {
assert_eq!(
headers.get("proxy-authorization"),
Some(&HeaderValue::from_static("Bearer token"))
);
assert_eq!(
headers.get("x-label"),
Some(&HeaderValue::from_bytes("café".as_bytes()).unwrap())
);
StatusCode::NO_CONTENT
}
let temp = tempdir().expect("temporary helper directory");
let invocation_file = temp.path().join("invocations");
let cross_listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("bind cross-origin server");
let cross_url = format!("http://{}/start", cross_listener.local_addr().unwrap());
tokio::spawn(async move {
axum::serve(
cross_listener,
Router::new()
.route("/start", get(|| async { Redirect::temporary("/final") }))
.route("/final", get(|| async { StatusCode::NO_CONTENT })),
)
.await
.expect("serve cross-origin requests");
});
let listener = TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let url = format!("http://{}/mcp", listener.local_addr().unwrap());
let redirect_url = cross_url.clone();
let app = Router::new().route("/mcp", post(handle)).route(
"/redirect",
get(move || std::future::ready(Redirect::temporary(&redirect_url))),
);
tokio::spawn(async move {
axum::serve(listener, app)
.await
.expect("serve test requests");
});
let command = format!(
"printf x >> '{}'; printf '{{\"Proxy-Authorization\":\"Bearer token\",\"X-Label\":\"café\"}}'",
invocation_file.display(),
);
let inner: Arc<dyn HttpClient> = Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
OutboundProxyPolicy::ReqwestDefault,
)));
let client = with_http_headers_helper(inner, &url, &command, temp.path().to_path_buf())
.expect("headers helper client");
let request = |session: &str| HttpRequestParams {
method: "POST".to_string(),
url: url.clone(),
headers: Vec::new(),
body: None,
timeout_ms: Some(5_000),
redirect_policy: HttpRedirectPolicy::Follow,
request_id: session.to_string(),
stream_response: true,
};
let mut cross_request = request("cross-origin");
cross_request.method = "GET".to_string();
cross_request.url = cross_url;
assert_eq!(
client.http_request(cross_request).await.unwrap().status,
204
);
assert!(!invocation_file.exists());
let mut redirect_request = request("redirect");
redirect_request.method = "GET".to_string();
redirect_request.url = url.replace("/mcp", "/redirect");
assert_eq!(
client.http_request(redirect_request).await.unwrap().status,
307
);
let (left, right) = tokio::join!(
client.http_request_stream(request("session-a")),
client.http_request_stream(request("session-b"))
);
assert_eq!(left.expect("left request").0.status, 204);
assert_eq!(right.expect("right request").0.status, 204);
assert_eq!(
std::fs::read_to_string(&invocation_file).expect("helper invocation count"),
"x"
);
}
codex-rs/rmcp-client/src/lib.rs
+2
View File
@@ -3,6 +3,7 @@ mod elicitation_client_service;
mod event_notification_transport;
mod executor_process_transport;
mod http_client_adapter;
mod http_headers;
mod in_process_transport;
mod incoming_jsonrpc;
mod local_stdio_transport;
@@ -28,6 +29,7 @@ pub use auth_status::discover_streamable_http_oauth;
pub use codex_protocol::protocol::McpAuthStatus;
pub use event_notification_transport::EventNotificationReceiver;
pub use http_client_adapter::StreamableHttpRedirectMode;
pub use http_headers::with_http_headers_helper;
pub use in_process_transport::InProcessTransportFactory;
pub use oauth::StoredOAuthCredentialSnapshot;
pub use oauth::StoredOAuthTokens;
codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs
+50 -5
View File
@@ -17,6 +17,8 @@ use codex_rmcp_client::WrappedOAuthTokenResponse;
use codex_rmcp_client::determine_streamable_http_auth_status;
use codex_rmcp_client::is_authentication_required_error;
use codex_rmcp_client::save_oauth_tokens;
use codex_rmcp_client::with_http_headers_helper;
use codex_utils_cargo_bin::cargo_bin;
use oauth2::AccessToken;
use oauth2::RefreshToken;
use oauth2::basic::BasicTokenType;
@@ -43,12 +45,22 @@ const EXPIRED_ACCESS_TOKEN: &str = "expired-access-token";
const REFRESH_TOKEN: &str = "valid-refresh-token";
const REFRESHED_ACCESS_TOKEN: &str = "refreshed-access-token";
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 UNREFRESHABLE_SERVER_URL: &str = "https://unrefreshable.example/mcp";
const UNEXPIRED_SERVER_URL: &str = "https://unexpired.example/mcp";
const REFRESHABLE_SERVER_URL: &str = "https://refreshable.example/mcp";
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refreshes_expired_persisted_token_before_initialize() -> anyhow::Result<()> {
assert_expired_token_refresh(/*with_headers_helper*/ false).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn refreshes_oauth_with_gateway_headers_helper() -> anyhow::Result<()> {
assert_expired_token_refresh(/*with_headers_helper*/ true).await
}
async fn assert_expired_token_refresh(with_headers_helper: bool) -> anyhow::Result<()> {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/.well-known/oauth-authorization-server/mcp"))
@@ -94,7 +106,12 @@ async fn refreshes_expired_persisted_token_before_initialize() -> anyhow::Result
format!("Bearer {REFRESHED_ACCESS_TOKEN}"),
))
.respond_with(|request: &Request| {
let body: Value = request.body_json().expect("valid JSON-RPC request");
let body: Value = match request.body_json() {
Ok(body) => body,
Err(_) => {
return ResponseTemplate::new(400).set_body_string("invalid JSON-RPC request");
}
};
match body.get("method").and_then(Value::as_str) {
Some("initialize") => ResponseTemplate::new(200).set_body_json(json!({
"jsonrpc": "2.0",
@@ -126,13 +143,32 @@ async fn refreshes_expired_persisted_token_before_initialize() -> anyhow::Result
// Credential storage resolves CODEX_HOME from the process environment.
// Run the client half of the test in an ignored helper test so it can use
// an isolated home without mutating the parent test runner's environment.
let status = Command::new(std::env::current_exe()?)
let mut command = Command::new(std::env::current_exe()?);
command
.args(["oauth_startup_child", "--exact", "--ignored", "--nocapture"])
.env("CODEX_HOME", codex_home.path())
.env(CHILD_SERVER_URL_ENV, server_url)
.status()
.await?;
.env("MCP_TEST_AMBIENT_SECRET", "must-not-reach-helper");
if with_headers_helper {
command.env(
CHILD_HELPER_COMMAND_ENV,
format!(
"\"{}\" --http-headers-helper",
cargo_bin("test_streamable_http_server")?.display(),
),
);
}
let status = command.status().await?;
assert!(status.success(), "OAuth startup child failed: {status}");
if with_headers_helper {
let requests = server.received_requests().await.unwrap_or_default();
assert!(requests.iter().all(|request| {
request
.headers
.get("proxy-authorization")
.is_some_and(|value| value == "Bearer gateway-token")
}));
}
server.verify().await;
Ok(())
}
@@ -334,6 +370,15 @@ async fn oauth_startup_child() -> anyhow::Result<()> {
// This mirrors create_client's transport and initialization setup, except
// it omits the direct bearer token. Supplying that token would bypass the
// persisted OAuth credentials and the startup refresh under test.
let mut http_client = Environment::default_for_tests().get_http_client();
if let Ok(helper_command) = std::env::var(CHILD_HELPER_COMMAND_ENV) {
http_client = with_http_headers_helper(
http_client,
&server_url,
&helper_command,
std::env::current_dir()?,
)?;
}
let client = RmcpClient::new_streamable_http_client(
SERVER_NAME,
&server_url,
@@ -342,7 +387,7 @@ async fn oauth_startup_child() -> anyhow::Result<()> {
/*env_http_headers*/ None,
OAuthCredentialsStoreMode::File,
AuthKeyringBackendKind::default(),
Environment::default_for_tests().get_http_client(),
http_client,
/*auth_provider*/ None,
)
.await?;