diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 6790fc387f..9a72d028f3 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4788,6 +4788,7 @@ dependencies = [ "tracing-test", "url", "urlencoding", + "uuid", "webbrowser", "which 8.0.0", "windows-sys 0.61.2", diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index 4702b0f53f..1edabb6ba2 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -55,6 +55,7 @@ tiny_http = { workspace = true } tokio = { workspace = true, features = [ "io-util", "macros", + "net", "process", "signal", "rt-multi-thread", @@ -90,7 +91,16 @@ keyring = { workspace = true, features = ["apple-native"] } [target.'cfg(target_os = "windows")'.dependencies] keyring = { workspace = true, features = ["windows-native"] } -windows-sys = { workspace = true, features = ["Win32_Storage_FileSystem"] } +uuid = { workspace = true, features = ["v4"] } +windows-sys = { workspace = true, features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_Security_Authorization", + "Win32_Storage_FileSystem", +] } + +[target.'cfg(target_os = "windows")'.dev-dependencies] +windows-sys = { workspace = true, features = ["Win32_System_Pipes"] } [target.'cfg(any(target_os = "freebsd", target_os = "openbsd"))'.dependencies] keyring = { workspace = true, features = ["sync-secret-service"] } diff --git a/codex-rs/rmcp-client/src/bounded_stdio_transport.rs b/codex-rs/rmcp-client/src/bounded_stdio_transport.rs index dc065995f5..3a6164d03c 100644 --- a/codex-rs/rmcp-client/src/bounded_stdio_transport.rs +++ b/codex-rs/rmcp-client/src/bounded_stdio_transport.rs @@ -14,24 +14,25 @@ use tokio::io::AsyncRead; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; use tokio::io::BufReader; -use tokio::process::ChildStdin; -use tokio::process::ChildStdout; use tokio::sync::Mutex; use tracing::debug; use tracing::warn; +use crate::local_stdio_io::Stdin; +use crate::local_stdio_io::Stdout; + /// Match the existing executor stdio transport and the production MCP limit. pub(crate) const MAX_MCP_STDIO_LINE_BYTES: usize = 8 * 1024 * 1024; pub(super) struct BoundedStdioTransport { - stdin: Arc>>, - stdout: BufReader, + stdin: Arc>>, + stdout: BufReader, pending_line: Vec, program_name: String, } impl BoundedStdioTransport { - pub(super) fn new(stdin: ChildStdin, stdout: ChildStdout, program_name: String) -> Self { + pub(super) fn new(stdin: Stdin, stdout: Stdout, program_name: String) -> Self { Self { stdin: Arc::new(Mutex::new(Some(stdin))), stdout: BufReader::new(stdout), @@ -43,7 +44,7 @@ impl BoundedStdioTransport { /// Pass both a partial line and buffered read-ahead to the SDK's legacy codec. pub(super) fn into_legacy_pipes( self, - ) -> io::Result<(Box, ChildStdin)> { + ) -> io::Result<(Box, Stdin)> { let stdin = self .stdin .try_lock() diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index 50a4f531d6..7e459657da 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -15,6 +15,7 @@ mod http_client_adapter; mod http_client_redirect; mod http_headers; mod in_process_transport; +mod local_stdio_io; mod local_stdio_transport; mod logging_client_handler; mod oauth; diff --git a/codex-rs/rmcp-client/src/local_stdio_io/mod.rs b/codex-rs/rmcp-client/src/local_stdio_io/mod.rs new file mode 100644 index 0000000000..481e6ca47e --- /dev/null +++ b/codex-rs/rmcp-client/src/local_stdio_io/mod.rs @@ -0,0 +1,63 @@ +//! Platform-specific pipe ownership beneath the shared MCP framing layer. + +use std::io; + +use codex_utils_pty::Child; +use codex_utils_pty::Command; + +#[cfg(windows)] +mod windows; + +#[cfg(windows)] +pub(super) type Stdin = tokio::net::windows::named_pipe::NamedPipeServer; +#[cfg(windows)] +pub(super) type Stdout = tokio::net::windows::named_pipe::NamedPipeServer; +#[cfg(windows)] +pub(super) type Stderr = tokio::net::windows::named_pipe::NamedPipeServer; + +#[cfg(not(windows))] +pub(super) type Stdin = tokio::process::ChildStdin; +#[cfg(not(windows))] +pub(super) type Stdout = tokio::process::ChildStdout; +#[cfg(not(windows))] +pub(super) type Stderr = tokio::process::ChildStderr; + +pub(super) struct LocalStdio { + pub(super) child: Child, + pub(super) stdin: Stdin, + pub(super) stdout: Stdout, + pub(super) stderr: Option, +} + +pub(super) fn spawn(command: Command) -> io::Result { + #[cfg(windows)] + { + let mut command = command; + let pipes = windows::Stdio::configure(&mut command)?; + Ok(LocalStdio { + child: command.spawn()?, + stdin: pipes.stdin, + stdout: pipes.stdout, + stderr: Some(pipes.stderr), + }) + } + #[cfg(not(windows))] + { + let mut child = command.spawn()?; + let stdin = child + .stdin + .take() + .ok_or_else(|| io::Error::other("MCP server stdin was not piped"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| io::Error::other("MCP server stdout was not piped"))?; + let stderr = child.stderr.take(); + Ok(LocalStdio { + child, + stdin, + stdout, + stderr, + }) + } +} diff --git a/codex-rs/rmcp-client/src/local_stdio_io/windows.rs b/codex-rs/rmcp-client/src/local_stdio_io/windows.rs new file mode 100644 index 0000000000..86dd57a9a1 --- /dev/null +++ b/codex-rs/rmcp-client/src/local_stdio_io/windows.rs @@ -0,0 +1,137 @@ +//! IOCP-backed MCP pipes with private, preconnected child endpoints. + +use std::fs::File; +use std::fs::OpenOptions; +use std::io; +use std::os::windows::fs::OpenOptionsExt; +use std::ptr; + +use codex_utils_pty::ChildStdin; +use codex_utils_pty::Command; +use tokio::net::windows::named_pipe::NamedPipeServer; +use tokio::net::windows::named_pipe::PipeMode; +use tokio::net::windows::named_pipe::ServerOptions; +use uuid::Uuid; +use windows_sys::Win32::Foundation::GENERIC_WRITE; +use windows_sys::Win32::Foundation::LocalFree; +use windows_sys::Win32::Security::Authorization::ConvertStringSecurityDescriptorToSecurityDescriptorW; +use windows_sys::Win32::Security::Authorization::SDDL_REVISION_1; +use windows_sys::Win32::Security::PSECURITY_DESCRIPTOR; +use windows_sys::Win32::Security::SECURITY_ATTRIBUTES; +use windows_sys::Win32::Storage::FileSystem::FILE_READ_ATTRIBUTES; +use windows_sys::Win32::Storage::FileSystem::SECURITY_IDENTIFICATION; +use windows_sys::Win32::Storage::FileSystem::SECURITY_SQOS_PRESENT; + +pub(super) struct Stdio { + pub(super) stdin: NamedPipeServer, + pub(super) stdout: NamedPipeServer, + pub(super) stderr: NamedPipeServer, +} + +impl Stdio { + pub(super) fn configure(command: &mut Command) -> io::Result { + let security = PipeSecurity::new()?; + let (stdin, child_stdin) = connected_pipe(Direction::ChildReads, &security)?; + let (stdout, child_stdout) = connected_pipe(Direction::ChildWrites, &security)?; + let (stderr, child_stderr) = connected_pipe(Direction::ChildWrites, &security)?; + command + .stdin(ChildStdin::File(child_stdin.into())) + .stdout_handle(child_stdout.into()) + .stderr_handle(child_stderr.into()); + Ok(Self { + stdin, + stdout, + stderr, + }) + } +} + +enum Direction { + ChildReads, + ChildWrites, +} + +fn connected_pipe( + direction: Direction, + security: &PipeSecurity, +) -> io::Result<(NamedPipeServer, File)> { + let name = format!(r"\\.\pipe\codex-mcp-{}", Uuid::new_v4().simple()); + let mut options = ServerOptions::new(); + options + .pipe_mode(PipeMode::Byte) + .first_pipe_instance(true) + .max_instances(1) + .reject_remote_clients(true); + let mut client = OpenOptions::new(); + client.security_qos_flags(SECURITY_IDENTIFICATION | SECURITY_SQOS_PRESENT); + match direction { + Direction::ChildReads => { + options.access_inbound(false).access_outbound(true); + client.read(true); + } + Direction::ChildWrites => { + options.access_inbound(true).access_outbound(false); + // Helpers can query the blocking mode of their stdout and stderr pipes. + client.access_mode(GENERIC_WRITE | FILE_READ_ATTRIBUTES); + } + } + let mut attributes = SECURITY_ATTRIBUTES { + nLength: u32::try_from(size_of::()).map_err(io::Error::other)?, + lpSecurityDescriptor: security.0, + bInheritHandle: 0, + }; + // SAFETY: attributes and its owned descriptor stay live through CreateNamedPipeW. + // The kernel copies the descriptor; the returned pipe owns its separate handle. + let server = unsafe { + options.create_with_security_attributes_raw(&name, ptr::from_mut(&mut attributes).cast()) + }?; + let child = client.open(&name)?; + // The child endpoint is already connected. Complete registration on the + // launcher's blocking thread before the child is allowed to execute. + tokio::runtime::Handle::current().block_on(server.connect())?; + Ok((server, child)) +} + +struct PipeSecurity(PSECURITY_DESCRIPTOR); + +impl PipeSecurity { + fn new() -> io::Result { + // Only the pipe owner and SYSTEM can open the single local instance. + let sddl: Vec = "D:P(A;;GA;;;SY)(A;;GA;;;OW)" + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + let mut descriptor = ptr::null_mut(); + // SAFETY: sddl is NUL-terminated and descriptor is a valid writable out + // pointer. On success Windows transfers a LocalFree-owned allocation. + if unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl.as_ptr(), + SDDL_REVISION_1, + &mut descriptor, + ptr::null_mut(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(Self(descriptor)) + } +} + +impl Drop for PipeSecurity { + fn drop(&mut self) { + // SAFETY: this is the sole owner of the successful SDDL conversion's + // allocation; no pipe creation call retains a pointer to it. + if !unsafe { LocalFree(self.0) }.is_null() { + tracing::warn!( + "Failed to release MCP pipe security descriptor: {}", + io::Error::last_os_error() + ); + } + } +} + +#[cfg(test)] +#[path = "windows_tests.rs"] +mod tests; diff --git a/codex-rs/rmcp-client/src/local_stdio_io/windows_tests.rs b/codex-rs/rmcp-client/src/local_stdio_io/windows_tests.rs new file mode 100644 index 0000000000..2a78e1234b --- /dev/null +++ b/codex-rs/rmcp-client/src/local_stdio_io/windows_tests.rs @@ -0,0 +1,37 @@ +//! Checks the access rights of the synchronous pipe endpoint inherited by a helper. + +#![cfg(test)] + +use std::io; +use std::os::windows::io::AsRawHandle; +use std::ptr; + +use windows_sys::Win32::System::Pipes::GetNamedPipeHandleStateW; + +use super::Direction; +use super::PipeSecurity; +use super::connected_pipe; + +#[test] +fn child_can_read_output_pipe_mode() -> io::Result<()> { + let runtime = tokio::runtime::Runtime::new()?; + let _guard = runtime.enter(); + let (_server, child) = connected_pipe(Direction::ChildWrites, &PipeSecurity::new()?)?; + let mut mode = 0; + // SAFETY: the file owns a live pipe handle and mode is writable for the call. + if unsafe { + GetNamedPipeHandleStateW( + child.as_raw_handle(), + &mut mode, + ptr::null_mut(), + ptr::null_mut(), + ptr::null_mut(), + ptr::null_mut(), + /*nmaxusernamesize*/ 0, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(()) +} diff --git a/codex-rs/rmcp-client/src/local_stdio_transport.rs b/codex-rs/rmcp-client/src/local_stdio_transport.rs index 2e908278dc..4b01c3bd3d 100644 --- a/codex-rs/rmcp-client/src/local_stdio_transport.rs +++ b/codex-rs/rmcp-client/src/local_stdio_transport.rs @@ -18,11 +18,12 @@ use rmcp::service::TxJsonRpcMessage; use rmcp::transport::Transport; use rmcp::transport::async_rw::AsyncRwTransport; use tokio::io::AsyncRead; -use tokio::process::ChildStderr; -use tokio::process::ChildStdin; use tokio::sync::oneshot; use crate::bounded_stdio_transport::BoundedStdioTransport; +use crate::local_stdio_io::LocalStdio; +use crate::local_stdio_io::Stderr; +use crate::local_stdio_io::Stdin; use crate::protocol_mode::McpProtocolMode; use codex_utils_pty::Child; @@ -138,7 +139,7 @@ impl Discovery { enum StdioTransport { /// Preserve rmcp's existing framing for servers using the initialize handshake. - Legacy(AsyncRwTransport, ChildStdin>), + Legacy(AsyncRwTransport, Stdin>), /// Bound frames and skip messages unknown to the client during 2026-07-28 discovery. Modern(BoundedStdioTransport), Closed, @@ -146,19 +147,16 @@ enum StdioTransport { impl LocalStdioTransport { pub(super) fn new( - mut child: Child, + stdio: LocalStdio, program_name: String, protocol_mode: McpProtocolMode, - ) -> io::Result<(Self, Option)> { - let stdin = child - .stdin - .take() - .ok_or_else(|| io::Error::other("MCP server stdin was not piped"))?; - let stdout = child - .stdout - .take() - .ok_or_else(|| io::Error::other("MCP server stdout was not piped"))?; - let stderr = child.stderr.take(); + ) -> (Self, Option) { + let LocalStdio { + child, + stdin, + stdout, + stderr, + } = stdio; let transport = match protocol_mode { McpProtocolMode::Legacy => { StdioTransport::Legacy(AsyncRwTransport::new(Box::new(stdout), stdin)) @@ -167,14 +165,14 @@ impl LocalStdioTransport { StdioTransport::Modern(BoundedStdioTransport::new(stdin, stdout, program_name)) } }; - Ok(( + ( Self { child: Some(Box::new(child)), transport, discovery: None, }, stderr, - )) + ) } pub(super) fn watch_discovery( diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index 4925929571..0a089db96b 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -33,6 +33,7 @@ use std::thread::sleep; use std::thread::spawn; use std::time::Duration; +use crate::local_stdio_io; use crate::local_stdio_transport::DiscoveryOutcome; use anyhow::Result; use anyhow::anyhow; @@ -503,26 +504,27 @@ impl LocalStdioServerLauncher { "MCP client is shut down", )); } - let child = command.spawn()?; + let stdio = local_stdio_io::spawn(command)?; #[cfg(windows)] - let (child, job) = match job { - Some(job) => match child + let (stdio, job) = match job { + Some(job) => match stdio + .child .id() .ok_or_else(|| io::Error::other("missing suspended MCP server process id")) .and_then(|process_id| job.assign_and_resume_process(process_id)) { - Ok(Some(root)) => (child, Some((job, root))), - Ok(None) => (child, None), + Ok(Some(root)) => (stdio, Some((job, root))), + Ok(None) => (stdio, None), Err(error) => { warn!( "Windows MCP process job containment failed; retrying without it: {error}" ); - drop(child); + drop(stdio); drop(job); - (build_command().spawn()?, None) + (local_stdio_io::spawn(build_command())?, None) } }, - None => (child, None), + None => (stdio, None), }; #[cfg(windows)] let protocol_mode = if job.is_none() && protocol_mode == McpProtocolMode::DiscoverLocalStdio @@ -532,13 +534,13 @@ impl LocalStdioServerLauncher { } else { protocol_mode }; - let process_id = child.id(); + let process_id = stdio.child.id(); // A legacy replacement still belongs to the discovery startup that launched it. let discovery_owned = (protocol_mode == McpProtocolMode::DiscoverLocalStdio || replacement_slot.is_some()) && process_id.is_some(); let (transport, stderr) = - LocalStdioTransport::new(child, program_name.clone(), protocol_mode)?; + LocalStdioTransport::new(stdio, program_name.clone(), protocol_mode); #[cfg(windows)] let terminator = match job { Some((job, root)) => Some(LocalProcessTerminator::Job { @@ -1036,6 +1038,44 @@ mod tests { use codex_protocol::shell_environment; use pretty_assertions::assert_eq; + #[cfg(windows)] + #[test] + fn idle_local_mcp_pipe_reads_leave_blocking_io_available() -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .max_blocking_threads(1) + .enable_all() + .build()?; + runtime.block_on(async { + let cwd = std::env::current_dir()?; + let launcher = LocalStdioServerLauncher::new(cwd.clone()); + for mode in [McpProtocolMode::Legacy, McpProtocolMode::V20260728] { + // set /p waits for stdin and leaves both output pipes open and quiet. + let command = StdioServerCommand::new( + "cmd.exe".into(), + ["/d", "/q", "/c", "set /p idle="] + .map(OsString::from) + .into(), + None, + Vec::new(), + None, + mode, + ); + let mut transport = launcher.launch(command).await?; + // Poll the stdout read before scheduling the filesystem operation. The + // launcher also drains stderr. A blocking read can occupy the only slot. + let pending = transport.receive().now_or_never().is_none(); + let metadata = + tokio::time::timeout(Duration::from_secs(5), tokio::fs::metadata(&cwd)).await; + // Release the child even on the failing, pre-fix path. + transport.close().await?; + anyhow::ensure!(pending, "helper exited before its idle read"); + metadata??; + } + Ok(()) + }) + } + #[test] fn remote_env_policy_uses_core_env_without_remote_source_vars() { let policy = ExecutorStdioServerLauncher::remote_env_policy(&[], &HashMap::new()); diff --git a/codex-rs/utils/pty/src/child_command.rs b/codex-rs/utils/pty/src/child_command.rs index 452e3ecdde..db1602ee32 100644 --- a/codex-rs/utils/pty/src/child_command.rs +++ b/codex-rs/utils/pty/src/child_command.rs @@ -67,6 +67,8 @@ pub enum ChildStdin { Null, #[cfg(unix)] File(std::os::fd::OwnedFd), + #[cfg(windows)] + File(std::os::windows::io::OwnedHandle), } /// A local command whose complete launch contract is known to both backends. @@ -186,6 +188,20 @@ impl Command { self } + /// Supply the child endpoint of a Windows output pipe. + #[cfg(windows)] + pub fn stdout_handle(&mut self, handle: std::os::windows::io::OwnedHandle) -> &mut Self { + self.inner.stdout(TokioStdio::from(handle)); + self + } + + /// Supply the child endpoint of a Windows diagnostic pipe. + #[cfg(windows)] + pub fn stderr_handle(&mut self, handle: std::os::windows::io::OwnedHandle) -> &mut Self { + self.inner.stderr(TokioStdio::from(handle)); + self + } + pub fn descriptor_policy(&mut self, policy: DescriptorPolicy) -> &mut Self { self.descriptor_policy = policy; self @@ -289,6 +305,8 @@ impl Command { ChildStdin::Null => TokioStdio::null(), #[cfg(unix)] ChildStdin::File(fd) => TokioStdio::from(fd), + #[cfg(windows)] + ChildStdin::File(handle) => TokioStdio::from(handle), }); #[cfg(unix)] {