Extract HTTP headers helper execution into a shared crate (#61018)

## What changed

Move shell command construction, bounded process execution, and JSON header parsing from `rmcp-client` into `codex-http-headers-helper`. Preserve the 10-second timeout, 64 KiB output limit, process-tree cleanup, and rejection of invalid or duplicate headers. Keep MCP environment setup and reserved-header validation in `rmcp-client`.

Mark helper-provided header values as sensitive so debug formatting redacts credentials.

## Testing

Add parser tests for credential-safe errors and debug redaction, and adapt the MCP reserved-header test to exercise helper execution.

GitOrigin-RevId: e32d96261ff3648722014971bbf3803d37176ab4
This commit is contained in:
Abhinav
2026-11-12 04:32:34 +00:00
committed by copyberry
parent 64b8d019c8
commit 88c5856258
9 changed files with 259 additions and 192 deletions
+14
View File
@@ -3967,6 +3967,19 @@ dependencies = [
"zstd",
]
[[package]]
name = "codex-http-headers-helper"
version = "0.0.0"
dependencies = [
"anyhow",
"codex-utils-process",
"codex-utils-pty",
"http 1.4.0",
"serde",
"serde_json",
"tokio",
]
[[package]]
name = "codex-image-generation-extension"
version = "0.0.0"
@@ -4793,6 +4806,7 @@ dependencies = [
"codex-config",
"codex-exec-server",
"codex-http-client",
"codex-http-headers-helper",
"codex-keyring-store",
"codex-network-proxy",
"codex-otel",
+2
View File
@@ -55,6 +55,7 @@ members = [
"hooks",
"history",
"http-client",
"http-headers-helper",
"secrets",
"exec",
"file-system",
@@ -216,6 +217,7 @@ codex-code-mode-runtime = { path = "code-mode-runtime" }
codex-plugin-host = { path = "plugin-host" }
codex-home = { path = "codex-home" }
codex-http-client = { path = "http-client" }
codex-http-headers-helper = { path = "http-headers-helper" }
codex-websocket-auth = { path = "websocket-auth" }
codex-websocket-client = { path = "websocket-client" }
codex-config = { path = "config" }
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "http-headers-helper",
crate_name = "codex_http_headers_helper",
)
+20
View File
@@ -0,0 +1,20 @@
[package]
name = "codex-http-headers-helper"
version.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
anyhow = { workspace = true }
codex-utils-process = { workspace = true }
codex-utils-pty = { workspace = true }
http = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["io-util", "process", "time"] }
[lints]
workspace = true
[lib]
doctest = false
@@ -0,0 +1,22 @@
#![cfg(test)]
use super::*;
#[test]
fn invalid_output_does_not_expose_credentials() {
for output in [
r#"{"Authorization": "secret", "Other": 123}"#,
r#"{"bad name": "secret"}"#,
r#"{"Authorization": "secret\r\nInjected: value"}"#,
r#"{"Authorization": "secret", "authorization": "other"}"#,
r#"{"Authorization":"secret","Authorization":"secret"}"#,
r#"["secret"]"#,
"secret",
] {
let error = parse_headers(output.as_bytes()).expect_err("invalid headers");
assert!(!error.to_string().contains("secret"));
}
let headers = parse_headers(br#"{"Authorization":"secret"}"#).unwrap();
assert!(headers["authorization"].is_sensitive());
assert!(!format!("{headers:?}").contains("secret"));
}
+169
View File
@@ -0,0 +1,169 @@
//! Runs one local HTTP headers helper with bounded output, time and process-tree lifetime.
//! Callers choose the command environment, working directory and refresh policy.
use anyhow::Result;
use anyhow::anyhow;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
use serde::de::Deserializer as _;
use serde::de::MapAccess;
use serde::de::Visitor;
use std::fmt;
#[cfg(windows)]
use std::os::windows::process::CommandExt;
use std::process::Command;
use std::process::Stdio;
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio::process::Child;
const HELPER_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_HELPER_OUTPUT_BYTES: usize = 64 * 1024;
struct HelperProcess {
child: Child,
#[cfg(unix)]
process_group_id: u32,
#[cfg(windows)]
job: codex_utils_pty::JobObject,
}
impl Drop for HelperProcess {
fn drop(&mut self) {
#[cfg(unix)]
let _ = codex_utils_pty::process_group::kill_process_group(self.process_group_id);
#[cfg(windows)]
let _ = self.job.terminate();
let _ = self.child.start_kill();
}
}
/// Builds a background shell command using the platform's command interpreter.
pub fn shell_command(command: &str) -> Command {
#[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 = codex_utils_process::background_command(shell);
#[cfg(windows)]
{
process.args(["/Q", "/D", "/C"]);
process.raw_arg(format!(r#""{command}""#));
}
#[cfg(not(windows))]
process.args(["-c", command]);
process
}
/// Executes a prepared command and validates its JSON headers without exposing output in errors.
pub async fn run(command: Command) -> Result<HeaderMap> {
let mut process = tokio::process::Command::from(command);
#[cfg(unix)]
process.process_group(0);
process
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.kill_on_drop(true);
#[cfg(windows)]
let (child, job) = {
let job = codex_utils_pty::JobObject::create_without_breakaway()
.map_err(|error| anyhow!("headers helper containment failed: {error}"))?;
let child = job
.spawn_contained(&mut process)
.map_err(|error| anyhow!("headers helper failed to start: {error}"))?;
(child, job)
};
#[cfg(not(windows))]
let child = process
.spawn()
.map_err(|error| anyhow!("headers helper failed to start: {error}"))?;
let mut process = HelperProcess {
#[cfg(unix)]
process_group_id: child
.id()
.ok_or_else(|| anyhow!("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!("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!("headers helper output exceeds 64 KiB"));
}
let status = process.child.wait().await?;
if !status.success() {
return Err(anyhow!("headers helper exited with status {status}"));
}
Ok(output)
})
.await
.map_err(|_| anyhow!("headers helper timed out after 10 seconds"))??;
parse_headers(&output)
}
// Deserialize entries in wire order. Deserializing into a map would silently
// overwrite exact duplicates before HTTP's case-insensitive name check.
struct HeaderEntries;
impl<'de> Visitor<'de> for HeaderEntries {
type Value = Vec<(String, String)>;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("a JSON object of string header names and values")
}
fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
let mut entries = Vec::new();
while let Some(entry) = map.next_entry()? {
entries.push(entry);
}
Ok(entries)
}
}
fn parse_headers(output: &[u8]) -> Result<HeaderMap> {
let text =
std::str::from_utf8(output).map_err(|_| anyhow!("headers helper wrote non-UTF-8 data"))?;
let mut json = serde_json::Deserializer::from_str(text.trim());
let entries = json
.deserialize_map(HeaderEntries)
.and_then(|entries| {
json.end()?;
Ok(entries)
})
.map_err(|_| anyhow!("headers helper must output a JSON object of strings"))?;
let mut parsed = HeaderMap::new();
for (name, value) in entries {
let name = HeaderName::from_bytes(name.as_bytes())
.map_err(|_| anyhow!("headers helper returned an invalid header name"))?;
let mut value = HeaderValue::from_str(&value)
.map_err(|_| anyhow!("headers helper returned an invalid header value"))?;
value.set_sensitive(true);
if parsed.insert(name, value).is_some() {
return Err(anyhow!("headers helper returned duplicate header names"));
}
}
Ok(parsed)
}
#[cfg(test)]
#[path = "headers_helper_tests.rs"]
mod tests;
+1
View File
@@ -19,6 +19,7 @@ codex-api = { workspace = true }
codex-config = { workspace = true }
codex-exec-server = { workspace = true }
codex-http-client = { workspace = true }
codex-http-headers-helper = { workspace = true }
codex-keyring-store = { workspace = true }
codex-network-proxy = { workspace = true }
codex-otel = { workspace = true }
+7 -182
View File
@@ -1,10 +1,5 @@
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::sync::Mutex;
use std::sync::PoisonError;
@@ -19,20 +14,10 @@ use codex_exec_server::HttpRedirectPolicy;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::HttpRequestResponse;
use codex_exec_server::HttpResponseBodyStream;
#[cfg(unix)]
use codex_utils_pty::process_group::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;
@@ -40,8 +25,6 @@ use url::Url;
use crate::utils::create_env_for_mcp_server;
use crate::www_authenticate::insufficient_scope_challenge;
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 {
@@ -69,68 +52,6 @@ struct HttpHeadersClient {
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();
@@ -429,103 +350,15 @@ impl HttpClient for HttpHeadersClient {
}
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);
let mut process = codex_http_headers_helper::shell_command(command);
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"))?;
.envs(create_env_for_mcp_server(/*extra_env*/ None, &[])?);
let headers = codex_http_headers_helper::run(process)
.await
.map_err(|err| anyhow!("MCP HTTP {err}"))?;
for name in headers.keys() {
// Helper values replace same-name configured headers, except explicit 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.
@@ -553,16 +386,8 @@ fn parse_helper_output(stdout: Vec<u8>) -> Result<HeaderMap> {
"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)
Ok(headers)
}
#[cfg(test)]
+18 -10
View File
@@ -1,16 +1,24 @@
use super::*;
#[cfg(unix)]
use http::HeaderValue;
use pretty_assertions::assert_eq;
#[test]
fn helper_output_errors_do_not_echo_secrets() {
for output in [
br#"{"Host":"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"));
}
#[tokio::test]
async fn reserved_header_errors_do_not_echo_secrets() {
let dir = tempfile::tempdir().expect("helper directory");
std::fs::write(dir.path().join("headers.json"), r#"{"Host":"secret"}"#).expect("write headers");
let command = if cfg!(windows) {
"type headers.json"
} else {
"cat headers.json"
};
let error = run_helper(command, dir.path())
.await
.expect_err("reserved header");
assert_eq!(
error.to_string(),
"MCP HTTP headers helper returned a reserved header"
);
}
#[cfg(unix)]