diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 243f944..26b5041 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -59,9 +59,11 @@ use std::process::{Child, Command, Stdio}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}; #[cfg(unix)] use tokio::net::UnixStream; +#[cfg(windows)] +use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeClient}; use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::sync::mpsc; @@ -81,9 +83,12 @@ const LOCAL_SLEEP_REPAINT_RETRY_WINDOW: Duration = Duration::from_secs(10); /// Sentinel `target_host` the server uses on a server-initiated `StreamOpen` that /// represents an SSH-agent connection (rather than a TCP target). The client -/// splices these into its local `SSH_AUTH_SOCK` instead of dialing TCP. The +/// splices these into its local platform ssh-agent instead of dialing TCP. The /// `:0` port and this reserved name are never valid TCP forward targets. const AGENT_STREAM_SENTINEL: &str = "@dosh-agent"; +#[cfg(windows)] +const WINDOWS_OPENSSH_AGENT_PIPE: &str = r"\\.\pipe\openssh-ssh-agent"; +type AgentWriter = Box; /// Current terminal size, with a sane fallback. /// @@ -134,7 +139,7 @@ struct Args { remote_forward: Vec, #[arg(short = 'D', long = "dynamic-forward")] dynamic_forward: Vec, - /// Forward the local ssh-agent (SSH_AUTH_SOCK) to the remote session. + /// Forward the local ssh-agent to the remote session. /// SECURITY: opt-in only; the server must also set allow_agent_forwarding. #[arg(short = 'A', long = "forward-agent")] forward_agent: bool, @@ -349,21 +354,10 @@ async fn main() -> Result<()> { // SSH-agent forwarding (opt-in via -A / forward_agent). SECURITY: this exposes // your local agent to the remote host for the session's lifetime; only ever // active on explicit opt-in AND when the server's allow_agent_forwarding is - // set. We capture the LOCAL agent socket path now so the run loop can splice - // each server-initiated agent stream into it. + // set. Resolve the local agent endpoint now so the run loop can splice each + // server-initiated agent stream into it. let forward_agent = args.forward_agent || config.forward_agent; - let agent_sock = if forward_agent { - match std::env::var_os("SSH_AUTH_SOCK") { - Some(path) if !path.is_empty() => Some(PathBuf::from(path)), - _ => { - return Err(anyhow!( - "agent forwarding requested but SSH_AUTH_SOCK is not set" - )); - } - } - } else { - None - }; + let agent_sock = resolve_forward_agent_endpoint(forward_agent)?; let forwarding_requested = !local_forwards.is_empty() || !remote_forwards.is_empty() || !dynamic_forwards.is_empty() @@ -5006,6 +5000,54 @@ fn resolve_addr(host: &str, port: u16) -> Result { .ok_or_else(|| anyhow!("no UDP address resolved for {host}:{port}")) } +fn resolve_forward_agent_endpoint(forward_agent: bool) -> Result> { + if !forward_agent { + return Ok(None); + } + + #[cfg(unix)] + { + return match std::env::var_os("SSH_AUTH_SOCK") { + Some(path) if !path.is_empty() => Ok(Some(PathBuf::from(path))), + _ => Err(anyhow!( + "agent forwarding requested but SSH_AUTH_SOCK is not set" + )), + }; + } + + #[cfg(windows)] + { + Ok(Some(PathBuf::from(WINDOWS_OPENSSH_AGENT_PIPE))) + } + + #[cfg(not(any(unix, windows)))] + { + Err(anyhow!( + "agent forwarding is not supported on this client platform" + )) + } +} + +#[cfg(unix)] +type LocalAgentStream = UnixStream; + +#[cfg(windows)] +type LocalAgentStream = NamedPipeClient; + +#[cfg(unix)] +async fn connect_local_agent(path: &Path) -> Result { + UnixStream::connect(path) + .await + .with_context(|| format!("connect local agent {}", path.display())) +} + +#[cfg(windows)] +async fn connect_local_agent(path: &Path) -> Result { + ClientOptions::new() + .open(path) + .with_context(|| format!("connect Windows OpenSSH agent pipe {}", path.display())) +} + async fn send_terminal_udp(socket: &UdpSocket, packet: &[u8], addr: SocketAddr) -> Result { match socket.send_to(packet, addr).await { Ok(_) => Ok(true), @@ -5855,9 +5897,6 @@ async fn run_terminal( forward_only: bool, agent_sock: Option, ) -> Result<()> { - #[cfg(not(unix))] - let _ = &agent_sock; - let _raw = if forward_only { None } else { @@ -5903,11 +5942,10 @@ async fn run_terminal( start_dynamic_forwards(&dynamic_forwards, forward_tx.clone(), stream_ids).await?; signal_background_ready(); let mut stream_writers: HashMap = HashMap::new(); - // Write halves for server-initiated SSH-agent streams, spliced into the local - // unix agent socket. Kept separate from the TCP `stream_writers` so the - // existing TCP forwarding paths are untouched. - #[cfg(unix)] - let mut agent_writers: HashMap = HashMap::new(); + // Write halves for server-initiated SSH-agent streams, spliced into the + // local platform agent endpoint. Kept separate from TCP `stream_writers` so + // the existing TCP forwarding paths are untouched. + let mut agent_writers: HashMap = HashMap::new(); let mut pending_socks_replies: HashSet = HashSet::new(); let mut opened_streams: HashSet = HashSet::new(); let mut stream_send_credit: HashMap = HashMap::new(); @@ -6629,99 +6667,83 @@ async fn run_terminal( // actually opted in (agent_sock is Some); otherwise reject, // so a server cannot reach our agent without consent. if open.target_host == AGENT_STREAM_SENTINEL { - #[cfg(not(unix))] - { + let Some(sock_path) = agent_sock.clone() else { send_stream_open_reject( &socket, addr, &cred, &mut send_seq, open.stream_id, - "agent forwarding is not supported on this client platform" - .to_string(), + "agent forwarding not enabled by client".to_string(), ) .await?; - } - #[cfg(unix)] - { - let Some(sock_path) = agent_sock.clone() else { + continue; + }; + match connect_local_agent(&sock_path).await { + Ok(stream) => { + let (mut reader, writer) = tokio::io::split(stream); + agent_writers.insert(open.stream_id, Box::new(writer)); + opened_streams.insert(open.stream_id); + stream_send_credit + .insert(open.stream_id, STREAM_INITIAL_WINDOW); + stream_next_send_offset.entry(open.stream_id).or_insert(0); + stream_next_recv_offset.entry(open.stream_id).or_insert(0); + send_stream_open_ok( + &socket, + addr, + &cred, + &mut send_seq, + open.stream_id, + ) + .await?; + let forward_tx = remote_forward_tx.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 16 * 1024]; + let mut graceful_eof = false; + loop { + match reader.read(&mut buf).await { + Ok(0) => { + graceful_eof = true; + break; + } + Ok(n) => { + if forward_tx + .send(ForwardEvent::Data { + stream_id: open.stream_id, + bytes: buf[..n].to_vec(), + }) + .await + .is_err() + { + return; + } + } + Err(_) => break, + } + } + let _ = forward_tx + .send(if graceful_eof { + ForwardEvent::Eof { + stream_id: open.stream_id, + } + } else { + ForwardEvent::Close { + stream_id: open.stream_id, + } + }) + .await; + }); + } + Err(err) => { send_stream_open_reject( &socket, addr, &cred, &mut send_seq, open.stream_id, - "agent forwarding not enabled by client".to_string(), + format!("connect local agent: {err}"), ) .await?; - continue; - }; - match UnixStream::connect(&sock_path).await { - Ok(stream) => { - let (mut reader, writer) = stream.into_split(); - agent_writers.insert(open.stream_id, writer); - opened_streams.insert(open.stream_id); - stream_send_credit - .insert(open.stream_id, STREAM_INITIAL_WINDOW); - stream_next_send_offset.entry(open.stream_id).or_insert(0); - stream_next_recv_offset.entry(open.stream_id).or_insert(0); - send_stream_open_ok( - &socket, - addr, - &cred, - &mut send_seq, - open.stream_id, - ) - .await?; - let forward_tx = remote_forward_tx.clone(); - tokio::spawn(async move { - let mut buf = [0u8; 16 * 1024]; - let mut graceful_eof = false; - loop { - match reader.read(&mut buf).await { - Ok(0) => { - graceful_eof = true; - break; - } - Ok(n) => { - if forward_tx - .send(ForwardEvent::Data { - stream_id: open.stream_id, - bytes: buf[..n].to_vec(), - }) - .await - .is_err() - { - return; - } - } - Err(_) => break, - } - } - let _ = forward_tx - .send(if graceful_eof { - ForwardEvent::Eof { - stream_id: open.stream_id, - } - } else { - ForwardEvent::Close { - stream_id: open.stream_id, - } - }) - .await; - }); - } - Err(err) => { - send_stream_open_reject( - &socket, - addr, - &cred, - &mut send_seq, - open.stream_id, - format!("connect local agent: {err}"), - ) - .await?; - } } } continue; @@ -6886,12 +6908,9 @@ async fn run_terminal( for bytes in &writes { let _ = writer.write_all(bytes).await; } - } else { - #[cfg(unix)] - if let Some(writer) = agent_writers.get_mut(&stream_id) { - for bytes in &writes { - let _ = writer.write_all(bytes).await; - } + } else if let Some(writer) = agent_writers.get_mut(&stream_id) { + for bytes in &writes { + let _ = writer.write_all(bytes).await; } } // Always return flow-control credit so a stream whose local @@ -6962,7 +6981,6 @@ async fn run_terminal( if let Some(writer) = stream_writers.get_mut(&eof.stream_id) { let _ = writer.shutdown().await; } - #[cfg(unix)] if let Some(writer) = agent_writers.get_mut(&eof.stream_id) { let _ = writer.shutdown().await; } @@ -6990,7 +7008,6 @@ async fn run_terminal( &mut stream_retired_order, ); stream_writers.remove(&eof.stream_id); - #[cfg(unix)] agent_writers.remove(&eof.stream_id); } } @@ -7022,7 +7039,6 @@ async fn run_terminal( &mut stream_retired_order, ); stream_writers.remove(&close.stream_id); - #[cfg(unix)] agent_writers.remove(&close.stream_id); } _ => {} @@ -7106,7 +7122,6 @@ async fn run_terminal( &mut stream_retired_order, ); stream_writers.remove(&stream_id); - #[cfg(unix)] agent_writers.remove(&stream_id); } } @@ -7130,7 +7145,6 @@ async fn run_terminal( &mut stream_retired_order, ); stream_writers.remove(&stream_id); - #[cfg(unix)] agent_writers.remove(&stream_id); send_stream_close(&socket, addr, &cred, &mut send_seq, stream_id).await?; stream_close_retransmit.insert( @@ -10028,11 +10042,11 @@ mod tests { queue_pending_user_input, queue_stale_pending_user_input, raw_contains_host_table, recv_response_until, refresh_live_addr, release_tag_download_url, release_tag_from_effective_url, release_version_from_tag, render_frame_bytes, - render_status_clear, render_status_overlay, requested_env, resolved_startup_command, - retire_stream_state, retransmit_stream_closes, retransmit_stream_eofs, - retransmit_stream_opens, retransmit_stream_window_adjusts, rewrite_forward_command, - sanitize_trace_name, selected_predict_mode, selected_udp_host, send_stream_eof, - server_version_mismatch, should_flush_terminal_input_after_contact, + render_status_clear, render_status_overlay, requested_env, resolve_forward_agent_endpoint, + resolved_startup_command, retire_stream_state, retransmit_stream_closes, + retransmit_stream_eofs, retransmit_stream_opens, retransmit_stream_window_adjusts, + rewrite_forward_command, sanitize_trace_name, selected_predict_mode, selected_udp_host, + send_stream_eof, server_version_mismatch, should_flush_terminal_input_after_contact, should_health_log_client_start, should_hold_during_startup_gate, should_hold_post_submit_input, should_reconnect_before_input_for_local_sleep, should_repaint_idle_terminal, should_strip_unowned_terminal_reports, @@ -10096,6 +10110,20 @@ mod tests { assert_eq!(vscode_safe_alias(":///"), "host"); } + #[test] + fn forward_agent_endpoint_disabled_is_none() { + assert_eq!(resolve_forward_agent_endpoint(false).unwrap(), None); + } + + #[cfg(windows)] + #[test] + fn windows_forward_agent_endpoint_uses_openssh_pipe() { + assert_eq!( + resolve_forward_agent_endpoint(true).unwrap().unwrap(), + std::path::PathBuf::from(super::WINDOWS_OPENSSH_AGENT_PIPE) + ); + } + #[test] fn cleanup_stream_state_removes_all_per_stream_state_only_for_target() { let stream_id = 42;