Add Windows client packaging and recovery tools

src/bin/dosh-client.rs

@@ -40,8 +40,9 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixStream};
-use tokio::signal::unix::{SignalKind, signal};
+#[cfg(unix)]
+use tokio::net::UnixStream;
+use tokio::net::{TcpListener, TcpStream, UdpSocket};
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
@@ -200,6 +201,9 @@ async fn main() -> Result<()> {
     if args.server.as_deref() == Some("selftest") {
         return run_selftest_command(&config, &args).await;
     }
+    if matches!(args.server.as_deref(), Some("recover" | "repair")) {
+        return run_recover_command(&config, &args).await;
+    }
     if args.server.as_deref() == Some("forward") {
         args = rewrite_forward_command(args)?;
     }
@@ -692,6 +696,19 @@ async fn run_doctor_command(config: &dosh::config::ClientConfig, args: &Args) ->
     run_doctor_for_host(config, args, &requested).await
 }
 
+async fn run_recover_command(config: &dosh::config::ClientConfig, args: &Args) -> Result<()> {
+    if args.command.len() != 1 {
+        return Err(anyhow!("usage: dosh recover <host>"));
+    }
+    let requested = args.command[0].clone();
+    println!("Dosh recovery for {requested}");
+    let removed = clear_cached_credentials(&config.credential_cache, &requested)
+        .context("clear cached attach credentials")?;
+    println!("[ok] removed {removed} cached credential(s)");
+    println!("[info] host trust was left unchanged");
+    run_doctor_for_host(config, args, &requested).await
+}
+
 async fn run_doctor_for_host(
     config: &dosh::config::ClientConfig,
     args: &Args,
@@ -2573,6 +2590,9 @@ 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 {
@@ -2589,7 +2609,9 @@ async fn run_terminal(
     // instead of waiting up to one `resize_tick`. The 250ms poll below stays as a
     // fallback for environments where the signal doesn't fire. `signal()` can fail
     // (rare), in which case we rely on the poll.
-    let mut winch = signal(SignalKind::window_change()).ok();
+    #[cfg(unix)]
+    let mut winch =
+        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok();
     let mut frame_buffer = FrameBuffer::default();
     // Resolve the prediction display policy (off / experimental / always). An
     // env var wins for ad-hoc tuning; otherwise the client config's
@@ -2618,6 +2640,7 @@ async fn run_terminal(
     // 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<u64, tokio::net::unix::OwnedWriteHalf> = HashMap::new();
     let mut pending_socks_replies: HashSet<u64> = HashSet::new();
     let mut opened_streams: HashSet<u64> = HashSet::new();
@@ -2689,10 +2712,13 @@ async fn run_terminal(
                 }
             }
             _ = async {
+                #[cfg(unix)]
                 match winch.as_mut() {
                     Some(w) => { w.recv().await; }
                     None => std::future::pending::<()>().await,
                 }
+                #[cfg(not(unix))]
+                std::future::pending::<()>().await
             } => {
                 maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?;
             }
@@ -2838,53 +2864,89 @@ 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 {
-                            let Some(sock_path) = agent_sock.clone() else {
+                            #[cfg(not(unix))]
+                            {
                                 send_stream_open_reject(
-                                    &socket, addr, &cred, &mut send_seq, open.stream_id,
-                                    "agent forwarding not enabled by client".to_string(),
-                                ).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];
-                                        loop {
-                                            match reader.read(&mut buf).await {
-                                                Ok(0) => 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(ForwardEvent::Close {
-                                            stream_id: open.stream_id,
-                                        }).await;
-                                    });
-                                }
-                                Err(err) => {
+                                    &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, addr, &cred, &mut send_seq, open.stream_id,
-                                        format!("connect local agent: {err}"),
-                                    ).await?;
+                                        &socket,
+                                        addr,
+                                        &cred,
+                                        &mut send_seq,
+                                        open.stream_id,
+                                        "agent forwarding not enabled by client".to_string(),
+                                    )
+                                    .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];
+                                            loop {
+                                                match reader.read(&mut buf).await {
+                                                    Ok(0) => 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(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;
@@ -3003,9 +3065,12 @@ async fn run_terminal(
                             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;
+                        } else {
+                            #[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
@@ -3063,6 +3128,7 @@ async fn run_terminal(
                         stream_next_recv_offset.remove(&close.stream_id);
                         stream_recv_pending.remove(&close.stream_id);
                         stream_writers.remove(&close.stream_id);
+                        #[cfg(unix)]
                         agent_writers.remove(&close.stream_id);
                     }
                     _ => {}
@@ -3114,6 +3180,7 @@ async fn run_terminal(
                         stream_next_recv_offset.remove(&stream_id);
                         stream_recv_pending.remove(&stream_id);
                         stream_writers.remove(&stream_id);
+                        #[cfg(unix)]
                         agent_writers.remove(&stream_id);
                         send_stream_close(&socket, addr, &cred, &mut send_seq, stream_id).await?;
                     }
@@ -4607,11 +4674,46 @@ fn emit_status(_bytes: &[u8]) -> Result<()> {
 }
 
 fn cache_path(root: &str, server: &str, session: &str, mode: &str) -> std::path::PathBuf {
-    let safe = format!("{server}_{session}_{mode}")
+    let safe = cache_key(server, session, mode);
+    expand_tilde(root).join(format!("{safe}.bin"))
+}
+
+fn cache_key(server: &str, session: &str, mode: &str) -> String {
+    format!("{server}_{session}_{mode}")
         .chars()
         .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
-        .collect::<String>();
-    expand_tilde(root).join(format!("{safe}.bin"))
+        .collect::<String>()
+}
+
+fn cache_server_prefix(server: &str) -> String {
+    format!("{server}_")
+        .chars()
+        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
+        .collect::<String>()
+}
+
+fn clear_cached_credentials(root: &str, server: &str) -> Result<usize> {
+    let dir = expand_tilde(root);
+    let Ok(entries) = fs::read_dir(&dir) else {
+        return Ok(0);
+    };
+    let prefix = cache_server_prefix(server);
+    let mut removed = 0usize;
+    for entry in entries {
+        let entry = entry?;
+        let path = entry.path();
+        if !path.is_file() {
+            continue;
+        }
+        let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
+            continue;
+        };
+        if name.starts_with(&prefix) && name.ends_with(".bin") {
+            fs::remove_file(&path)?;
+            removed += 1;
+        }
+    }
+    Ok(removed)
 }
 
 fn load_cache(path: &std::path::Path) -> Result<CachedCredential> {
@@ -4665,18 +4767,19 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 mod tests {
     use super::{
         DisconnectStatus, DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor,
-        RemoteForward, SshConfig, StatusAction, auth_allows, ensure_tui_safe_status_overlay,
-        latest_release_download_url, load_first_native_identity_with_prompt, parse_dynamic_forward,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, raw_contains_host_table,
-        recv_response_until, render_status_clear, render_status_overlay, requested_env,
-        resolved_startup_command, rewrite_forward_command, ssh_destination_host, ssh_username,
-        ssh_with_user, startup_command, toml_bare_key_or_quoted, update_check_requested,
-        valid_forward_host,
+        RemoteForward, SshConfig, StatusAction, auth_allows, cache_key, cache_server_prefix,
+        clear_cached_credentials, ensure_tui_safe_status_overlay, latest_release_download_url,
+        load_first_native_identity_with_prompt, parse_dynamic_forward, parse_local_forward,
+        parse_remote_forward, parse_ssh_config, raw_contains_host_table, recv_response_until,
+        render_status_clear, render_status_overlay, requested_env, resolved_startup_command,
+        rewrite_forward_command, ssh_destination_host, ssh_username, ssh_with_user,
+        startup_command, toml_bare_key_or_quoted, update_check_requested, valid_forward_host,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
     use dosh::protocol::{self, Frame, PacketKind};
     use ed25519_dalek::{SigningKey, VerifyingKey};
+    use std::fs;
     use std::time::Duration;
 
     fn test_args(server: &str, command: &[&str]) -> super::Args {
@@ -5452,6 +5555,30 @@ mod tests {
         assert!(valid_forward_host("server.example.com"));
     }
 
+    #[test]
+    fn recover_removes_only_matching_cached_credentials() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path().display().to_string();
+        let matching = dir
+            .path()
+            .join(format!("{}.bin", cache_key("user@host", "default", "rw")));
+        let other = dir
+            .path()
+            .join(format!("{}.bin", cache_key("other", "default", "rw")));
+        let similar = dir
+            .path()
+            .join(format!("{}.bin", cache_key("user@host2", "default", "rw")));
+        fs::write(&matching, b"cached").unwrap();
+        fs::write(&other, b"cached").unwrap();
+        fs::write(&similar, b"cached").unwrap();
+
+        assert_eq!(cache_server_prefix("user@host"), "user_host_");
+        assert_eq!(clear_cached_credentials(&root, "user@host").unwrap(), 1);
+        assert!(!matching.exists());
+        assert!(other.exists());
+        assert!(similar.exists());
+    }
+
     // --- Item 1: disconnect status line state machine ---
 
     /// The status line stays hidden while the link is fresh and only appears once