From 71c839b42d795ac3d4a3e38f74a252ce18845e20 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 17 Feb 2026 12:08:36 -0800 Subject: [PATCH] feat(zsh-sidecar): add standalone sidecar crate and protocol tests --- codex-rs/Cargo.lock | 16 + codex-rs/Cargo.toml | 1 + codex-rs/zsh-sidecar/Cargo.toml | 24 + codex-rs/zsh-sidecar/src/main.rs | 994 +++++++++++++++++++++++++ codex-rs/zsh-sidecar/tests/protocol.rs | 952 +++++++++++++++++++++++ 5 files changed, 1987 insertions(+) create mode 100644 codex-rs/zsh-sidecar/Cargo.toml create mode 100644 codex-rs/zsh-sidecar/src/main.rs create mode 100644 codex-rs/zsh-sidecar/tests/protocol.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 3456f9d731..9634173395 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2536,6 +2536,22 @@ dependencies = [ "winres", ] +[[package]] +name = "codex-zsh-sidecar" +version = "0.0.0" +dependencies = [ + "anyhow", + "base64 0.22.1", + "clap", + "serde", + "serde_json", + "shlex", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "color-eyre" version = "0.6.5" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 75b3d18d3e..431ecc46c9 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -17,6 +17,7 @@ members = [ "cli", "config", "shell-command", + "zsh-sidecar", "core", "hooks", "secrets", diff --git a/codex-rs/zsh-sidecar/Cargo.toml b/codex-rs/zsh-sidecar/Cargo.toml new file mode 100644 index 0000000000..7e7e20d598 --- /dev/null +++ b/codex-rs/zsh-sidecar/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "codex-zsh-sidecar" +version.workspace = true +edition.workspace = true +license.workspace = true + +[[bin]] +name = "codex-zsh-sidecar" +path = "src/main.rs" + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } +base64 = { workspace = true } +clap = { workspace = true, features = ["derive"] } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +shlex = { workspace = true } +tokio = { workspace = true, features = ["io-std", "io-util", "macros", "net", "process", "rt-multi-thread", "sync", "time"] } +tracing = { workspace = true } +tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] } +uuid = { workspace = true, features = ["v4"] } diff --git a/codex-rs/zsh-sidecar/src/main.rs b/codex-rs/zsh-sidecar/src/main.rs new file mode 100644 index 0000000000..8e8fd455af --- /dev/null +++ b/codex-rs/zsh-sidecar/src/main.rs @@ -0,0 +1,994 @@ +use anyhow::Context; +use anyhow::Result; +use base64::Engine as _; +use clap::Parser; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value as JsonValue; +use std::collections::HashMap; +use std::path::PathBuf; +use std::process::Stdio; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncReadExt as _; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +#[cfg(unix)] +use tokio::net::UnixListener; +#[cfg(unix)] +use tokio::net::UnixStream; +use tokio::process::Command; +use tokio::sync::mpsc; +use tokio::task::JoinSet; +use uuid::Uuid; + +const JSONRPC_VERSION: &str = "2.0"; +const METHOD_ZSH_INITIALIZE: &str = "zsh/initialize"; +const METHOD_ZSH_EXEC_START: &str = "zsh/execStart"; +const METHOD_ZSH_EXEC_STDIN: &str = "zsh/execStdin"; +const METHOD_ZSH_EXEC_RESIZE: &str = "zsh/execResize"; +const METHOD_ZSH_EXEC_INTERRUPT: &str = "zsh/execInterrupt"; +const METHOD_ZSH_SHUTDOWN: &str = "zsh/shutdown"; +const METHOD_ZSH_REQUEST_APPROVAL: &str = "zsh/requestApproval"; +const METHOD_ZSH_EVENT_EXEC_STARTED: &str = "zsh/event/execStarted"; +const METHOD_ZSH_EVENT_EXEC_STDOUT: &str = "zsh/event/execStdout"; +const METHOD_ZSH_EVENT_EXEC_STDERR: &str = "zsh/event/execStderr"; +const METHOD_ZSH_EVENT_EXEC_EXITED: &str = "zsh/event/execExited"; +const EXEC_WRAPPER_ENV_VAR: &str = "EXEC_WRAPPER"; +const SIDECAREXEC_WRAPPER_MODE_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_MODE"; +const SIDECAREXEC_WRAPPER_SOCKET_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_SOCKET"; +const SIDECAREXEC_WRAPPER_ORIGIN_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_ORIGIN"; + +#[derive(Parser, Debug)] +struct Args { + #[arg(long)] + zsh_path: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(untagged)] +enum JsonRpcId { + Number(i64), + String(String), +} + +#[derive(Debug, Clone, Serialize)] +struct JsonRpcRequest { + jsonrpc: &'static str, + id: JsonRpcId, + method: &'static str, + params: T, +} + +#[derive(Debug, Clone, Serialize)] +struct JsonRpcSuccess { + jsonrpc: &'static str, + id: JsonRpcId, + result: T, +} + +#[derive(Debug, Clone, Serialize)] +struct JsonRpcErrorResponse { + jsonrpc: &'static str, + id: JsonRpcId, + error: JsonRpcError, +} + +#[derive(Debug, Clone, Serialize)] +struct JsonRpcError { + code: i64, + message: String, +} + +#[derive(Debug, Clone, Serialize)] +struct JsonRpcNotification { + jsonrpc: &'static str, + method: &'static str, + params: T, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +/// Host -> sidecar parameters for `zsh/execStart`. +/// +/// This defines the top-level command invocation the sidecar should run and +/// the execution environment to apply before zsh begins intercepting nested +/// `execve` calls via wrapper mode. +struct ExecStartParams { + exec_id: String, + command: Vec, + cwd: String, + #[serde(default)] + env: Option>, + #[serde(default)] + approval_reason: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ExecInterruptParams { + exec_id: String, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +enum ApprovalDecision { + Approved, + ApprovedForSession, + ApprovedExecpolicyAmendment, + Denied, + Abort, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct RequestApprovalParams { + approval_id: String, + exec_id: String, + command: Vec, + cwd: String, + reason: Option, + proposed_execpolicy_amendment: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct RequestApprovalResult { + decision: ApprovalDecision, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ExecPolicyAmendmentProposal { + command_prefix: Vec, + rationale: Option, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct InitializeResult { + protocol_version: u32, + capabilities: Capabilities, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct Capabilities { + interactive_pty: bool, +} + +#[derive(Debug, Serialize)] +struct EmptyResult {} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ExecStartedEvent { + exec_id: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ExecChunkEvent { + exec_id: String, + chunk_base64: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct ExecExitedEvent { + exec_id: String, + exit_code: i32, + signal: Option, + timed_out: Option, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +/// Wrapper-mode request sent over the Unix socket for each intercepted execve. +/// +/// Unlike `ExecStartParams`, this payload is produced by the wrapper subprocess +/// (invoked by patched zsh via `EXEC_WRAPPER`) and consumed by the sidecar to +/// decide whether host approval is required for this specific subcommand. +struct WrapperExecRequest { + file: String, + argv: Vec, + cwd: String, + origin: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +/// Execution origin metadata provided by the patched zsh fork. +/// +/// This is used to suppress approvals for shell startup/bootstrap commands +/// while still enforcing approvals for model/user command execution. +enum WrapperExecOrigin { + /// Command was spawned while evaluating login-shell startup files + /// (for example `.zprofile` / `.zlogin`) before user command execution. + LoginStartup, + /// Command was spawned while evaluating non-login rc startup files + /// (for example `.zshrc`) before user command execution. + RcStartup, + /// Command comes from the top-level user-provided command string. + /// This is the primary approval surface for shell-tool requests. + UserCommand, + /// Command is a nested child command derived from user command evaluation + /// (for example a command reached through shell operators or subshells). + Subcommand, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +/// Sidecar decision returned to wrapper mode for an intercepted execve. +/// +/// `Run` continues execution of the target program. +/// `Deny` blocks execution and returns a failure to the shell process. +struct WrapperExecResponse { + action: WrapperExecAction, + reason: Option, +} + +#[derive(Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum WrapperExecAction { + Run, + Deny, +} + +/// Runs the zsh sidecar entrypoint. +/// +/// Behavior: +/// - In normal mode, speaks JSON-RPC over stdio with the host (core). +/// - In wrapper mode (`CODEX_ZSH_SIDECAR_WRAPPER_MODE`), handles a single +/// exec-wrapper callback request and exits. +#[tokio::main] +async fn main() -> Result<()> { + if std::env::var_os(SIDECAREXEC_WRAPPER_MODE_ENV_VAR).is_some() { + return run_exec_wrapper_mode(); + } + + tracing_subscriber::fmt() + .with_env_filter("warn") + .with_writer(std::io::stderr) + .init(); + let _args = Args::parse(); + let mut stdout = tokio::io::stdout(); + + let stdin = tokio::io::stdin(); + let mut lines = BufReader::new(stdin).lines(); + + loop { + let Some(line) = lines.next_line().await.context("read stdin")? else { + break; + }; + + if line.trim().is_empty() { + continue; + } + + let value: JsonValue = match serde_json::from_str(&line) { + Ok(v) => v, + Err(err) => { + tracing::warn!("invalid JSON-RPC input: {err}"); + continue; + } + }; + + let Some(id_value) = value.get("id") else { + continue; + }; + let id: JsonRpcId = match serde_json::from_value(id_value.clone()) { + Ok(v) => v, + Err(err) => { + tracing::warn!("invalid request id: {err}"); + continue; + } + }; + + let method = value + .get("method") + .and_then(JsonValue::as_str) + .unwrap_or_default(); + + match method { + METHOD_ZSH_INITIALIZE => { + write_json_line( + &mut stdout, + &JsonRpcSuccess { + jsonrpc: JSONRPC_VERSION, + id, + result: InitializeResult { + protocol_version: 1, + capabilities: Capabilities { + interactive_pty: false, + }, + }, + }, + ) + .await?; + } + METHOD_ZSH_EXEC_START => { + let params: ExecStartParams = match parse_params(&value) { + Ok(p) => p, + Err(message) => { + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32602, + message, + }, + }, + ) + .await?; + continue; + } + }; + + if params.command.is_empty() { + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32602, + message: "execStart.command is empty".to_string(), + }, + }, + ) + .await?; + continue; + } + + let mut cmd = Command::new(¶ms.command[0]); + if params.command.len() > 1 { + cmd.args(¶ms.command[1..]); + } + cmd.current_dir(¶ms.cwd); + cmd.stdin(Stdio::null()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + cmd.kill_on_drop(true); + cmd.env_clear(); + if let Some(env) = params.env.as_ref() { + cmd.envs(env); + } + #[cfg(unix)] + let wrapper_socket_path = { + params + .env + .as_ref() + .and_then(|env| env.get(SIDECAREXEC_WRAPPER_SOCKET_ENV_VAR)) + .map(std::path::PathBuf::from) + .unwrap_or_else(|| { + let socket_id = Uuid::new_v4().as_simple().to_string(); + let temp_dir = std::env::temp_dir(); + let canonical_temp_dir = temp_dir.canonicalize().unwrap_or(temp_dir); + canonical_temp_dir.join(format!("czs-{}.sock", &socket_id[..12])) + }) + }; + #[cfg(unix)] + let listener = { + let _ = std::fs::remove_file(&wrapper_socket_path); + UnixListener::bind(&wrapper_socket_path).with_context(|| { + format!("bind wrapper socket at {}", wrapper_socket_path.display()) + })? + }; + #[cfg(unix)] + { + cmd.env( + SIDECAREXEC_WRAPPER_SOCKET_ENV_VAR, + wrapper_socket_path.to_string_lossy().to_string(), + ); + let wrapper_path = + std::env::current_exe().context("resolve current sidecar binary path")?; + cmd.env( + EXEC_WRAPPER_ENV_VAR, + wrapper_path.to_string_lossy().to_string(), + ); + } + cmd.env(SIDECAREXEC_WRAPPER_MODE_ENV_VAR, "1"); + + let mut child = match cmd.spawn() { + Ok(c) => c, + Err(err) => { + #[cfg(unix)] + { + let _ = std::fs::remove_file(&wrapper_socket_path); + } + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32000, + message: format!("failed to spawn command: {err}"), + }, + }, + ) + .await?; + continue; + } + }; + + let exec_id = params.exec_id.clone(); + + write_json_line( + &mut stdout, + &JsonRpcSuccess { + jsonrpc: JSONRPC_VERSION, + id, + result: EmptyResult {}, + }, + ) + .await?; + write_json_line( + &mut stdout, + &JsonRpcNotification { + jsonrpc: JSONRPC_VERSION, + method: METHOD_ZSH_EVENT_EXEC_STARTED, + params: ExecStartedEvent { + exec_id: exec_id.clone(), + }, + }, + ) + .await?; + + let mut tasks = JoinSet::new(); + let (stream_tx, mut stream_rx) = + mpsc::unbounded_channel::<(&'static str, Vec)>(); + if let Some(mut out) = child.stdout.take() { + let tx = stream_tx.clone(); + tasks.spawn(async move { + let mut buf = [0_u8; 8192]; + loop { + let read = match out.read(&mut buf).await { + Ok(0) => break, + Ok(n) => n, + Err(err) => { + tracing::warn!("stdout read error: {err}"); + break; + } + }; + let _ = tx.send((METHOD_ZSH_EVENT_EXEC_STDOUT, buf[..read].to_vec())); + } + }); + } + if let Some(mut err) = child.stderr.take() { + let tx = stream_tx.clone(); + tasks.spawn(async move { + let mut buf = [0_u8; 8192]; + loop { + let read = match err.read(&mut buf).await { + Ok(0) => break, + Ok(n) => n, + Err(err) => { + tracing::warn!("stderr read error: {err}"); + break; + } + }; + let _ = tx.send((METHOD_ZSH_EVENT_EXEC_STDERR, buf[..read].to_vec())); + } + }); + } + drop(stream_tx); + + let wait = child.wait(); + tokio::pin!(wait); + let mut child_exit = None; + + #[cfg(unix)] + while child_exit.is_none() || !stream_rx.is_closed() { + tokio::select! { + result = &mut wait, if child_exit.is_none() => { + child_exit = Some(result.context("wait for command exit")?); + } + stream = stream_rx.recv(), if !stream_rx.is_closed() => { + if let Some((method, chunk)) = stream { + write_json_line( + &mut stdout, + &JsonRpcNotification { + jsonrpc: JSONRPC_VERSION, + method, + params: ExecChunkEvent { + exec_id: exec_id.clone(), + chunk_base64: base64::engine::general_purpose::STANDARD.encode(chunk), + }, + } + ).await?; + } + } + accept_result = listener.accept() => { + let (stream, _) = match accept_result { + Ok(pair) => pair, + Err(err) => { + tracing::warn!("failed to accept wrapper request: {err}"); + continue; + } + }; + handle_wrapper_request( + &mut stdout, + &mut lines, + stream, + exec_id.clone(), + params.approval_reason.clone(), + ).await?; + } + } + } + #[cfg(not(unix))] + while child_exit.is_none() || !stream_rx.is_closed() { + tokio::select! { + result = &mut wait, if child_exit.is_none() => { + child_exit = Some(result.context("wait for command exit")?); + } + stream = stream_rx.recv(), if !stream_rx.is_closed() => { + if let Some((method, chunk)) = stream { + write_json_line( + &mut stdout, + &JsonRpcNotification { + jsonrpc: JSONRPC_VERSION, + method, + params: ExecChunkEvent { + exec_id: exec_id.clone(), + chunk_base64: base64::engine::general_purpose::STANDARD.encode(chunk), + }, + } + ).await?; + } + } + } + } + + while tasks.join_next().await.is_some() {} + #[cfg(unix)] + { + let _ = std::fs::remove_file(&wrapper_socket_path); + } + + let status = child_exit.context("missing child exit status")?; + let exit_code = status.code().unwrap_or(-1); + #[cfg(unix)] + let signal = { + use std::os::unix::process::ExitStatusExt; + status.signal().map(|sig: i32| sig.to_string()) + }; + #[cfg(not(unix))] + let signal = None; + write_json_line( + &mut stdout, + &JsonRpcNotification { + jsonrpc: JSONRPC_VERSION, + method: METHOD_ZSH_EVENT_EXEC_EXITED, + params: ExecExitedEvent { + exec_id: exec_id.clone(), + exit_code, + signal, + timed_out: Some(false), + }, + }, + ) + .await?; + } + METHOD_ZSH_EXEC_INTERRUPT => { + let params: ExecInterruptParams = match parse_params(&value) { + Ok(p) => p, + Err(message) => { + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32602, + message, + }, + }, + ) + .await?; + continue; + } + }; + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32002, + message: format!("unknown exec id: {}", params.exec_id), + }, + }, + ) + .await?; + } + METHOD_ZSH_EXEC_STDIN | METHOD_ZSH_EXEC_RESIZE => { + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32004, + message: "method not supported in sidecar phase 1".to_string(), + }, + }, + ) + .await?; + } + METHOD_ZSH_SHUTDOWN => { + write_json_line( + &mut stdout, + &JsonRpcSuccess { + jsonrpc: JSONRPC_VERSION, + id, + result: EmptyResult {}, + }, + ) + .await?; + break; + } + _ => { + write_json_line( + &mut stdout, + &JsonRpcErrorResponse { + jsonrpc: JSONRPC_VERSION, + id, + error: JsonRpcError { + code: -32601, + message: format!("unknown method: {method}"), + }, + }, + ) + .await?; + } + } + } + + Ok(()) +} + +#[cfg(unix)] +/// Runs exec-wrapper mode for a single intercepted execve. +/// +/// The patched zsh process invokes the sidecar binary as `EXEC_WRAPPER`. +/// This mode forwards the intercepted exec metadata over the wrapper socket, +/// waits for host policy (`run` or `deny`), and then either executes the +/// target program or exits with failure. +fn run_exec_wrapper_mode() -> Result<()> { + use std::io::Read; + use std::io::Write; + use std::os::unix::net::UnixStream as StdUnixStream; + + let args: Vec = std::env::args().collect(); + if args.len() < 2 { + anyhow::bail!("exec wrapper mode requires target executable path"); + } + let file = args[1].clone(); + let argv = if args.len() > 2 { + args[2..].to_vec() + } else { + vec![file.clone()] + }; + let cwd = std::env::current_dir() + .context("resolve wrapper cwd")? + .to_string_lossy() + .to_string(); + let socket_path = std::env::var(SIDECAREXEC_WRAPPER_SOCKET_ENV_VAR) + .context("missing wrapper socket path env var")?; + + let mut stream = StdUnixStream::connect(&socket_path) + .with_context(|| format!("connect to wrapper socket at {socket_path}"))?; + let request = WrapperExecRequest { + file: file.clone(), + argv: argv.clone(), + cwd, + origin: wrapper_exec_origin_from_env(), + }; + let encoded = serde_json::to_string(&request).context("serialize wrapper request")?; + stream + .write_all(encoded.as_bytes()) + .context("write wrapper request")?; + stream + .write_all(b"\n") + .context("write wrapper request newline")?; + stream + .shutdown(std::net::Shutdown::Write) + .context("shutdown wrapper write")?; + + let mut response_buf = String::new(); + stream + .read_to_string(&mut response_buf) + .context("read wrapper response")?; + let response: WrapperExecResponse = + serde_json::from_str(response_buf.trim()).context("parse wrapper response")?; + + if response.action == WrapperExecAction::Deny { + if let Some(reason) = response.reason { + eprintln!("Execution denied: {reason}"); + } else { + eprintln!("Execution denied"); + } + std::process::exit(1); + } + + let mut command = std::process::Command::new(&file); + if argv.len() > 1 { + command.args(&argv[1..]); + } + let status = command.status().context("spawn wrapped executable")?; + std::process::exit(status.code().unwrap_or(1)); +} + +#[cfg(not(unix))] +fn run_exec_wrapper_mode() -> Result<()> { + anyhow::bail!("exec wrapper mode is only supported on unix"); +} + +async fn wait_for_approval_result( + lines: &mut tokio::io::Lines>, + expected_id: JsonRpcId, +) -> Result { + loop { + let Some(line) = lines.next_line().await.context("read stdin")? else { + anyhow::bail!("stdin closed while waiting for approval response"); + }; + if line.trim().is_empty() { + continue; + } + + let value: JsonValue = + serde_json::from_str(&line).context("parse approval response JSON-RPC message")?; + let Some(id_value) = value.get("id") else { + continue; + }; + let id: JsonRpcId = serde_json::from_value(id_value.clone()) + .context("parse approval response JSON-RPC id")?; + if id != expected_id { + tracing::warn!("ignoring unexpected JSON-RPC message while waiting for approval"); + continue; + } + + if let Some(error) = value.get("error") { + let message = error + .get("message") + .and_then(JsonValue::as_str) + .unwrap_or("unknown host approval callback error"); + anyhow::bail!("host rejected approval callback: {message}"); + } + + let result: RequestApprovalResult = serde_json::from_value( + value + .get("result") + .cloned() + .ok_or_else(|| anyhow::anyhow!("missing approval callback result"))?, + ) + .context("parse approval callback result")?; + return Ok(result.decision); + } +} + +#[cfg(unix)] +async fn handle_wrapper_request( + stdout: &mut tokio::io::Stdout, + lines: &mut tokio::io::Lines>, + mut stream: UnixStream, + exec_id: String, + approval_reason: Option, +) -> Result<()> { + let mut request_buf = Vec::new(); + stream + .read_to_end(&mut request_buf) + .await + .context("read wrapper request from socket")?; + let request_line = String::from_utf8(request_buf).context("decode wrapper request as utf-8")?; + let request: WrapperExecRequest = + serde_json::from_str(request_line.trim()).context("parse wrapper request payload")?; + + if should_skip_host_approval_for_request(&request) { + return write_json_line( + &mut stream, + &WrapperExecResponse { + action: WrapperExecAction::Run, + reason: None, + }, + ) + .await; + } + + let command_for_approval = if request.argv.is_empty() { + vec![request.file.clone()] + } else { + request.argv.clone() + }; + let approval_callback_id = JsonRpcId::String(format!("approval-cb-{}", Uuid::new_v4())); + let approval_id = Uuid::new_v4().to_string(); + let approval_request = JsonRpcRequest { + jsonrpc: JSONRPC_VERSION, + id: approval_callback_id.clone(), + method: METHOD_ZSH_REQUEST_APPROVAL, + params: RequestApprovalParams { + approval_id, + exec_id: exec_id.clone(), + command: command_for_approval, + cwd: request.cwd, + reason: approval_reason, + proposed_execpolicy_amendment: None, + }, + }; + write_json_line(stdout, &approval_request).await?; + let decision = wait_for_approval_result(lines, approval_callback_id).await?; + + let response = match decision { + ApprovalDecision::Approved + | ApprovalDecision::ApprovedForSession + | ApprovalDecision::ApprovedExecpolicyAmendment => WrapperExecResponse { + action: WrapperExecAction::Run, + reason: None, + }, + ApprovalDecision::Denied => WrapperExecResponse { + action: WrapperExecAction::Deny, + reason: Some("command denied by host approval policy".to_string()), + }, + ApprovalDecision::Abort => WrapperExecResponse { + action: WrapperExecAction::Deny, + reason: Some("command aborted by host approval policy".to_string()), + }, + }; + write_json_line(&mut stream, &response).await +} + +fn wrapper_exec_origin_from_env() -> Option { + let raw = std::env::var(SIDECAREXEC_WRAPPER_ORIGIN_ENV_VAR).ok()?; + parse_wrapper_exec_origin(raw.as_str()) +} + +fn parse_wrapper_exec_origin(raw: &str) -> Option { + // These numeric values are emitted by the patched zsh fork via: + // `CODEX_ZSH_SIDECAR_WRAPPER_ORIGIN=`. + // + // Mapping comes from zsh's `enum exec_wrapper_origin`: + // 0 = user command + // 1 = login startup (.zprofile/.zlogin/etc.) + // 2 = rc startup (.zshrc/.zshenv/etc.) + // 3 = subcommand + match raw { + "0" => Some(WrapperExecOrigin::UserCommand), + "1" => Some(WrapperExecOrigin::LoginStartup), + "2" => Some(WrapperExecOrigin::RcStartup), + "3" => Some(WrapperExecOrigin::Subcommand), + _ => None, + } +} + +/// Decide whether a wrapper-intercepted execve should bypass host approval. +/// +/// Rationale: +/// - We intentionally suppress startup-file commands (`login_startup`, `rc_startup`) +/// so clients only see approvals for model-generated command execution, not shell bootstrap. +/// - We also suppress internal zsh re-exec handoffs (`zsh -c` / `zsh -fc`) because +/// they are implementation details and not actionable user-facing approvals. +/// - All other commands remain approval-eligible and are surfaced to host policy. +fn should_skip_host_approval_for_request(request: &WrapperExecRequest) -> bool { + if matches!( + request.origin, + Some(WrapperExecOrigin::LoginStartup | WrapperExecOrigin::RcStartup) + ) { + return true; + } + + let request_file = PathBuf::from(&request.file); + let request_file_name = request_file + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or_default(); + let argv_program_name = request + .argv + .first() + .map(PathBuf::from) + .and_then(|path| path.file_name().map(std::borrow::ToOwned::to_owned)) + .and_then(|name| name.to_str().map(str::to_owned)) + .unwrap_or_default(); + let looks_like_zsh_exec = request_file_name == "zsh" || argv_program_name == "zsh"; + let has_command_string_flag = request + .argv + .iter() + .filter(|arg| arg.starts_with('-')) + .any(|arg| arg.chars().skip(1).any(|ch| ch == 'c')); + looks_like_zsh_exec && has_command_string_flag +} + +fn parse_params Deserialize<'de>>(value: &JsonValue) -> std::result::Result { + let params = value + .get("params") + .cloned() + .ok_or_else(|| "missing params".to_string())?; + serde_json::from_value(params).map_err(|err| format!("invalid params: {err}")) +} + +async fn write_json_line( + writer: &mut W, + message: &T, +) -> Result<()> { + let encoded = serde_json::to_string(message).context("serialize JSON-RPC message")?; + writer + .write_all(encoded.as_bytes()) + .await + .context("write message")?; + writer.write_all(b"\n").await.context("write newline")?; + writer.flush().await.context("flush message")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn wrapper_request( + origin: Option, + file: &str, + argv: &[&str], + ) -> WrapperExecRequest { + WrapperExecRequest { + file: file.to_string(), + argv: argv.iter().map(ToString::to_string).collect(), + cwd: "/tmp".to_string(), + origin, + } + } + + #[test] + fn skips_host_approval_for_login_startup_origin() { + let request = wrapper_request( + Some(WrapperExecOrigin::LoginStartup), + "/usr/bin/echo", + &["/usr/bin/echo", "hi"], + ); + assert!(should_skip_host_approval_for_request(&request)); + } + + #[test] + fn skips_host_approval_for_rc_startup_origin() { + let request = wrapper_request( + Some(WrapperExecOrigin::RcStartup), + "/usr/bin/echo", + &["/usr/bin/echo", "hi"], + ); + assert!(should_skip_host_approval_for_request(&request)); + } + + #[test] + fn does_not_skip_host_approval_for_user_command_origin() { + let request = wrapper_request( + Some(WrapperExecOrigin::UserCommand), + "/usr/bin/true", + &["/usr/bin/true"], + ); + assert!(!should_skip_host_approval_for_request(&request)); + } + + #[test] + fn skips_host_approval_for_internal_zsh_reexec_without_origin() { + let request = wrapper_request(None, "/Users/test/bin/zsh", &["zsh", "-fc", "true"]); + assert!(should_skip_host_approval_for_request(&request)); + } + + #[test] + fn parses_numeric_wrapper_origin_values() { + assert_eq!( + parse_wrapper_exec_origin("1"), + Some(WrapperExecOrigin::LoginStartup) + ); + assert_eq!( + parse_wrapper_exec_origin("2"), + Some(WrapperExecOrigin::RcStartup) + ); + assert_eq!( + parse_wrapper_exec_origin("0"), + Some(WrapperExecOrigin::UserCommand) + ); + } +} diff --git a/codex-rs/zsh-sidecar/tests/protocol.rs b/codex-rs/zsh-sidecar/tests/protocol.rs new file mode 100644 index 0000000000..aca47065f6 --- /dev/null +++ b/codex-rs/zsh-sidecar/tests/protocol.rs @@ -0,0 +1,952 @@ +#![cfg(unix)] + +use anyhow::Context; +use anyhow::Result; +use serde_json::Value as JsonValue; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::net::UnixListener; +use tokio::process::Child; +use tokio::process::ChildStdin; +use tokio::process::ChildStdout; +use tokio::process::Command; +use tokio::time::Duration; +use tokio::time::timeout; + +const JSONRPC_VERSION: &str = "2.0"; +const EXEC_START_REQUEST_ID: i64 = 2; +const WRAPPER_MODE_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_MODE"; +const WRAPPER_SOCKET_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_SOCKET"; +const WRAPPER_ORIGIN_ENV_VAR: &str = "CODEX_ZSH_SIDECAR_WRAPPER_ORIGIN"; + +#[tokio::test] +async fn exec_start_emits_multiple_subcommand_approvals_for_compound_command() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + let mut exec_start_acked = false; + let mut intercepted_subcommand_callbacks = 0usize; + let mut intercepted_true_callbacks = 0usize; + let mut saw_exec_exited = false; + let mut exit_code = None; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + + if let Some((id, _reason, command)) = parse_approval_request(&value) { + if command.first().is_some_and(|c| c == "/usr/bin/true") { + intercepted_subcommand_callbacks += 1; + intercepted_true_callbacks += 1; + } + harness.respond_approval(id, "approved").await?; + continue; + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + exec_start_acked = true; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + exit_code = value + .pointer("/params/exitCode") + .and_then(JsonValue::as_i64) + .map(|code| code as i32); + } + } + + harness.shutdown().await?; + + assert!(exec_start_acked, "expected execStart success response"); + assert_eq!(exit_code, Some(0), "expected successful command exit"); + assert!( + intercepted_subcommand_callbacks >= 2, + "expected at least two intercepted subcommand approvals, got {intercepted_subcommand_callbacks}" + ); + assert!( + intercepted_true_callbacks >= 2, + "expected at least two intercepted /usr/bin/true approvals, got {intercepted_true_callbacks}" + ); + Ok(()) +} + +#[tokio::test] +async fn exec_start_uses_custom_approval_reason_when_provided() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command( + "/usr/bin/true && /usr/bin/true", + Some("test custom zsh approval reason"), + ) + .await?; + + let mut saw_custom_reason = false; + let mut saw_exec_exited = false; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + if let Some((id, reason, _command)) = parse_approval_request(&value) { + if reason.as_deref() == Some("test custom zsh approval reason") { + saw_custom_reason = true; + } + harness.respond_approval(id, "approved").await?; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + } + } + + harness.shutdown().await?; + assert!( + saw_custom_reason, + "expected at least one approval callback with custom reason" + ); + Ok(()) +} + +#[tokio::test] +async fn exec_start_does_not_emit_exec_start_approval_callback() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + let mut saw_exec_start_success = false; + let mut saw_exec_exited = false; + let mut saw_exec_start_reason = false; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + if let Some((id, reason, _command)) = parse_approval_request(&value) { + if reason.as_deref() == Some("zsh sidecar execStart command approval") { + saw_exec_start_reason = true; + } + harness.respond_approval(id, "approved").await?; + continue; + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + saw_exec_start_success = true; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + } + } + + harness.shutdown().await?; + assert!( + saw_exec_start_success, + "expected execStart success response" + ); + assert!( + !saw_exec_start_reason, + "did not expect execStart approval callback reason" + ); + Ok(()) +} + +#[tokio::test] +async fn exec_start_does_not_emit_internal_zsh_reexec_approval_callback() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + let mut saw_exec_start_success = false; + let mut saw_exec_exited = false; + let mut saw_internal_zsh_reexec = false; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + if let Some((id, _reason, command)) = parse_approval_request(&value) { + let command_program_name = command + .first() + .map(std::path::PathBuf::from) + .and_then(|path| path.file_name().map(std::borrow::ToOwned::to_owned)) + .and_then(|name| name.to_str().map(str::to_owned)) + .unwrap_or_default(); + let has_command_string_flag = command + .iter() + .filter(|arg| arg.starts_with('-')) + .any(|arg| arg.chars().skip(1).any(|ch| ch == 'c')); + if command_program_name == "zsh" && has_command_string_flag { + saw_internal_zsh_reexec = true; + } + harness.respond_approval(id, "approved").await?; + continue; + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + saw_exec_start_success = true; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + } + } + + harness.shutdown().await?; + assert!( + saw_exec_start_success, + "expected execStart success response" + ); + assert!( + !saw_internal_zsh_reexec, + "did not expect host approval callback for internal zsh re-exec" + ); + Ok(()) +} + +#[tokio::test] +async fn denying_second_subcommand_fails_exec() -> Result<()> { + let Some(outcome) = run_compound_exec_with_second_subcommand_decision("denied").await? else { + return Ok(()); + }; + assert!( + outcome.saw_exec_start_success, + "expected execStart success response" + ); + assert!( + outcome.subcommand_callbacks >= 2, + "expected at least two subcommand callbacks before exit, got {}", + outcome.subcommand_callbacks + ); + assert_ne!( + outcome.exit_code, + Some(0), + "denying the second subcommand should cause non-zero exit" + ); + Ok(()) +} + +#[tokio::test] +async fn aborting_second_subcommand_fails_exec() -> Result<()> { + let Some(outcome) = run_compound_exec_with_second_subcommand_decision("abort").await? else { + return Ok(()); + }; + assert!( + outcome.saw_exec_start_success, + "expected execStart success response" + ); + assert!( + outcome.subcommand_callbacks >= 2, + "expected at least two subcommand callbacks before exit, got {}", + outcome.subcommand_callbacks + ); + assert_ne!( + outcome.exit_code, + Some(0), + "aborting the second subcommand should cause non-zero exit" + ); + Ok(()) +} + +#[tokio::test] +async fn approved_for_session_subcommand_decision_allows_exec() -> Result<()> { + let Some(outcome) = + run_compound_exec_with_second_subcommand_decision("approved_for_session").await? + else { + return Ok(()); + }; + assert!( + outcome.saw_exec_start_success, + "expected execStart success response" + ); + assert!( + outcome.subcommand_callbacks >= 2, + "expected at least two subcommand callbacks before exit, got {}", + outcome.subcommand_callbacks + ); + assert_eq!( + outcome.exit_code, + Some(0), + "approved_for_session should allow successful command exit" + ); + Ok(()) +} + +#[tokio::test] +async fn approved_execpolicy_amendment_subcommand_decision_allows_exec() -> Result<()> { + let Some(outcome) = + run_compound_exec_with_second_subcommand_decision("approved_execpolicy_amendment").await? + else { + return Ok(()); + }; + assert!( + outcome.saw_exec_start_success, + "expected execStart success response" + ); + assert!( + outcome.subcommand_callbacks >= 2, + "expected at least two subcommand callbacks before exit, got {}", + outcome.subcommand_callbacks + ); + assert_eq!( + outcome.exit_code, + Some(0), + "approved_execpolicy_amendment should allow successful command exit" + ); + Ok(()) +} + +#[tokio::test] +async fn approval_callback_ignores_unexpected_response_id() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + let mut sent_wrong_id_once = false; + let mut saw_exec_start_success = false; + let mut saw_exec_exited = false; + let mut exit_code = None; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + if let Some((id, _reason, _command)) = parse_approval_request(&value) { + if !sent_wrong_id_once { + harness + .respond_approval( + JsonValue::String("definitely-wrong-id".to_string()), + "approved", + ) + .await?; + sent_wrong_id_once = true; + } + harness.respond_approval(id, "approved").await?; + continue; + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + saw_exec_start_success = true; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + exit_code = value + .pointer("/params/exitCode") + .and_then(JsonValue::as_i64) + .map(|code| code as i32); + } + } + + harness.shutdown().await?; + + assert!(sent_wrong_id_once, "expected wrong-id response to be sent"); + assert!( + saw_exec_start_success, + "expected execStart success despite wrong callback id response" + ); + assert_eq!(exit_code, Some(0), "expected successful command exit"); + Ok(()) +} + +#[tokio::test] +async fn malformed_approval_response_terminates_sidecar() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + loop { + let value = harness.read_next_message().await?; + if let Some((id, _reason, _command)) = parse_approval_request(&value) { + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": id, + "result": {} + })) + .await?; + break; + } + } + + let status = timeout(Duration::from_secs(3), harness.child.wait()) + .await + .context("timed out waiting for sidecar crash on malformed callback response")??; + assert!( + !status.success(), + "sidecar should fail fast on malformed callback response" + ); + Ok(()) +} + +#[tokio::test] +async fn returns_jsonrpc_error_for_unknown_method() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 55, + "method": "zsh/notRealMethod", + "params": {} + })) + .await?; + let response = harness.wait_for_response(55).await?; + assert_eq!( + response.pointer("/error/code"), + Some(&JsonValue::from(-32601)) + ); + + harness.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn returns_jsonrpc_invalid_params_for_exec_start_with_empty_command() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": EXEC_START_REQUEST_ID, + "method": "zsh/execStart", + "params": { + "execId": "exec-invalid", + "command": [], + "cwd": std::env::current_dir()?.to_string_lossy().to_string(), + "env": {} + } + })) + .await?; + let response = harness.wait_for_response(EXEC_START_REQUEST_ID).await?; + assert_eq!( + response.pointer("/error/code"), + Some(&JsonValue::from(-32602)) + ); + + harness.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn exec_events_are_ordered_exec_started_before_output_and_single_exit() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/printf 'hi\\n'", None) + .await?; + + let mut saw_exec_started = false; + let mut saw_output_before_started = false; + let mut exec_exited_count = 0usize; + let mut saw_exec_start_success = false; + + while exec_exited_count == 0 { + let value = harness.read_next_message().await?; + if let Some((id, _reason, _command)) = parse_approval_request(&value) { + harness.respond_approval(id, "approved").await?; + continue; + } + + match value.get("method").and_then(JsonValue::as_str) { + Some("zsh/event/execStarted") => { + saw_exec_started = true; + } + Some("zsh/event/execStdout") | Some("zsh/event/execStderr") => { + if !saw_exec_started { + saw_output_before_started = true; + } + } + Some("zsh/event/execExited") => { + exec_exited_count += 1; + } + _ => {} + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + saw_exec_start_success = true; + } + } + + for _ in 0..4 { + let Some(value) = harness + .read_next_message_with_timeout(Duration::from_millis(100)) + .await? + else { + break; + }; + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + exec_exited_count += 1; + } + } + + harness.shutdown().await?; + + assert!( + saw_exec_start_success, + "expected execStart success response" + ); + assert!(saw_exec_started, "expected execStarted event"); + assert!(!saw_output_before_started, "saw output before execStarted"); + assert_eq!(exec_exited_count, 1, "expected one execExited event"); + Ok(()) +} + +#[tokio::test] +async fn exec_interrupt_returns_unknown_exec_id_error_in_phase1() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 77, + "method": "zsh/execInterrupt", + "params": { + "execId": "exec-does-not-exist" + } + })) + .await?; + let response = harness.wait_for_response(77).await?; + assert_eq!( + response.pointer("/error/code"), + Some(&JsonValue::from(-32002)) + ); + + harness.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn exec_stdin_and_resize_return_not_supported_error_in_phase1() -> Result<()> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(()); + }; + + harness.initialize().await?; + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 78, + "method": "zsh/execStdin", + "params": { + "execId": "exec-does-not-exist", + "chunkBase64": "aGk=" + } + })) + .await?; + let stdin_response = harness.wait_for_response(78).await?; + assert_eq!( + stdin_response.pointer("/error/code"), + Some(&JsonValue::from(-32004)) + ); + + harness + .write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 79, + "method": "zsh/execResize", + "params": { + "execId": "exec-does-not-exist", + "cols": 80, + "rows": 24 + } + })) + .await?; + let resize_response = harness.wait_for_response(79).await?; + assert_eq!( + resize_response.pointer("/error/code"), + Some(&JsonValue::from(-32004)) + ); + + harness.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn wrapper_mode_with_invalid_socket_fails_fast() -> Result<()> { + let sidecar = env!("CARGO_BIN_EXE_codex-zsh-sidecar"); + let mut child = Command::new(sidecar) + .arg("/usr/bin/true") + .env(WRAPPER_MODE_ENV_VAR, "1") + .env( + WRAPPER_SOCKET_ENV_VAR, + "/tmp/definitely-not-a-real-codex-zsh-wrapper.sock", + ) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .spawn() + .context("spawn wrapper mode sidecar process")?; + + let status = timeout(Duration::from_secs(3), child.wait()) + .await + .context("timed out waiting for wrapper mode process failure")??; + let stderr = child + .stderr + .take() + .context("missing stderr for wrapper mode process")?; + let mut lines = BufReader::new(stderr).lines(); + let mut stderr_text = String::new(); + while let Some(line) = lines.next_line().await? { + stderr_text.push_str(&line); + stderr_text.push('\n'); + } + + assert!( + !status.success(), + "wrapper mode should fail when socket path is invalid" + ); + assert!( + stderr_text.contains("wrapper socket"), + "expected wrapper socket failure message, got: {stderr_text}" + ); + Ok(()) +} + +#[tokio::test] +async fn wrapper_mode_sends_origin_metadata_from_env() -> Result<()> { + let sidecar = env!("CARGO_BIN_EXE_codex-zsh-sidecar"); + let socket_path = std::env::temp_dir().join(format!( + "czs-wrapper-test-{}.sock", + uuid::Uuid::new_v4().as_simple() + )); + let _ = std::fs::remove_file(&socket_path); + let listener = UnixListener::bind(&socket_path) + .with_context(|| format!("bind test listener at {}", socket_path.display()))?; + + let mut child = Command::new(sidecar) + .arg("/usr/bin/true") + .env(WRAPPER_MODE_ENV_VAR, "1") + .env( + WRAPPER_SOCKET_ENV_VAR, + socket_path.to_string_lossy().to_string(), + ) + .env(WRAPPER_ORIGIN_ENV_VAR, "1") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .spawn() + .context("spawn wrapper mode sidecar process")?; + + let (mut stream, _) = timeout(Duration::from_secs(3), listener.accept()) + .await + .context("timed out waiting for wrapper connection")??; + let mut request_buf = Vec::new(); + stream + .read_to_end(&mut request_buf) + .await + .context("read wrapper request payload")?; + let request: JsonValue = + serde_json::from_slice(&request_buf).context("parse wrapper request payload")?; + assert_eq!( + request.pointer("/origin"), + Some(&JsonValue::from("login_startup")) + ); + stream + .write_all(br#"{"action":"run","reason":null}"#) + .await + .context("write wrapper allow response")?; + stream.shutdown().await.context("shutdown wrapper stream")?; + + let status = timeout(Duration::from_secs(3), child.wait()) + .await + .context("timed out waiting for wrapper mode process exit")??; + let _ = std::fs::remove_file(&socket_path); + assert!( + status.success(), + "wrapper mode should succeed when action=run" + ); + Ok(()) +} + +struct SidecarHarness { + child: Child, + stdin: ChildStdin, + lines: tokio::io::Lines>, + zsh_path: std::path::PathBuf, +} + +impl SidecarHarness { + async fn start() -> Result> { + let Some(zsh_path) = std::env::var_os("CODEX_TEST_ZSH_PATH") else { + eprintln!("skipping direct sidecar protocol test: CODEX_TEST_ZSH_PATH is not set"); + return Ok(None); + }; + let zsh_path = std::path::PathBuf::from(zsh_path); + if !zsh_path.is_file() { + anyhow::bail!( + "CODEX_TEST_ZSH_PATH is set but is not a file: {}", + zsh_path.display() + ); + } + + let sidecar = env!("CARGO_BIN_EXE_codex-zsh-sidecar"); + let mut child = Command::new(sidecar) + .arg("--zsh-path") + .arg(&zsh_path) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::inherit()) + .spawn() + .context("spawn codex-zsh-sidecar")?; + + let stdin = child.stdin.take().context("missing sidecar stdin")?; + let stdout = child.stdout.take().context("missing sidecar stdout")?; + + Ok(Some(Self { + child, + stdin, + lines: BufReader::new(stdout).lines(), + zsh_path, + })) + } + + async fn initialize(&mut self) -> Result<()> { + self.write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 1, + "method": "zsh/initialize", + "params": { + "sessionId": "test-session" + } + })) + .await?; + self.wait_for_response(1).await?; + Ok(()) + } + + async fn start_exec_with_command( + &mut self, + shell_command: &str, + approval_reason: Option<&str>, + ) -> Result<()> { + self.write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": EXEC_START_REQUEST_ID, + "method": "zsh/execStart", + "params": { + "execId": "exec-test-1", + "command": [self.zsh_path.to_string_lossy(), "-fc", shell_command], + "cwd": std::env::current_dir()?.to_string_lossy().to_string(), + "env": {}, + "approvalReason": approval_reason + } + })) + .await + } + + async fn respond_approval(&mut self, id: JsonValue, decision: &str) -> Result<()> { + self.write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": id, + "result": { + "decision": decision + } + })) + .await + } + + async fn shutdown(&mut self) -> Result<()> { + self.write_json_line(&serde_json::json!({ + "jsonrpc": JSONRPC_VERSION, + "id": 3, + "method": "zsh/shutdown", + "params": { + "graceMs": 100 + } + })) + .await?; + self.wait_for_response(3).await?; + + let status = timeout(Duration::from_secs(3), self.child.wait()) + .await + .context("timed out waiting for sidecar process exit")??; + assert!(status.success(), "sidecar should exit cleanly"); + Ok(()) + } + + async fn read_next_message(&mut self) -> Result { + let line = timeout(Duration::from_secs(10), self.lines.next_line()) + .await + .context("timed out reading sidecar output")?? + .context("sidecar stdout closed unexpectedly")?; + serde_json::from_str(&line).with_context(|| format!("parse sidecar JSON line: {line}")) + } + + async fn read_next_message_with_timeout( + &mut self, + duration: Duration, + ) -> Result> { + let line = match timeout(duration, self.lines.next_line()).await { + Ok(line) => line?, + Err(_) => return Ok(None), + }; + let Some(line) = line else { + return Ok(None); + }; + let value = serde_json::from_str(&line) + .with_context(|| format!("parse sidecar JSON line: {line}"))?; + Ok(Some(value)) + } + + async fn wait_for_response(&mut self, id: i64) -> Result { + loop { + let value = self.read_next_message().await?; + if value.get("id").and_then(JsonValue::as_i64) == Some(id) { + return Ok(value); + } + } + } + + async fn write_json_line(&mut self, value: &JsonValue) -> Result<()> { + let encoded = serde_json::to_string(value).context("serialize JSON line")?; + self.stdin + .write_all(encoded.as_bytes()) + .await + .context("write JSON line")?; + self.stdin + .write_all(b"\n") + .await + .context("write line break")?; + self.stdin.flush().await.context("flush stdin") + } +} + +fn parse_approval_request(value: &JsonValue) -> Option<(JsonValue, Option, Vec)> { + if value.get("method").and_then(JsonValue::as_str) != Some("zsh/requestApproval") { + return None; + } + let id = value.get("id")?.clone(); + let reason = value + .pointer("/params/reason") + .and_then(JsonValue::as_str) + .map(ToString::to_string); + let command = value + .pointer("/params/command") + .and_then(JsonValue::as_array) + .map(|items| { + items + .iter() + .filter_map(JsonValue::as_str) + .map(ToString::to_string) + .collect() + }) + .unwrap_or_default(); + Some((id, reason, command)) +} + +struct CompoundExecOutcome { + saw_exec_start_success: bool, + subcommand_callbacks: usize, + exit_code: Option, +} + +async fn run_compound_exec_with_second_subcommand_decision( + second_subcommand_decision: &str, +) -> Result> { + let Some(mut harness) = SidecarHarness::start().await? else { + return Ok(None); + }; + + harness.initialize().await?; + harness + .start_exec_with_command("/usr/bin/true && /usr/bin/true", None) + .await?; + + let mut subcommand_callbacks = 0usize; + let mut saw_exec_start_success = false; + let mut saw_exec_exited = false; + let mut exit_code = None; + + while !saw_exec_exited { + let value = harness.read_next_message().await?; + + if let Some((id, _reason, command)) = parse_approval_request(&value) { + if command.first().is_some_and(|c| c == "/usr/bin/true") { + subcommand_callbacks += 1; + if subcommand_callbacks == 2 { + harness + .respond_approval(id, second_subcommand_decision) + .await?; + } else { + harness.respond_approval(id, "approved").await?; + } + } else { + harness.respond_approval(id, "approved").await?; + } + continue; + } + + if value.get("id").and_then(JsonValue::as_i64) == Some(EXEC_START_REQUEST_ID) + && value.get("result").is_some() + { + saw_exec_start_success = true; + continue; + } + + if value.get("method").and_then(JsonValue::as_str) == Some("zsh/event/execExited") { + saw_exec_exited = true; + exit_code = value + .pointer("/params/exitCode") + .and_then(JsonValue::as_i64) + .map(|code| code as i32); + } + } + + harness.shutdown().await?; + Ok(Some(CompoundExecOutcome { + saw_exec_start_success, + subcommand_callbacks, + exit_code, + })) +}