Support Windows agent forwarding

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<dyn AsyncWrite + Unpin + Send>;
 
 /// Current terminal size, with a sane fallback.
 ///
@@ -134,7 +139,7 @@ struct Args {
     remote_forward: Vec<String>,
     #[arg(short = 'D', long = "dynamic-forward")]
     dynamic_forward: Vec<String>,
-    /// 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
+    // set. Resolve the local agent endpoint now so the run loop can splice each
-    // each server-initiated agent stream into it.
+    // server-initiated agent stream into it.
     let forward_agent = args.forward_agent || config.forward_agent;
-    let agent_sock = if forward_agent {
+    let agent_sock = resolve_forward_agent_endpoint(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 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<SocketAddr> {
         .ok_or_else(|| anyhow!("no UDP address resolved for {host}:{port}"))
 }
 
+fn resolve_forward_agent_endpoint(forward_agent: bool) -> Result<Option<PathBuf>> {
+    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<LocalAgentStream> {
+    UnixStream::connect(path)
+        .await
+        .with_context(|| format!("connect local agent {}", path.display()))
+}
+
+#[cfg(windows)]
+async fn connect_local_agent(path: &Path) -> Result<LocalAgentStream> {
+    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<bool> {
     match socket.send_to(packet, addr).await {
         Ok(_) => Ok(true),
@@ -5855,9 +5897,6 @@ async fn run_terminal(
     forward_only: bool,
     agent_sock: Option<PathBuf>,
 ) -> 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<u64, tokio::net::tcp::OwnedWriteHalf> = HashMap::new();
-    // Write halves for server-initiated SSH-agent streams, spliced into the local
+    // Write halves for server-initiated SSH-agent streams, spliced into the
-    // unix agent socket. Kept separate from the TCP `stream_writers` so the
+    // local platform agent endpoint. Kept separate from TCP `stream_writers` so
-    // existing TCP forwarding paths are untouched.
+    // the existing TCP forwarding paths are untouched.
-    #[cfg(unix)]
+    let mut agent_writers: HashMap<u64, AgentWriter> = HashMap::new();
-    let mut agent_writers: HashMap<u64, tokio::net::unix::OwnedWriteHalf> = HashMap::new();
     let mut pending_socks_replies: HashSet<u64> = HashSet::new();
     let mut opened_streams: HashSet<u64> = HashSet::new();
     let mut stream_send_credit: HashMap<u64, usize> = HashMap::new();
@@ -6629,21 +6667,6 @@ 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))]
-                            {
-                                send_stream_open_reject(
-                                    &socket,
-                                    addr,
-                                    &cred,
-                                    &mut send_seq,
-                                    open.stream_id,
-                                "agent forwarding is not supported on this client platform"
-                                        .to_string(),
-                                )
-                                .await?;
-                            }
-                            #[cfg(unix)]
-                            {
                             let Some(sock_path) = agent_sock.clone() else {
                                 send_stream_open_reject(
                                     &socket,
@@ -6656,10 +6679,10 @@ async fn run_terminal(
                                 .await?;
                                 continue;
                             };
-                                match UnixStream::connect(&sock_path).await {
+                            match connect_local_agent(&sock_path).await {
                                 Ok(stream) => {
-                                        let (mut reader, writer) = stream.into_split();
+                                    let (mut reader, writer) = tokio::io::split(stream);
-                                        agent_writers.insert(open.stream_id, writer);
+                                    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);
@@ -6723,7 +6746,6 @@ async fn run_terminal(
                                     .await?;
                                 }
                             }
-                            }
                             continue;
                         }
                         match TcpStream::connect((open.target_host.as_str(), open.target_port)).await {
@@ -6886,14 +6908,11 @@ async fn run_terminal(
                             for bytes in &writes {
                                 let _ = writer.write_all(bytes).await;
                             }
-                        } else {
+                        } else if let Some(writer) = agent_writers.get_mut(&stream_id) {
-                            #[cfg(unix)]
-                            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
                         // writer is already gone cannot wedge the server's send window.
                         send_stream_window_adjust(
@@ -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,
+        render_status_clear, render_status_overlay, requested_env, resolve_forward_agent_endpoint,
-        retire_stream_state, retransmit_stream_closes, retransmit_stream_eofs,
+        resolved_startup_command, retire_stream_state, retransmit_stream_closes,
-        retransmit_stream_opens, retransmit_stream_window_adjusts, rewrite_forward_command,
+        retransmit_stream_eofs, retransmit_stream_opens, retransmit_stream_window_adjusts,
-        sanitize_trace_name, selected_predict_mode, selected_udp_host, send_stream_eof,
+        rewrite_forward_command, sanitize_trace_name, selected_predict_mode, selected_udp_host,
-        server_version_mismatch, should_flush_terminal_input_after_contact,
+        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;