Resolve HTTP MCP bearer tokens in executor environments (#39926)

## Why

Executor-owned HTTP MCP servers need to read their bearer credentials from the
selected executor environment instead of the host process.

## What changed

- Preserve `bearer_token_env_var` for executor-owned HTTP MCP configurations and
  resolve it when the executor sends each request.
- Extend delegated HTTP headers with executor-local environment references while
  rejecting missing, empty, or protected credential variables.
- Keep transport-provided bearer authentication compatible with MCP redirect and
  OAuth handling without sending a placeholder authorization value.

## Testing

- Cover authenticated executor-owned MCP requests end to end.
- Cover delegated header resolution and rejection of protected variables.
- Cover parsing executor-owned bearer configuration and transport-provided bearer
  behavior.

GitOrigin-RevId: 442bf7382198ffb69eba797eb36d4c74faabda88
This commit is contained in:
jif authored and copyberry committed 2026-08-21 12:50:15 +00:00
1 parent ff0e95007c
commit 93c54bca38
21 files changed
+361 -57

No files matched your search

@@ -4,6 +4,7 @@ use app_test_support::TestAppServer;
use axum::Json; use axum::Json;
use axum::Router; use axum::Router;
use axum::body::Bytes; use axum::body::Bytes;
use axum::response::IntoResponse;
use axum::routing::get; use axum::routing::get;
use axum::routing::post; use axum::routing::post;
use codex_app_server_protocol::CapabilityRootLocation; use codex_app_server_protocol::CapabilityRootLocation;
@@ -43,10 +44,14 @@ use rmcp::transport::streamable_http_server::session::local::LocalSessionManager
use serde_json::json; use serde_json::json;
use std::borrow::Cow; use std::borrow::Cow;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::process::Stdio;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tempfile::TempDir; use tempfile::TempDir;
use tokio::io::AsyncBufReadExt;
use tokio::io::BufReader;
use tokio::net::TcpListener; use tokio::net::TcpListener;
use tokio::process::Command;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::time::timeout; use tokio::time::timeout;
use url::Url; 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_OAUTH_ACCESS_TOKEN: &str = "executor-access-token";
const EXECUTOR_ENV_NAME: &str = "MCP_EXECUTOR_MARKER"; const EXECUTOR_ENV_NAME: &str = "MCP_EXECUTOR_MARKER";
const EXECUTOR_ENV_VALUE: &str = "executor-only"; 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 EXECUTOR_ID: &str = "executor-1";
const REFRESH_PROBE_SERVER_NAME: &str = "refresh_probe"; const REFRESH_PROBE_SERVER_NAME: &str = "refresh_probe";
const TOOL_CALL_ID: &str = "executor-mcp-call"; const TOOL_CALL_ID: &str = "executor-mcp-call";
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Result<()> { 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 responses_server = responses::start_mock_server().await;
let http_listener = TcpListener::bind("127.0.0.1:0").await?; let http_listener = TcpListener::bind("127.0.0.1:0").await?;
let http_addr = http_listener.local_addr()?; 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()), Arc::new(LocalSessionManager::default()),
http_server_config, 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_authorization_tx, mut oauth_authorization_rx) = mpsc::unbounded_channel();
let oauth_mcp_router = Router::new() let oauth_mcp_router = Router::new()
.nest_service("/oauth-mcp", oauth_mcp_service) .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); .merge(oauth_mcp_router);
let http_server_handle = tokio::spawn(async move { let http_server_handle = tokio::spawn(async move {
let _ = axum::serve(http_listener, http_router).await; 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, &oauth_credentials_path,
serde_json::to_vec(&json!({"host": host_oauth_credential.clone()}))?, serde_json::to_vec(&json!({"host": host_oauth_credential.clone()}))?,
)?; )?;
let codex_bin = toml::Value::String( let mut executor = Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
codex_utils_cargo_bin::cargo_bin("codex")? .args(["exec-server", "--listen", "ws://127.0.0.1:0"])
.to_string_lossy() .stdout(Stdio::piped())
.into_owned(), .kill_on_drop(true)
); .env("CODEX_HOME", codex_home.path())
let http_proxy = toml::Value::String(format!("http://{http_addr}")); .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( std::fs::write(
codex_home.path().join("environments.toml"), codex_home.path().join("environments.toml"),
format!( format!(
@@ -197,11 +231,7 @@ include_local = true
[[environments]] [[environments]]
id = "{EXECUTOR_ID}" id = "{EXECUTOR_ID}"
program = {codex_bin} url = "{executor_url}"
args = ["exec-server", "--listen", "stdio"]
[environments.env]
{EXECUTOR_ENV_NAME} = "{EXECUTOR_ENV_VALUE}"
HTTP_PROXY = {http_proxy}
"# "#
), ),
)?; )?;
@@ -224,6 +254,7 @@ HTTP_PROXY = {http_proxy}
(HTTP_MCP_SERVER_NAME): { (HTTP_MCP_SERVER_NAME): {
"url": EXECUTOR_HTTP_MCP_URL, "url": EXECUTOR_HTTP_MCP_URL,
"environment_id": "local", "environment_id": "local",
"bearer_token_env_var": EXECUTOR_HTTP_AUTH_ENV_NAME,
"startup_timeout_sec": 10, "startup_timeout_sec": 10,
}, },
(OAUTH_MCP_SERVER_NAME): { (OAUTH_MCP_SERVER_NAME): {
@@ -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<dyn HttpClient>,
}
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<HttpRequestResponse, ExecServerError>> {
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)
}
}
+1
View File
@@ -99,6 +99,7 @@ mod client_capabilities;
pub(crate) mod codex_apps; pub(crate) mod codex_apps;
pub(crate) mod connection_manager; pub(crate) mod connection_manager;
pub(crate) mod elicitation; pub(crate) mod elicitation;
mod executor_environment_http_client;
pub(crate) mod mcp; pub(crate) mod mcp;
mod openai_docs_source_attribution; mod openai_docs_source_attribution;
mod pagination; mod pagination;
+2 -11
View File
@@ -183,22 +183,13 @@ fn bind_environment_env_vars(config: &mut McpServerConfig) -> Result<(), String>
let is_local_environment = config.is_local_environment(); let is_local_environment = config.is_local_environment();
let env_vars = match &mut config.transport { let env_vars = match &mut config.transport {
McpServerTransportConfig::Stdio { env_vars, .. } => env_vars, McpServerTransportConfig::Stdio { env_vars, .. } => env_vars,
// Never resolve executor-owned environment references in the host process. // Bearer credentials resolve on the executor; other header variables do not yet.
// Remove this rejection once the owning executor resolves these fields.
McpServerTransportConfig::StreamableHttp { McpServerTransportConfig::StreamableHttp {
bearer_token_env_var, env_http_headers, ..
env_http_headers,
..
} => { } => {
if is_local_environment { if is_local_environment {
return Ok(()); 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 if env_http_headers
.as_ref() .as_ref()
.is_some_and(|headers| !headers.is_empty()) .is_some_and(|headers| !headers.is_empty())
+15 -14
View File
@@ -786,7 +786,7 @@ fn environment_placement_rejects_orchestrator_env_vars() {
} }
#[test] #[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 plugin_root = plugin_root();
let outcome = parse_executor_plugin_mcp_config( let outcome = parse_executor_plugin_mcp_config(
&plugin_root_uri(&plugin_root), &plugin_root_uri(&plugin_root),
@@ -807,19 +807,20 @@ fn remote_environment_placement_rejects_http_env_references() {
assert_eq!( assert_eq!(
outcome, outcome,
PluginMcpConfigParseOutcome { PluginMcpConfigParseOutcome {
servers: BTreeMap::new(), servers: BTreeMap::from([(
errors: vec![ "bearer".to_string(),
PluginMcpServerParseError { serde_json::from_value(serde_json::json!({
name: "bearer".to_string(), "url": "https://example.com/bearer",
message: "`bearer_token_env_var` requires executor-side environment resolution for an executor-owned HTTP MCP" "bearer_token_env_var": "TOKEN",
.to_string(), "environment_id": "executor-1",
}, }))
PluginMcpServerParseError { .expect("executor-owned bearer configuration"),
name: "headers".to_string(), )]),
message: "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP" errors: vec![PluginMcpServerParseError {
.to_string(), name: "headers".to_string(),
}, message: "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP"
], .to_string(),
}],
} }
); );
} }
+16 -4
View File
@@ -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::normalize_codex_apps_tool_title;
use crate::codex_apps::prepare_openai_file_params_for_model; use crate::codex_apps::prepare_openai_file_params_for_model;
use crate::elicitation::ElicitationRequestManager; use crate::elicitation::ElicitationRequestManager;
use crate::executor_environment_http_client::ExecutorEnvironmentHttpClient;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::mcp::ToolPluginProvenance; use crate::mcp::ToolPluginProvenance;
use crate::openai_docs_source_attribution::maybe_with_openai_docs_source_attribution; 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::McpProtocolMode;
use codex_rmcp_client::RmcpClient; use codex_rmcp_client::RmcpClient;
use codex_rmcp_client::StdioServerLauncher; use codex_rmcp_client::StdioServerLauncher;
use codex_rmcp_client::StreamableHttpBearerToken;
use codex_rmcp_client::StreamableHttpRedirectMode; use codex_rmcp_client::StreamableHttpRedirectMode;
use codex_rmcp_client::ToolWithConnectorId; use codex_rmcp_client::ToolWithConnectorId;
use codex_rmcp_client::is_authentication_required_error; 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()) .http_client_for_server(server.config(), resolved_environment.as_ref())
.map_err(|error| StartupOutcomeError::from(anyhow!(error)))?; .map_err(|error| StartupOutcomeError::from(anyhow!(error)))?;
let http_client = maybe_with_openai_docs_source_attribution(&url, http_client); let http_client = maybe_with_openai_docs_source_attribution(&url, http_client);
let resolved_bearer_token = let (http_client, resolved_bearer_token) =
match resolve_bearer_token(server_name, bearer_token_env_var.as_deref()) { if !is_local_environment && let Some(env_var) = bearer_token_env_var.as_ref() {
Ok(token) => token, (
Err(error) => return Err(error.into()), Arc::new(ExecutorEnvironmentHttpClient {
bearer_token_env_var: env_var.clone(),
http_client,
}) as Arc<dyn codex_exec_server::HttpClient>,
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() { let redirect_mode = if server.is_agent_plugin() {
StreamableHttpRedirectMode::AgentPluginV1 StreamableHttpRedirectMode::AgentPluginV1
+3
View File
@@ -329,6 +329,9 @@ fn referenced_environment_variables(config: &McpServerConfig) -> Vec<(String, Op
.filter(|env_var| !env_var.is_remote_source()) .filter(|env_var| !env_var.is_remote_source())
.map(|env_var| env_var.name().to_string()) .map(|env_var| env_var.name().to_string())
.collect::<Vec<_>>(), .collect::<Vec<_>>(),
McpServerTransportConfig::StreamableHttp { .. } if !config.is_local_environment() => {
Vec::new()
}
McpServerTransportConfig::StreamableHttp { McpServerTransportConfig::StreamableHttp {
bearer_token_env_var, bearer_token_env_var,
env_http_headers, env_http_headers,
@@ -677,8 +677,11 @@ impl ExecutorCapabilityDiscoverySnapshot {
pub struct HttpHeader { pub struct HttpHeader {
/// Header name as it appears on the HTTP wire. /// Header name as it appears on the HTTP wire.
pub name: String, pub name: String,
/// Header value after UTF-8 conversion. /// Literal header value, or prefix for an executor-local environment value.
pub value: String, 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<String>,
} }
/// Redirect behavior for an executor-side HTTP request. /// Redirect behavior for an executor-side HTTP request.
@@ -12,6 +12,10 @@ use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClientFactory; use codex_http_client::HttpClientFactory;
use codex_http_client::RouteAwareClientPool; use codex_http_client::RouteAwareClientPool;
use codex_http_client::RouteAwareRequestError; 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::FutureExt;
use futures::StreamExt; use futures::StreamExt;
use futures::future::BoxFuture; use futures::future::BoxFuture;
@@ -36,6 +40,23 @@ use crate::rpc::RpcNotificationSender;
use crate::rpc::internal_error; use crate::rpc::internal_error;
use crate::rpc::invalid_params; 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. /// HTTP capability implementation backed by the shared route-aware transport.
#[derive(Clone)] #[derive(Clone)]
pub struct RouteAwareHttpClient { pub struct RouteAwareHttpClient {
@@ -288,7 +309,34 @@ impl RouteAwareHttpRequestRunner {
let name = HeaderName::from_bytes(header.name.as_bytes()).map_err(|error| { let name = HeaderName::from_bytes(header.name.as_bytes()).map_err(|error| {
invalid_params(format!("http/request header name is invalid: {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!( invalid_params(format!(
"http/request header value is invalid for {}: {error}", "http/request header value is invalid for {}: {error}",
header.name header.name
@@ -306,6 +354,7 @@ impl RouteAwareHttpRequestRunner {
Some(HttpHeader { Some(HttpHeader {
name: name.as_str().to_string(), name: name.as_str().to_string(),
value: value.to_str().ok()?.to_string(), value: value.to_str().ok()?.to_string(),
value_env_var: None,
}) })
}) })
.collect() .collect()
@@ -136,6 +136,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: vec![HttpHeader { headers: vec![HttpHeader {
name: "accept".to_string(), name: "accept".to_string(),
value: "text/event-stream".to_string(), value: "text/event-stream".to_string(),
value_env_var: None,
}], }],
body: None, body: None,
timeout_ms: None, timeout_ms: None,
@@ -154,6 +155,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: vec![HttpHeader { headers: vec![HttpHeader {
name: "content-type".to_string(), name: "content-type".to_string(),
value: "text/event-stream".to_string(), value: "text/event-stream".to_string(),
value_env_var: None,
}], }],
body: Vec::new().into(), body: Vec::new().into(),
}, },
@@ -222,6 +224,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: vec![HttpHeader { headers: vec![HttpHeader {
name: "accept".to_string(), name: "accept".to_string(),
value: "text/event-stream".to_string(), value: "text/event-stream".to_string(),
value_env_var: None,
}], }],
body: None, body: None,
timeout_ms: None, timeout_ms: None,
@@ -239,6 +242,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: vec![HttpHeader { headers: vec![HttpHeader {
name: "content-type".to_string(), name: "content-type".to_string(),
value: "text/event-stream".to_string(), value: "text/event-stream".to_string(),
value_env_var: None,
}], }],
body: Vec::new().into(), body: Vec::new().into(),
} }
+97 -5
View File
@@ -49,7 +49,14 @@ struct CapturedHttpRequest {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> { async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> {
// Phase 1: start exec-server and complete the JSON-RPC handshake. // 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?; initialize_exec_server(&mut server).await?;
// Phase 2: start a local HTTP peer and ask exec-server to POST to it. // 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 { serde_json::to_value(HttpRequestParams {
method: "POST".to_string(), method: "POST".to_string(),
url, url,
headers: vec![HttpHeader { headers: vec![
name: "x-codex-test".to_string(), HttpHeader {
value: "buffered".to_string(), 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()), body: Some(b"request-body".to_vec().into()),
timeout_ms: Some(5_000), timeout_ms: Some(5_000),
redirect_policy: HttpRedirectPolicy::Follow, 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.request_line.as_str(),
captured.headers.get("x-codex-test").map(String::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(), captured.body.as_slice(),
), ),
( (
"POST /mcp?case=buffered HTTP/1.1", "POST /mcp?case=buffered HTTP/1.1",
Some("buffered"), Some("buffered"),
Some("Bearer executor-token"),
Some("Bearer linear-token"),
b"request-body".as_slice(), b"request-body".as_slice(),
) )
); );
@@ -113,6 +140,70 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()>
Ok(()) 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 /// What this tests: delegated HTTP accepts URLs with client-side fragments and
/// does not include those fragments in the request sent over the network. /// does not include those fragments in the request sent over the network.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[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 { headers: vec![HttpHeader {
name: "accept".to_string(), name: "accept".to_string(),
value: "text/event-stream".to_string(), value: "text/event-stream".to_string(),
value_env_var: None,
}], }],
body: None, body: None,
timeout_ms: Some(5_000), timeout_ms: Some(5_000),
@@ -711,6 +711,7 @@ fn protocol_headers(headers: &HeaderMap) -> Vec<HttpHeader> {
Some(HttpHeader { Some(HttpHeader {
name: name.as_str().to_string(), name: name.as_str().to_string(),
value: std::str::from_utf8(value.as_bytes()).ok()?.to_string(), value: std::str::from_utf8(value.as_bytes()).ok()?.to_string(),
value_env_var: None,
}) })
}) })
.collect() .collect()
@@ -107,10 +107,12 @@ fn selects_bearer_challenge_from_a_later_www_authenticate_field_value() {
HttpHeader { HttpHeader {
name: "www-authenticate".to_string(), name: "www-authenticate".to_string(),
value: r#"Basic realm="example""#.to_string(), value: r#"Basic realm="example""#.to_string(),
value_env_var: None,
}, },
HttpHeader { HttpHeader {
name: "WWW-Authenticate".to_string(), name: "WWW-Authenticate".to_string(),
value: r#"Bearer error="insufficient_scope", scope="files:read""#.to_string(), value: r#"Bearer error="insufficient_scope", scope="files:read""#.to_string(),
value_env_var: None,
}, },
]; ];
@@ -25,6 +25,7 @@ fn protocol_headers_preserve_utf8_values() {
vec![HttpHeader { vec![HttpHeader {
name: "x-plugin-name".to_string(), name: "x-plugin-name".to_string(),
value: "café".to_string(), value: "café".to_string(),
value_env_var: None,
}] }]
); );
} }
@@ -166,6 +166,7 @@ impl SameOriginRedirectHttpClient {
params.headers.push(HttpHeader { params.headers.push(HttpHeader {
name: "referer".to_string(), name: "referer".to_string(),
value: referer.to_string(), value: referer.to_string(),
value_env_var: None,
}); });
} }
@@ -38,6 +38,7 @@ fn headers<const N: usize>(headers: [(&str, &str); N]) -> Vec<HttpHeader> {
.map(|(name, value)| HttpHeader { .map(|(name, value)| HttpHeader {
name: name.to_string(), name: name.to_string(),
value: value.to_string(), value: value.to_string(),
value_env_var: None,
}) })
.collect() .collect()
} }
+1
View File
@@ -185,6 +185,7 @@ impl HttpHeadersClient {
value: std::str::from_utf8(value.as_bytes()) value: std::str::from_utf8(value.as_bytes())
.map_err(|error| ExecServerError::HttpRequest(error.to_string()))? .map_err(|error| ExecServerError::HttpRequest(error.to_string()))?
.to_string(), .to_string(),
value_env_var: None,
}); });
} }
if let Some(deadline) = deadline { if let Some(deadline) = deadline {
+1
View File
@@ -52,6 +52,7 @@ pub use rmcp_client::ElicitationResponse;
pub use rmcp_client::ListToolsWithConnectorIdResult; pub use rmcp_client::ListToolsWithConnectorIdResult;
pub use rmcp_client::RmcpClient; pub use rmcp_client::RmcpClient;
pub use rmcp_client::SendElicitation; pub use rmcp_client::SendElicitation;
pub use rmcp_client::StreamableHttpBearerToken;
pub use rmcp_client::ToolWithConnectorId; pub use rmcp_client::ToolWithConnectorId;
pub use startup_error::is_authentication_required_error; pub use startup_error::is_authentication_required_error;
pub use stdio_server_launcher::ExecutorStdioServerLauncher; pub use stdio_server_launcher::ExecutorStdioServerLauncher;
@@ -169,6 +169,7 @@ impl OAuthHttpClientAdapter {
Ok(HttpHeader { Ok(HttpHeader {
name: name.as_str().to_string(), name: name.as_str().to_string(),
value: value.to_str().map_err(oauth_http_client_error)?.to_string(), value: value.to_str().map_err(oauth_http_client_error)?.to_string(),
value_env_var: None,
}) })
}) })
.collect::<Result<Vec<_>, OAuthHttpClientError>>()?; .collect::<Result<Vec<_>, OAuthHttpClientError>>()?;
+18 -6
View File
@@ -130,6 +130,15 @@ enum ClientState {
Closed, 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)] #[derive(Clone)]
enum TransportRecipe { enum TransportRecipe {
InProcess { InProcess {
@@ -142,7 +151,7 @@ enum TransportRecipe {
StreamableHttp { StreamableHttp {
server_name: String, server_name: String,
url: String, url: String,
bearer_token: Option<String>, bearer_token: Option<StreamableHttpBearerToken>,
http_headers: Option<HashMap<String, String>>, http_headers: Option<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>, env_http_headers: Option<HashMap<String, String>>,
store_mode: OAuthCredentialsStoreMode, store_mode: OAuthCredentialsStoreMode,
@@ -514,7 +523,7 @@ impl RmcpClient {
Self::new_streamable_http_client_with_protocol_mode_and_redirect_mode( Self::new_streamable_http_client_with_protocol_mode_and_redirect_mode(
server_name, server_name,
url, url,
bearer_token, bearer_token.map(StreamableHttpBearerToken::Resolved),
http_headers, http_headers,
env_http_headers, env_http_headers,
store_mode, store_mode,
@@ -531,7 +540,7 @@ impl RmcpClient {
pub async fn new_streamable_http_client_with_protocol_mode_and_redirect_mode( pub async fn new_streamable_http_client_with_protocol_mode_and_redirect_mode(
server_name: &str, server_name: &str,
url: &str, url: &str,
bearer_token: Option<String>, bearer_token: Option<StreamableHttpBearerToken>,
http_headers: Option<HashMap<String, String>>, http_headers: Option<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>, env_http_headers: Option<HashMap<String, String>>,
store_mode: OAuthCredentialsStoreMode, store_mode: OAuthCredentialsStoreMode,
@@ -1012,7 +1021,10 @@ impl RmcpClient {
redirect_mode, redirect_mode,
initialize_deadline, initialize_deadline,
} => { } => {
let has_configured_headers = http_headers let has_configured_headers = matches!(
bearer_token,
Some(StreamableHttpBearerToken::ProvidedByHttpClient)
) || http_headers
.as_ref() .as_ref()
.is_some_and(|headers| !headers.is_empty()) .is_some_and(|headers| !headers.is_empty())
|| env_http_headers || env_http_headers
@@ -1133,8 +1145,8 @@ impl RmcpClient {
} else { } else {
let mut http_config = let mut http_config =
StreamableHttpClientTransportConfig::with_uri(url.clone()); StreamableHttpClientTransportConfig::with_uri(url.clone());
if let Some(bearer_token) = bearer_token.clone() { if let Some(StreamableHttpBearerToken::Resolved(bearer_token)) = bearer_token {
http_config = http_config.auth_header(bearer_token); http_config = http_config.auth_header(bearer_token.clone());
} }
let transport = StreamableHttpClientTransport::with_client( let transport = StreamableHttpClientTransport::with_client(
@@ -12,11 +12,15 @@ use codex_config::types::OAuthCredentialsStoreMode;
use codex_exec_server::Environment; use codex_exec_server::Environment;
use codex_exec_server::ExecServerError; use codex_exec_server::ExecServerError;
use codex_exec_server::HttpClient; use codex_exec_server::HttpClient;
use codex_exec_server::HttpRedirectPolicy;
use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpRequestParams;
use codex_exec_server::HttpRequestResponse; use codex_exec_server::HttpRequestResponse;
use codex_exec_server::HttpResponseBodyStream; use codex_exec_server::HttpResponseBodyStream;
use codex_rmcp_client::McpProtocolMode;
use codex_rmcp_client::RmcpClient; use codex_rmcp_client::RmcpClient;
use codex_rmcp_client::StoredOAuthTokens; use codex_rmcp_client::StoredOAuthTokens;
use codex_rmcp_client::StreamableHttpBearerToken;
use codex_rmcp_client::StreamableHttpRedirectMode;
use codex_rmcp_client::WrappedOAuthTokenResponse; use codex_rmcp_client::WrappedOAuthTokenResponse;
use codex_rmcp_client::save_oauth_tokens; use codex_rmcp_client::save_oauth_tokens;
use codex_rmcp_client::stored_oauth_credential_snapshot; use codex_rmcp_client::stored_oauth_credential_snapshot;
@@ -48,6 +52,7 @@ const FILE_ACCESS_TOKEN: &str = "stale-file-access-token";
struct RecordingHttpClient { struct RecordingHttpClient {
inner: Arc<dyn HttpClient>, inner: Arc<dyn HttpClient>,
bearer_tokens: Arc<Mutex<Vec<String>>>, bearer_tokens: Arc<Mutex<Vec<String>>>,
redirect_policies: Arc<Mutex<Vec<HttpRedirectPolicy>>>,
} }
impl RecordingHttpClient { impl RecordingHttpClient {
@@ -55,10 +60,15 @@ impl RecordingHttpClient {
Self { Self {
inner, inner,
bearer_tokens: Arc::new(Mutex::new(Vec::new())), 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 let Some(header) = params
.headers .headers
.iter() .iter()
@@ -85,7 +95,7 @@ impl HttpClient for RecordingHttpClient {
&self, &self,
params: HttpRequestParams, params: HttpRequestParams,
) -> BoxFuture<'_, Result<HttpRequestResponse, ExecServerError>> { ) -> BoxFuture<'_, Result<HttpRequestResponse, ExecServerError>> {
self.record_bearer_token(&params); self.record_request(&params);
self.inner.http_request(params) self.inner.http_request(params)
} }
@@ -93,7 +103,7 @@ impl HttpClient for RecordingHttpClient {
&self, &self,
params: HttpRequestParams, params: HttpRequestParams,
) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> { ) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> {
self.record_bearer_token(&params); self.record_request(&params);
self.inner.http_request_stream(params) 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::<String>::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)] #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn auto_store_remains_pinned_across_session_recovery() -> anyhow::Result<()> { async fn auto_store_remains_pinned_across_session_recovery() -> anyhow::Result<()> {
let (_server, base_url) = spawn_streamable_http_server().await?; let (_server, base_url) = spawn_streamable_http_server().await?;