Expand SSH identity path tokens

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "1.0.0-rc22"
+version = "1.0.0-rc28"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc22"
+version = "1.0.0-rc28"
 edition = "2024"
 license = "MIT"
 

README.md

@@ -108,6 +108,10 @@ dosh_host = "prod.example.com"
 port = 50000
 ```
 
+`ProxyJump` and `ProxyCommand` can be used for SSH setup, but the Dosh UDP
+endpoint still has to be reachable directly from the client. Set `dosh_host` to
+the reachable UDP hostname or IP when it differs from the SSH alias target.
+
 ## Rust Library
 
 Dosh exposes a Rust transport for encrypted, reconnecting application streams.

src/bin/dosh-client.rs

@@ -36,7 +36,10 @@ use dosh::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     TransportEvent,
 };
-use dosh::udp::{is_transient_udp_error, is_transient_udp_send_error};
+use dosh::udp::{
+    is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+    send_udp_retrying_transient,
+};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::BTreeMap;
@@ -957,6 +960,14 @@ async fn run_doctor_for_host(
             }
         };
     println!("[info] native endpoint: {udp_host}:{dosh_port}");
+    if let Some(warning) = native_proxy_udp_warning(
+        &parsed_ssh_config,
+        requested_udp_host.as_deref(),
+        &udp_host,
+        dosh_port,
+    ) {
+        println!("[warn] native endpoint: {warning}");
+    }
     match resolve_addr(&udp_host, dosh_port) {
         Ok(addr) => println!("[ok] udp resolve: {addr}"),
         Err(err) => println!("[fail] udp resolve: {err:#}"),
@@ -1855,6 +1866,30 @@ impl FileServiceClient {
             other => bail!("unexpected mkdir response: {other:?}"),
         }
     }
+
+    fn readlink(&mut self, path: &str) -> Result<String> {
+        self.send(FileRequest::Readlink {
+            path: path.to_string(),
+        })?;
+        match self.recv()? {
+            FileResponse::LinkTarget { target } => Ok(target),
+            FileResponse::Error { message } => Err(anyhow!(message)),
+            other => bail!("unexpected readlink response: {other:?}"),
+        }
+    }
+
+    fn symlink(&mut self, path: &str, target: &str, overwrite: bool) -> Result<()> {
+        self.send(FileRequest::Symlink {
+            path: path.to_string(),
+            target: target.to_string(),
+            overwrite,
+        })?;
+        match self.recv()? {
+            FileResponse::Ok => Ok(()),
+            FileResponse::Error { message } => Err(anyhow!(message)),
+            other => bail!("unexpected symlink response: {other:?}"),
+        }
+    }
 }
 
 fn run_cp_command(_config: &dosh::config::ClientConfig, args: &Args) -> Result<()> {
@@ -2186,6 +2221,18 @@ fn upload_path(
     let metadata =
         fs::symlink_metadata(local).with_context(|| format!("stat {}", local.display()))?;
     let remote = upload_destination(client, local, remote, metadata.is_dir())?;
+    if metadata.file_type().is_symlink() {
+        let target =
+            fs::read_link(local).with_context(|| format!("readlink {}", local.display()))?;
+        let target = target
+            .to_str()
+            .ok_or_else(|| anyhow!("symlink target is not valid UTF-8: {}", local.display()))?;
+        client.symlink(&remote, target, opts.overwrite)?;
+        if opts.progress {
+            eprintln!("linked {} -> {}", remote, target);
+        }
+        return Ok(());
+    }
     if metadata.is_dir() {
         anyhow::ensure!(
             opts.recursive,
@@ -2255,8 +2302,11 @@ fn upload_file(
         overwrite: opts.overwrite,
         resume: opts.resume,
     })?;
-    let offset = match client.recv()? {
-        FileResponse::Resume { offset } => offset.min(size),
+    let (offset, prefix_sha256) = match client.recv()? {
+        FileResponse::Resume {
+            offset,
+            prefix_sha256,
+        } => (offset.min(size), prefix_sha256),
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected upload start response: {other:?}"),
     };
@@ -2264,6 +2314,7 @@ fn upload_file(
     let mut hasher = Sha256::new();
     if offset > 0 {
         hash_prefix(&mut file, offset, &mut hasher)?;
+        ensure_resume_prefix_matches(prefix_sha256, Some(digest_prefix(&hasher)), offset, remote)?;
     }
     file.seek(std::io::SeekFrom::Start(offset))?;
     let mut sent = offset;
@@ -2328,8 +2379,9 @@ fn download_path(
             Ok(())
         }
         FileKind::File => download_file(client, remote, &destination, &meta, opts),
+        FileKind::Symlink => download_symlink(client, remote, &destination, opts),
         _ => Err(anyhow!(
-            "remote path is not a regular file or directory: {remote}"
+            "remote path is not a regular file, directory, or symlink: {remote}"
         )),
     }
 }
@@ -2362,8 +2414,16 @@ fn copy_remote_path(
             Ok(())
         }
         FileKind::File => copy_remote_file(src_client, dst_client, src, &dst, &meta, opts),
+        FileKind::Symlink => {
+            let target = src_client.readlink(src)?;
+            dst_client.symlink(&dst, &target, opts.overwrite)?;
+            if opts.progress {
+                eprintln!("linked {dst} -> {target}");
+            }
+            Ok(())
+        }
         _ => Err(anyhow!(
-            "remote path is not a regular file or directory: {src}"
+            "remote path is not a regular file, directory, or symlink: {src}"
         )),
     }
 }
@@ -2404,8 +2464,11 @@ fn copy_remote_file(
         overwrite: opts.overwrite,
         resume: opts.resume,
     })?;
-    let offset = match dst_client.recv()? {
-        FileResponse::Resume { offset } => offset.min(meta.len),
+    let (offset, dst_prefix_sha256) = match dst_client.recv()? {
+        FileResponse::Resume {
+            offset,
+            prefix_sha256,
+        } => (offset.min(meta.len), prefix_sha256),
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected remote copy start response: {other:?}"),
     };
@@ -2414,7 +2477,17 @@ fn copy_remote_file(
         offset,
     })?;
     match src_client.recv()? {
-        FileResponse::Start { .. } => {}
+        FileResponse::Start {
+            offset: start_offset,
+            prefix_sha256,
+            ..
+        } => {
+            anyhow::ensure!(
+                start_offset == offset,
+                "remote copy source offset mismatch: got {start_offset}, expected {offset}"
+            );
+            ensure_matching_resume_prefixes(dst_prefix_sha256, prefix_sha256, offset, dst)?;
+        }
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected remote copy source start response: {other:?}"),
     }
@@ -2487,20 +2560,22 @@ fn download_file(
     {
         fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
     }
-    let offset = if opts.resume && local.exists() {
-        fs::metadata(local)?.len().min(meta.len)
+    let original_len = if local.exists() {
+        Some(fs::metadata(local)?.len())
+    } else {
+        None
+    };
+    let offset = if opts.resume {
+        original_len.unwrap_or(0).min(meta.len)
     } else {
         0
     };
     let mut file = if offset > 0 {
-        let file = fs::OpenOptions::new()
+        fs::OpenOptions::new()
             .read(true)
             .append(true)
             .open(local)
-            .with_context(|| format!("open {}", local.display()))?;
-        file.set_len(offset)
-            .with_context(|| format!("truncate {}", local.display()))?;
-        file
+            .with_context(|| format!("open {}", local.display()))?
     } else {
         fs::OpenOptions::new()
             .create(true)
@@ -2510,16 +2585,29 @@ fn download_file(
             .with_context(|| format!("create {}", local.display()))?
     };
     let mut hasher = Sha256::new();
-    if offset > 0 {
+    let local_prefix_sha256 = if offset > 0 {
         let mut prefix = fs::File::open(local)?;
         hash_prefix(&mut prefix, offset, &mut hasher)?;
-    }
+        Some(digest_prefix(&hasher))
+    } else {
+        None
+    };
     client.send(FileRequest::Get {
         path: remote.to_string(),
         offset,
     })?;
     match client.recv()? {
-        FileResponse::Start { .. } => {}
+        FileResponse::Start {
+            offset: start_offset,
+            prefix_sha256,
+            ..
+        } => {
+            anyhow::ensure!(
+                start_offset == offset,
+                "download offset mismatch: got {start_offset}, expected {offset}"
+            );
+            ensure_resume_prefix_matches(prefix_sha256, local_prefix_sha256, offset, remote)?;
+        }
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected download start response: {other:?}"),
     }
@@ -2538,8 +2626,16 @@ fn download_file(
             FileResponse::Done { sha256, bytes } => {
                 file.flush()?;
                 let digest: [u8; 32] = hasher.finalize().into();
+                if (sha256 != digest || bytes != meta.len)
+                    && opts.resume
+                    && let Some(original_len) = original_len
+                {
+                    let _ = file.set_len(original_len);
+                }
                 anyhow::ensure!(sha256 == digest, "download checksum mismatch for {remote}");
                 anyhow::ensure!(bytes == meta.len, "download size mismatch for {remote}");
+                file.set_len(meta.len)
+                    .with_context(|| format!("truncate {}", local.display()))?;
                 set_local_mode(local, meta.mode, opts.preserve)?;
                 set_local_mtime(local, meta.modified_secs, opts.preserve)?;
                 if opts.progress {
@@ -2558,6 +2654,45 @@ fn download_file(
     }
 }
 
+fn download_symlink(
+    client: &mut FileServiceClient,
+    remote: &str,
+    local: &Path,
+    opts: &CpOptions,
+) -> Result<()> {
+    if let Some(parent) = local.parent()
+        && !parent.as_os_str().is_empty()
+    {
+        fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
+    }
+    let target = client.readlink(remote)?;
+    create_local_symlink(local, &target, opts.overwrite)?;
+    if opts.progress {
+        eprintln!("linked {} -> {}", local.display(), target);
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn create_local_symlink(path: &Path, target: &str, overwrite: bool) -> Result<()> {
+    if let Ok(metadata) = fs::symlink_metadata(path) {
+        anyhow::ensure!(overwrite, "destination exists: {}", path.display());
+        anyhow::ensure!(
+            !metadata.is_dir(),
+            "destination is a directory: {}",
+            path.display()
+        );
+        fs::remove_file(path).with_context(|| format!("remove {}", path.display()))?;
+    }
+    std::os::unix::fs::symlink(target, path)
+        .with_context(|| format!("symlink {} -> {}", path.display(), target))
+}
+
+#[cfg(not(unix))]
+fn create_local_symlink(_path: &Path, _target: &str, _overwrite: bool) -> Result<()> {
+    bail!("symlink creation is not supported on this client platform")
+}
+
 fn hash_prefix(file: &mut fs::File, bytes: u64, hasher: &mut Sha256) -> Result<()> {
     file.seek(std::io::SeekFrom::Start(0))?;
     let mut remaining = bytes;
@@ -2574,6 +2709,46 @@ fn hash_prefix(file: &mut fs::File, bytes: u64, hasher: &mut Sha256) -> Result<(
     Ok(())
 }
 
+fn digest_prefix(hasher: &Sha256) -> [u8; 32] {
+    hasher.clone().finalize().into()
+}
+
+fn ensure_resume_prefix_matches(
+    remote_prefix_sha256: Option<[u8; 32]>,
+    local_prefix_sha256: Option<[u8; 32]>,
+    offset: u64,
+    path: &str,
+) -> Result<()> {
+    if offset == 0 {
+        return Ok(());
+    }
+    let Some(remote_prefix_sha256) = remote_prefix_sha256 else {
+        bail!("resume prefix digest missing for {path}");
+    };
+    let Some(local_prefix_sha256) = local_prefix_sha256 else {
+        bail!("local resume prefix digest missing for {path}");
+    };
+    anyhow::ensure!(
+        remote_prefix_sha256 == local_prefix_sha256,
+        "resume prefix mismatch for {path}; retry without --resume or remove the partial destination"
+    );
+    Ok(())
+}
+
+fn ensure_matching_resume_prefixes(
+    destination_prefix_sha256: Option<[u8; 32]>,
+    source_prefix_sha256: Option<[u8; 32]>,
+    offset: u64,
+    path: &str,
+) -> Result<()> {
+    ensure_resume_prefix_matches(
+        source_prefix_sha256,
+        destination_prefix_sha256,
+        offset,
+        path,
+    )
+}
+
 fn path_ends_separator(path: &Path) -> bool {
     let raw = path.as_os_str().to_string_lossy();
     raw.ends_with('/') || raw.ends_with('\\')
@@ -3529,6 +3704,25 @@ fn selected_udp_host(
     }
 }
 
+fn ssh_config_uses_proxy(ssh_config: &SshConfig) -> bool {
+    ssh_config.proxy_jump.is_some() || ssh_config.proxy_command.is_some()
+}
+
+fn native_proxy_udp_warning(
+    ssh_config: &SshConfig,
+    requested_udp_host: Option<&str>,
+    udp_host: &str,
+    dosh_port: u16,
+) -> Option<String> {
+    if ssh_config_uses_proxy(ssh_config) && requested_udp_host.is_none() {
+        Some(format!(
+            "SSH uses ProxyJump/ProxyCommand, but Dosh UDP still needs a directly reachable host; set dosh_host if {udp_host}:{dosh_port} is not reachable from this client"
+        ))
+    } else {
+        None
+    }
+}
+
 fn ssh_username(server: &str) -> Option<String> {
     let (user, _) = server.rsplit_once('@')?;
     if user.is_empty() {
@@ -3789,14 +3983,11 @@ async fn try_native_auth(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3841,7 +4032,7 @@ async fn try_native_auth(
     let auth = sign_native_user_auth(
         config,
         cli_identity,
-        ssh_config,
+        &NativeIdentityContext::new(server, ssh_port, ssh_config),
         &hello,
         &server_hello.hello,
         requested_forwardings,
@@ -3859,13 +4050,9 @@ async fn try_native_auth(
         CLIENT_TO_SERVER,
         &auth_body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3947,14 +4134,11 @@ async fn try_native_auth_check(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3999,7 +4183,7 @@ async fn try_native_auth_check(
     let auth = sign_native_user_auth(
         config,
         cli_identity,
-        ssh_config,
+        &NativeIdentityContext::new(server, ssh_port, ssh_config),
         &hello,
         &server_hello.hello,
         Vec::new(),
@@ -4016,13 +4200,9 @@ async fn try_native_auth_check(
         CLIENT_TO_SERVER,
         &protocol::to_body(&NativeUserAuthBody { auth })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthCheckOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -4038,13 +4218,14 @@ async fn try_native_auth_check(
 fn sign_native_user_auth(
     config: &dosh::config::ClientConfig,
     cli_identity: Option<&Path>,
-    ssh_config: &SshConfig,
+    identity_context: &NativeIdentityContext<'_>,
     hello: &dosh::native::NativeClientHello,
     server_hello: &dosh::native::NativeServerHello,
     requested_forwardings: Vec<ForwardingRequest>,
     allow_passphrase_prompt: bool,
 ) -> Result<dosh::native::NativeUserAuth> {
     let mut errors = Vec::new();
+    let ssh_config = identity_context.ssh_config;
     if config.use_ssh_agent && !ssh_config.identities_only {
         match ssh_agent::sign_user_auth_with_agent(
             hello,
@@ -4059,9 +4240,9 @@ fn sign_native_user_auth(
     }
 
     let identity = if allow_passphrase_prompt {
-        load_first_native_identity(config, cli_identity, ssh_config)
+        load_first_native_identity(config, cli_identity, identity_context)
     } else {
-        load_first_native_identity_noninteractive(config, cli_identity, ssh_config)
+        load_first_native_identity_noninteractive(config, cli_identity, identity_context)
     };
     match identity {
         Ok(identity) => {
@@ -4080,12 +4261,12 @@ fn sign_native_user_auth(
 fn load_first_native_identity(
     config: &dosh::config::ClientConfig,
     cli_identity: Option<&Path>,
-    ssh_config: &SshConfig,
+    identity_context: &NativeIdentityContext<'_>,
 ) -> Result<ssh_key::PrivateKey> {
     load_first_native_identity_with_prompt(
         config,
         cli_identity,
-        ssh_config,
+        identity_context,
         prompt_identity_passphrase,
     )
 }
@@ -4093,9 +4274,9 @@ fn load_first_native_identity(
 fn load_first_native_identity_noninteractive(
     config: &dosh::config::ClientConfig,
     cli_identity: Option<&Path>,
-    ssh_config: &SshConfig,
+    identity_context: &NativeIdentityContext<'_>,
 ) -> Result<ssh_key::PrivateKey> {
-    load_first_native_identity_with_prompt(config, cli_identity, ssh_config, |path| {
+    load_first_native_identity_with_prompt(config, cli_identity, identity_context, |path| {
         Err(anyhow!(
             "{} is encrypted; unlock it in ssh-agent or use an unencrypted key for proxy-stdio",
             path.display()
@@ -4106,22 +4287,30 @@ fn load_first_native_identity_noninteractive(
 fn load_first_native_identity_with_prompt<F>(
     config: &dosh::config::ClientConfig,
     cli_identity: Option<&Path>,
-    ssh_config: &SshConfig,
+    identity_context: &NativeIdentityContext<'_>,
     mut prompt: F,
 ) -> Result<ssh_key::PrivateKey>
 where
     F: FnMut(&Path) -> Result<String>,
 {
+    let ssh_config = identity_context.ssh_config;
+    let token_context = &identity_context.path_tokens;
     let mut paths = Vec::new();
     if let Some(path) = cli_identity {
         push_identity_path(&mut paths, path.to_path_buf());
     }
     for path in &ssh_config.identity_files {
-        push_identity_path(&mut paths, expand_tilde(path));
+        push_identity_path(
+            &mut paths,
+            expand_tilde(&expand_ssh_path_tokens(path, token_context)),
+        );
     }
     if !ssh_config.identities_only {
         for path in &config.identity_files {
-            push_identity_path(&mut paths, expand_tilde(path));
+            push_identity_path(
+                &mut paths,
+                expand_tilde(&expand_ssh_path_tokens(path, token_context)),
+            );
         }
     }
 
@@ -4159,6 +4348,98 @@ fn push_identity_path(paths: &mut Vec<PathBuf>, path: PathBuf) {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct SshPathTokenContext {
+    original_host: String,
+    hostname: String,
+    port: u16,
+    remote_user: String,
+    local_user: String,
+    home_dir: Option<String>,
+}
+
+struct NativeIdentityContext<'a> {
+    ssh_config: &'a SshConfig,
+    path_tokens: SshPathTokenContext,
+}
+
+impl<'a> NativeIdentityContext<'a> {
+    fn new(server: &str, ssh_port: Option<u16>, ssh_config: &'a SshConfig) -> Self {
+        Self {
+            ssh_config,
+            path_tokens: ssh_path_token_context(server, ssh_port, ssh_config),
+        }
+    }
+}
+
+fn ssh_path_token_context(
+    server: &str,
+    ssh_port: Option<u16>,
+    ssh_config: &SshConfig,
+) -> SshPathTokenContext {
+    let original_host = ssh_destination_host(server);
+    let hostname = ssh_config
+        .hostname
+        .clone()
+        .unwrap_or_else(|| original_host.clone());
+    let port = ssh_config.port.or(ssh_port).unwrap_or(22);
+    let remote_user = ssh_username(server)
+        .or(ssh_config.user.clone())
+        .unwrap_or_else(local_username);
+    let local_user = local_username();
+    let home_dir = dirs::home_dir().map(|path| path.to_string_lossy().to_string());
+    SshPathTokenContext {
+        original_host,
+        hostname,
+        port,
+        remote_user,
+        local_user,
+        home_dir,
+    }
+}
+
+fn local_username() -> String {
+    std::env::var("USER")
+        .or_else(|_| std::env::var("USERNAME"))
+        .unwrap_or_else(|_| "unknown".to_string())
+}
+
+fn expand_ssh_path_tokens(path: &str, context: &SshPathTokenContext) -> String {
+    let mut out = String::with_capacity(path.len());
+    let mut chars = path.chars();
+    while let Some(ch) = chars.next() {
+        if ch != '%' {
+            out.push(ch);
+            continue;
+        }
+        let Some(token) = chars.next() else {
+            out.push('%');
+            break;
+        };
+        match token {
+            '%' => out.push('%'),
+            'd' => {
+                if let Some(home_dir) = &context.home_dir {
+                    out.push_str(home_dir);
+                } else {
+                    out.push('%');
+                    out.push(token);
+                }
+            }
+            'h' => out.push_str(&context.hostname),
+            'n' => out.push_str(&context.original_host),
+            'p' => out.push_str(&context.port.to_string()),
+            'r' => out.push_str(&context.remote_user),
+            'u' => out.push_str(&context.local_user),
+            other => {
+                out.push('%');
+                out.push(other);
+            }
+        }
+    }
+    out
+}
+
 fn prompt_identity_passphrase(path: &Path) -> Result<String> {
     rpassword::prompt_password(format!("Enter passphrase for {}: ", path.display()))
         .with_context(|| format!("read passphrase for {}", path.display()))
@@ -4182,7 +4463,7 @@ async fn bootstrap_attach(
     let body = protocol::to_body(&req)?;
     let packet =
         protocol::encode_plain(PacketKind::BootstrapAttachRequest, [0u8; 16], 1, 0, &body)?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_secs(5)).await?;
     let expected_key_id = protocol::session_key_id(&bootstrap.session_key);
     let ok = recv_response_until(socket, Duration::from_secs(5), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4262,7 +4543,7 @@ async fn try_ticket_attach(
         0,
         &protocol::to_body(&envelope)?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
 
     let ok = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4338,7 +4619,7 @@ async fn try_resume_fresh_process(
         CLIENT_TO_SERVER,
         &body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4393,7 +4674,7 @@ async fn try_live_resume(
         &body,
     )?;
     *send_seq += 1;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4440,14 +4721,9 @@ where
             return Err(anyhow!("timed out waiting for server response"));
         }
         let remaining = wait - elapsed;
-        let (n, _) = match tokio::time::timeout(remaining, socket.recv_from(&mut buf))
+        let (n, _) = recv_udp_retrying_transient(socket, &mut buf, remaining)
             .await
-            .context("timed out waiting for server response")?
-        {
-            Ok(value) => value,
-            Err(err) if is_transient_udp_error(&err) => continue,
-            Err(err) => return Err(err.into()),
-        };
+            .context("timed out waiting for server response")?;
         let Ok(packet) = protocol::decode(&buf[..n]) else {
             continue;
         };
@@ -6269,7 +6545,9 @@ async fn flush_stream_pending_data(
         if credit < bytes.len() {
             break;
         }
-        let bytes = pending.pop_front().expect("pending front exists");
+        let Some(bytes) = pending.pop_front() else {
+            break;
+        };
         *stream_send_credit.entry(stream_id).or_default() -= bytes.len();
         let offset = *stream_next_send_offset.entry(stream_id).or_default();
         stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));
@@ -7764,22 +8042,23 @@ mod tests {
     use super::{
         ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward,
         FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
-        POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode, Predictor,
-        RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig, StartupGateMode,
-        StatusAction, UpdateOptions, UpdateRole, auth_allows, cache_key, cache_server_prefix,
-        cleanup_stream_state, clear_cached_credentials, ensure_tui_safe_status_overlay,
-        input_contains_focus_in, input_matches_escape, is_local_status_target,
-        is_resume_response_for_client, latest_release_download_url,
-        load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, parse_update_options,
-        post_submit_hold_duration, queue_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_status_clear,
-        render_status_overlay, requested_env, resolved_startup_command, retire_stream_state,
-        retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
-        server_version_mismatch, should_flush_terminal_input_after_contact,
-        should_hold_during_startup_gate, should_hold_post_submit_input,
-        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_command_target,
+        NativeIdentityContext, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode,
+        Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig,
+        SshPathTokenContext, StartupGateMode, StatusAction, UpdateOptions, UpdateRole, auth_allows,
+        cache_key, cache_server_prefix, cleanup_stream_state, clear_cached_credentials,
+        ensure_tui_safe_status_overlay, expand_ssh_path_tokens, input_contains_focus_in,
+        input_matches_escape, is_local_status_target, is_resume_response_for_client,
+        latest_release_download_url, load_first_native_identity_with_prompt,
+        native_proxy_udp_warning, parse_dynamic_forward, parse_escape_key, parse_local_forward,
+        parse_remote_forward, parse_ssh_config, parse_update_options, post_submit_hold_duration,
+        queue_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_status_clear, render_status_overlay, requested_env, resolved_startup_command,
+        retire_stream_state, retransmit_stream_opens, rewrite_forward_command,
+        selected_predict_mode, selected_udp_host, server_version_mismatch,
+        should_flush_terminal_input_after_contact, should_hold_during_startup_gate,
+        should_hold_post_submit_input, should_repaint_idle_alternate_screen,
+        split_after_command_submit, ssh_command_target, ssh_config_uses_proxy,
         ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
         strip_stale_mouse_reports, strip_terminal_focus_reports, terminal_private_mode_transition,
         toml_bare_key_or_quoted, unix_update_script, update_installer_url, update_version_status,
@@ -8155,6 +8434,58 @@ mod tests {
         assert!(selected_udp_host(Some("any"), "alias", &ssh).is_err());
     }
 
+    #[test]
+    fn ssh_config_proxy_detection_covers_jump_and_command() {
+        assert!(!ssh_config_uses_proxy(&SshConfig::default()));
+        assert!(ssh_config_uses_proxy(&SshConfig {
+            proxy_jump: Some("bastion".to_string()),
+            ..SshConfig::default()
+        }));
+        assert!(ssh_config_uses_proxy(&SshConfig {
+            proxy_command: Some("ssh bastion -W %h:%p".to_string()),
+            ..SshConfig::default()
+        }));
+    }
+
+    #[test]
+    fn ssh_identity_paths_expand_common_openssh_tokens() {
+        let context = SshPathTokenContext {
+            original_host: "prod".to_string(),
+            hostname: "10.0.0.5".to_string(),
+            port: 2222,
+            remote_user: "deploy".to_string(),
+            local_user: "palav".to_string(),
+            home_dir: Some("/home/palav".to_string()),
+        };
+        assert_eq!(
+            expand_ssh_path_tokens("~/.ssh/%r@%h:%p-%n-%u-%%-%x", &context),
+            "~/.ssh/deploy@10.0.0.5:2222-prod-palav-%-%x"
+        );
+        assert_eq!(
+            expand_ssh_path_tokens("%d/.ssh/%r_%h", &context),
+            "/home/palav/.ssh/deploy_10.0.0.5"
+        );
+    }
+
+    #[test]
+    fn native_proxy_udp_warning_requires_missing_explicit_dosh_host() {
+        let proxied = SshConfig {
+            proxy_jump: Some("bastion".to_string()),
+            ..SshConfig::default()
+        };
+        let warning = native_proxy_udp_warning(&proxied, None, "internal.lan", 50000)
+            .expect("proxied SSH without dosh_host should warn");
+        assert!(warning.contains("ProxyJump/ProxyCommand"));
+        assert!(warning.contains("internal.lan:50000"));
+        assert!(
+            native_proxy_udp_warning(&proxied, Some("public.example.com"), "internal.lan", 50000)
+                .is_none()
+        );
+        assert!(
+            native_proxy_udp_warning(&SshConfig::default(), None, "internal.lan", 50000).is_none()
+        );
+    }
+
     #[test]
     fn parses_user_from_ssh_destination() {
         assert_eq!(ssh_username("alice@example.com"), Some("alice".to_string()));
@@ -8193,12 +8524,13 @@ mod tests {
         write_encrypted_identity(&path, &signing, "let-me-in");
         let mut config = ClientConfig::default();
         config.identity_files.clear();
-        let loaded = load_first_native_identity_with_prompt(
-            &config,
-            Some(&path),
-            &SshConfig::default(),
-            |_| Ok("let-me-in".to_string()),
-        )
+        let ssh_config = SshConfig::default();
+        let identity_context =
+            NativeIdentityContext::new("alice@example.com", Some(2222), &ssh_config);
+        let loaded =
+            load_first_native_identity_with_prompt(&config, Some(&path), &identity_context, |_| {
+                Ok("let-me-in".to_string())
+            })
             .unwrap();
 
         assert_eq!(
@@ -8215,12 +8547,13 @@ mod tests {
         write_encrypted_identity(&path, &signing, "correct");
         let mut config = ClientConfig::default();
         config.identity_files.clear();
-        let err = load_first_native_identity_with_prompt(
-            &config,
-            Some(&path),
-            &SshConfig::default(),
-            |_| Ok("wrong".to_string()),
-        )
+        let ssh_config = SshConfig::default();
+        let identity_context =
+            NativeIdentityContext::new("alice@example.com", Some(2222), &ssh_config);
+        let err =
+            load_first_native_identity_with_prompt(&config, Some(&path), &identity_context, |_| {
+                Ok("wrong".to_string())
+            })
             .unwrap_err();
 
         assert!(err.to_string().contains("no usable native identity"));
@@ -8240,7 +8573,9 @@ mod tests {
             identities_only: true,
             ..SshConfig::default()
         };
-        let err = load_first_native_identity_with_prompt(&config, None, &ssh_config, |_| {
+        let identity_context =
+            NativeIdentityContext::new("alice@example.com", Some(2222), &ssh_config);
+        let err = load_first_native_identity_with_prompt(&config, None, &identity_context, |_| {
             unreachable!("unencrypted key should not prompt")
         })
         .unwrap_err();
@@ -8261,7 +8596,38 @@ mod tests {
             identities_only: true,
             ..SshConfig::default()
         };
-        let loaded = load_first_native_identity_with_prompt(&config, None, &ssh_config, |_| {
+        let identity_context =
+            NativeIdentityContext::new("alice@example.com", Some(2222), &ssh_config);
+        let loaded =
+            load_first_native_identity_with_prompt(&config, None, &identity_context, |_| {
+                unreachable!("unencrypted key should not prompt")
+            })
+            .unwrap();
+
+        assert_eq!(
+            loaded.public_key().key_data().ed25519().unwrap().as_ref(),
+            VerifyingKey::from(&signing).as_bytes()
+        );
+    }
+
+    #[test]
+    fn ssh_config_identity_file_tokens_are_expanded_for_native_auth() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("deploy_10.0.0.5_2222");
+        let signing = SigningKey::from_bytes(&[65u8; 32]);
+        write_identity(&path, &signing);
+        let mut config = ClientConfig::default();
+        config.identity_files.clear();
+        let ssh_config = SshConfig {
+            hostname: Some("10.0.0.5".to_string()),
+            identity_files: vec![format!("{}/%r_%h_%p", dir.path().display())],
+            identities_only: true,
+            port: Some(2222),
+            ..SshConfig::default()
+        };
+        let identity_context = NativeIdentityContext::new("deploy@prod", None, &ssh_config);
+        let loaded =
+            load_first_native_identity_with_prompt(&config, None, &identity_context, |_| {
                 unreachable!("unencrypted key should not prompt")
             })
             .unwrap();

src/bin/dosh-server.rs

@@ -361,6 +361,7 @@ impl NativeAuthRateLimiter {
 struct Session {
     pty: Option<PtyHandle>,
     parser: vt100::Parser,
+    restored_screen: Option<RestoredScreenSnapshot>,
     clients: HashMap<[u8; 16], ClientState>,
     output_seq: u64,
     recent: VecDeque<Vec<u8>>,
@@ -387,6 +388,14 @@ struct Session {
     last_screen_persist_at: Instant,
 }
 
+#[derive(Clone)]
+struct RestoredScreenSnapshot {
+    cols: u16,
+    rows: u16,
+    output_seq: u64,
+    snapshot: Vec<u8>,
+}
+
 #[derive(Clone)]
 struct ClientState {
     endpoint: SocketAddr,
@@ -512,6 +521,7 @@ impl ServerState {
             Session {
                 pty,
                 parser: vt100::Parser::new(rows.max(1), cols.max(1), self.config.scrollback),
+                restored_screen: None,
                 clients: HashMap::new(),
                 output_seq: 0,
                 recent: VecDeque::with_capacity(self.config.scrollback),
@@ -641,15 +651,23 @@ impl ServerState {
                 .unwrap_or((80, 24));
             let mut parser = vt100::Parser::new(rows, cols, self.config.scrollback);
             let mut output_seq = 0;
+            let mut restored_screen = None;
             if let Some(saved) = saved {
                 parser.process(&saved.snapshot);
                 output_seq = saved.output_seq;
+                restored_screen = Some(RestoredScreenSnapshot {
+                    cols,
+                    rows,
+                    output_seq,
+                    snapshot: saved.snapshot,
+                });
             }
             self.sessions.insert(
                 name.clone(),
                 Session {
                     pty: Some(pty),
                     parser,
+                    restored_screen,
                     clients: HashMap::new(),
                     output_seq,
                     recent: VecDeque::with_capacity(self.config.scrollback),
@@ -762,6 +780,19 @@ fn apply_terminal_size(
     Ok(())
 }
 
+fn attach_snapshot(session: &Session, cols: u16, rows: u16) -> Vec<u8> {
+    let cols = cols.max(1);
+    let rows = rows.max(1);
+    if let Some(restored) = &session.restored_screen
+        && restored.output_seq == session.output_seq
+        && restored.cols == cols
+        && restored.rows == rows
+    {
+        return restored.snapshot.clone();
+    }
+    screen_snapshot(session.parser.screen())
+}
+
 fn mode_allows_terminal_updates(mode: &str) -> bool {
     mode != "view-only" && mode != "forward-only"
 }
@@ -1189,7 +1220,7 @@ async fn handle_native_user_auth(
                 apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, cols, rows);
             let output_seq = session.output_seq;
             locked.insert_client(
                 &session_name,
@@ -1337,7 +1368,7 @@ async fn handle_bootstrap_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let bootstrap_session = req.bootstrap.session.clone();
             locked.insert_client(
@@ -1473,7 +1504,7 @@ async fn handle_ticket_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let ticket_session = req.session.clone();
             locked.insert_client(
@@ -1586,6 +1617,7 @@ async fn handle_resume(
             .sessions
             .get_mut(&req.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
+        let (send_seq, allow_terminal_updates) = {
             let client = session
                 .clients
                 .get_mut(&packet.header.conn_id)
@@ -1599,7 +1631,9 @@ async fn handle_resume(
             client.rows = req.rows;
             client.last_seen = Instant::now();
             client.send_seq += 1;
-        if mode_allows_terminal_updates(&client.mode) {
+            (client.send_seq, mode_allows_terminal_updates(&client.mode))
+        };
+        if allow_terminal_updates {
             apply_terminal_size(
                 session.pty.as_ref(),
                 &mut session.parser,
@@ -1607,8 +1641,8 @@ async fn handle_resume(
                 req.rows,
             )?;
         }
-        let snapshot = screen_snapshot(session.parser.screen());
-        (client.send_seq, session.output_seq, snapshot)
+        let snapshot = attach_snapshot(session, req.cols, req.rows);
+        (send_seq, session.output_seq, snapshot)
     };
     let frame = Frame {
         session: session_name,
@@ -2532,8 +2566,10 @@ struct UploadState {
     hasher: Sha256,
     written: u64,
     expected_size: u64,
+    mode: Option<u32>,
     modified_secs: Option<u64>,
     atomic: bool,
+    original_len: Option<u64>,
 }
 
 async fn run_file_stream_service(
@@ -2645,6 +2681,46 @@ async fn handle_file_request(
             send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
                 .await
         }
+        FileRequest::Readlink { path } => {
+            let path = clean_remote_path(&path, home)?;
+            let target =
+                fs::read_link(&path).with_context(|| format!("readlink {}", path.display()))?;
+            let target = target
+                .to_str()
+                .ok_or_else(|| anyhow!("symlink target is not valid UTF-8"))?
+                .to_string();
+            send_file_response_to_client(
+                state,
+                socket,
+                client_id,
+                stream_id,
+                FileResponse::LinkTarget { target },
+            )
+            .await
+        }
+        FileRequest::Symlink {
+            path,
+            target,
+            overwrite,
+        } => {
+            let path = clean_remote_path(&path, home)?;
+            if let Some(parent) = path.parent() {
+                fs::create_dir_all(parent)
+                    .with_context(|| format!("create {}", parent.display()))?;
+            }
+            if let Ok(metadata) = fs::symlink_metadata(&path) {
+                anyhow::ensure!(overwrite, "destination exists: {}", path.display());
+                anyhow::ensure!(
+                    !metadata.is_dir(),
+                    "destination is a directory: {}",
+                    path.display()
+                );
+                fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
+            }
+            create_symlink(&target, &path)?;
+            send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
+                .await
+        }
         FileRequest::Remove { path, recursive } => {
             let path = clean_remote_path(&path, home)?;
             let metadata =
@@ -2682,21 +2758,31 @@ async fn handle_file_request(
                 bail!("destination exists: {}", final_path.display());
             }
             let mut hasher = Sha256::new();
-            let (file, temp_path, written, atomic) = if resume && final_path.exists() {
+            let (file, temp_path, written, atomic, original_len, prefix_sha256) =
+                if resume && final_path.exists() {
+                    let original_len = fs::metadata(&final_path)
+                        .with_context(|| format!("stat {}", final_path.display()))?
+                        .len();
+                    let written = original_len.min(size);
+                    if written > 0 {
+                        let mut prefix = fs::File::open(&final_path)?;
+                        hash_exact_prefix(&mut prefix, written, &mut hasher)?;
+                    }
+                    let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
                     let mut file = fs::OpenOptions::new()
                         .read(true)
                         .append(true)
                         .open(&final_path)
                         .with_context(|| format!("open {}", final_path.display()))?;
-                let written = file.metadata()?.len().min(size);
-                file.set_len(written)
-                    .with_context(|| format!("truncate {}", final_path.display()))?;
-                if written > 0 {
-                    let mut prefix = fs::File::open(&final_path)?;
-                    hash_exact_prefix(&mut prefix, written, &mut hasher)?;
-                }
                     file.seek(SeekFrom::Start(written))?;
-                (file, None, written, false)
+                    (
+                        file,
+                        None,
+                        written,
+                        false,
+                        Some(original_len),
+                        prefix_sha256,
+                    )
                 } else {
                     let temp_path = upload_temp_path(&final_path, stream_id);
                     let file = fs::OpenOptions::new()
@@ -2705,7 +2791,7 @@ async fn handle_file_request(
                         .write(true)
                         .open(&temp_path)
                         .with_context(|| format!("create {}", temp_path.display()))?;
-                (file, Some(temp_path), 0, true)
+                    (file, Some(temp_path), 0, true, None, None)
                 };
             *upload = Some(UploadState {
                 final_path,
@@ -2714,23 +2800,20 @@ async fn handle_file_request(
                 hasher,
                 written,
                 expected_size: size,
+                mode,
                 modified_secs,
                 atomic,
+                original_len,
             });
-            if let Some(mode) = mode {
-                let target = upload
-                    .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
-                    .clone();
-                set_mode(&target, mode)?;
-            }
             send_file_response_to_client(
                 state,
                 socket,
                 client_id,
                 stream_id,
-                FileResponse::Resume { offset: written },
+                FileResponse::Resume {
+                    offset: written,
+                    prefix_sha256,
+                },
             )
             .await
         }
@@ -2752,8 +2835,6 @@ async fn handle_file_request(
             let Some(mut active) = upload.take() else {
                 bail!("no upload in progress");
             };
-            active.file.flush()?;
-            active.file.sync_all()?;
             anyhow::ensure!(
                 active.written == active.expected_size,
                 "upload size mismatch: got {}, expected {}",
@@ -2764,9 +2845,17 @@ async fn handle_file_request(
             if digest != sha256 {
                 if let Some(temp_path) = &active.temp_path {
                     let _ = fs::remove_file(temp_path);
+                } else if let Some(original_len) = active.original_len {
+                    let _ = active.file.set_len(original_len);
                 }
                 bail!("upload checksum mismatch");
             }
+            active
+                .file
+                .set_len(active.expected_size)
+                .with_context(|| format!("truncate {}", active.final_path.display()))?;
+            active.file.flush()?;
+            active.file.sync_all()?;
             drop(active.file);
             if active.atomic
                 && let Some(temp_path) = active.temp_path.take()
@@ -2779,6 +2868,9 @@ async fn handle_file_request(
                     )
                 })?;
             }
+            if let Some(mode) = active.mode {
+                set_mode(&active.final_path, mode)?;
+            }
             if let Some(modified_secs) = active.modified_secs {
                 filetime::set_file_mtime(
                     &active.final_path,
@@ -2806,9 +2898,20 @@ async fn handle_file_request(
                 "remote path is not a file: {}",
                 meta.path
             );
+            anyhow::ensure!(
+                offset <= meta.len,
+                "download offset {offset} is beyond file size {}",
+                meta.len
+            );
             let mut file =
                 fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
             file.seek(SeekFrom::Start(offset))?;
+            let mut hasher = Sha256::new();
+            if offset > 0 {
+                let mut prefix = fs::File::open(&path)?;
+                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
+            }
+            let prefix_sha256 = (offset > 0).then(|| hasher.clone().finalize().into());
             send_file_response_to_client(
                 state,
                 socket,
@@ -2817,14 +2920,10 @@ async fn handle_file_request(
                 FileResponse::Start {
                     meta: meta.clone(),
                     offset,
+                    prefix_sha256,
                 },
             )
             .await?;
-            let mut hasher = Sha256::new();
-            if offset > 0 {
-                let mut prefix = fs::File::open(&path)?;
-                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
-            }
             let mut sent = offset;
             let mut buf = vec![0u8; CHUNK_SIZE];
             loop {
@@ -3006,6 +3105,17 @@ fn set_mode(path: &Path, mode: u32) -> Result<()> {
     fs::set_permissions(path, permissions).with_context(|| format!("chmod {}", path.display()))
 }
 
+#[cfg(unix)]
+fn create_symlink(target: &str, path: &Path) -> Result<()> {
+    std::os::unix::fs::symlink(target, path)
+        .with_context(|| format!("symlink {} -> {}", path.display(), target))
+}
+
+#[cfg(not(unix))]
+fn create_symlink(_target: &str, _path: &Path) -> Result<()> {
+    bail!("symlink creation is not supported on this server platform")
+}
+
 fn upload_temp_path(final_path: &Path, stream_id: u64) -> PathBuf {
     let name = final_path
         .file_name()
@@ -3209,7 +3319,10 @@ async fn flush_stream_pending_data_to_client(
                 if credit < bytes.len() {
                     return Ok(());
                 }
-                let bytes = pending.pop_front().expect("pending front exists");
+                let Some(bytes) = pending.pop_front() else {
+                    client.stream_pending_data.remove(&stream_id);
+                    return Ok(());
+                };
                 if pending.is_empty() {
                     client.stream_pending_data.remove(&stream_id);
                 }
@@ -3454,6 +3567,7 @@ async fn broadcast_output(
             .get_mut(&output.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
         session.parser.process(&output.bytes);
+        session.restored_screen = None;
         session.output_seq += 1;
         let output_seq = session.output_seq;
         session.recent.push_back(output.bytes.clone());
@@ -4049,6 +4163,34 @@ mod tests {
         assert_eq!(ServerConfig::default().client_timeout_secs, 2_592_000);
     }
 
+    #[test]
+    fn attach_snapshot_prefers_restored_screen_for_same_size_reattach() {
+        let restored = b"\x1b[?1049lRESTORED_SCREEN_MARKER".to_vec();
+        let mut session = Session {
+            pty: None,
+            parser: vt100::Parser::new(24, 80, 100),
+            restored_screen: Some(RestoredScreenSnapshot {
+                cols: 80,
+                rows: 24,
+                output_seq: 7,
+                snapshot: restored.clone(),
+            }),
+            clients: HashMap::new(),
+            output_seq: 7,
+            recent: VecDeque::new(),
+            empty_since: Some(Instant::now()),
+            holder_control: None,
+            persistent: true,
+            bytes_since_persist: 0,
+            last_persisted_seq: 7,
+            last_screen_persist_at: Instant::now(),
+        };
+
+        session.parser.set_size(24, 80);
+        assert_eq!(attach_snapshot(&session, 80, 24), restored);
+        assert_ne!(attach_snapshot(&session, 100, 30), restored);
+    }
+
     #[tokio::test]
     async fn unknown_resume_reject_keeps_client_id() {
         let (pty_tx, _rx) = mpsc::unbounded_channel();
@@ -4399,6 +4541,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4463,6 +4606,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4539,6 +4683,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4679,6 +4824,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4776,6 +4922,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {

src/client.rs

@@ -13,6 +13,7 @@ use crate::protocol::{
 };
 use crate::ssh_agent;
 use crate::transport::{DoshTransport, SessionRole, SessionTransportConfig, TransportConfig};
+use crate::udp::{recv_udp_retrying_transient, send_udp_retrying_transient};
 use anyhow::{Context, Result, anyhow, bail};
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::path::PathBuf;
@@ -214,10 +215,10 @@ impl DoshClientBuilder {
                 hello: hello.clone(),
             })?,
         )?;
-        socket.send_to(&packet, peer_addr).await?;
+        send_udp_retrying_transient(&socket, &packet, peer_addr, timeout).await?;
 
         let mut buf = vec![0u8; 65535];
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeServerHello {
             if packet.header.kind == PacketKind::AttachReject {
@@ -261,8 +262,8 @@ impl DoshClientBuilder {
             CLIENT_TO_SERVER,
             &protocol::to_body(&NativeUserAuthBody { auth })?,
         )?;
-        socket.send_to(&auth_packet, peer_addr).await?;
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        send_udp_retrying_transient(&socket, &auth_packet, peer_addr, timeout).await?;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeAuthOk {
             if packet.header.kind == PacketKind::AttachReject {

src/file_transfer.rs

@@ -18,6 +18,14 @@ pub enum FileRequest {
         path: String,
         mode: Option<u32>,
     },
+    Readlink {
+        path: String,
+    },
+    Symlink {
+        path: String,
+        target: String,
+        overwrite: bool,
+    },
     Remove {
         path: String,
         recursive: bool,
@@ -46,13 +54,35 @@ pub enum FileRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum FileResponse {
     Ok,
-    Resume { offset: u64 },
-    Stat { meta: FileMeta },
-    List { entries: Vec<FileEntry> },
-    Start { meta: FileMeta, offset: u64 },
-    Chunk { offset: u64, bytes: Vec<u8> },
-    Done { sha256: [u8; 32], bytes: u64 },
-    Error { message: String },
+    Resume {
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Stat {
+        meta: FileMeta,
+    },
+    List {
+        entries: Vec<FileEntry>,
+    },
+    LinkTarget {
+        target: String,
+    },
+    Start {
+        meta: FileMeta,
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Chunk {
+        offset: u64,
+        bytes: Vec<u8>,
+    },
+    Done {
+        sha256: [u8; 32],
+        bytes: u64,
+    },
+    Error {
+        message: String,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

src/server.rs

@@ -13,7 +13,7 @@ use crate::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     service_name_from_target,
 };
-use crate::udp::is_transient_udp_send_error;
+use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use anyhow::{Context, Result, anyhow, bail};
 use ed25519_dalek::SigningKey;
 use std::collections::{HashMap, HashSet};
@@ -169,7 +169,11 @@ impl DoshServer {
         let mut buf = vec![0u8; 65535];
         loop {
             self.expire_pending();
-            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let (n, peer) = match self.socket.recv_from(&mut buf).await {
+                Ok(value) => value,
+                Err(err) if is_transient_udp_error(&err) => continue,
+                Err(err) => return Err(err.into()),
+            };
             let packet = match protocol::decode(&buf[..n]) {
                 Ok(packet) => packet,
                 Err(_) => continue,

src/transport.rs

@@ -596,11 +596,13 @@ impl StreamMux {
             if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
                 break;
             }
-            let bytes = self
+            let Some(bytes) = self
                 .pending_data
                 .get_mut(&stream_id)
                 .and_then(VecDeque::pop_front)
-                .expect("pending front exists");
+            else {
+                break;
+            };
             if self
                 .pending_data
                 .get(&stream_id)

src/udp.rs

@@ -1,5 +1,9 @@
 //! UDP error classification shared by Dosh clients, servers, and embedders.
 
+use std::net::SocketAddr;
+use std::time::{Duration, Instant};
+use tokio::net::UdpSocket;
+
 pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
     matches!(
         err.kind(),
@@ -13,6 +17,54 @@ pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
     is_transient_udp_error(err)
 }
 
+pub async fn send_udp_retrying_transient(
+    socket: &UdpSocket,
+    packet: &[u8],
+    peer: SocketAddr,
+    wait: Duration,
+) -> std::io::Result<()> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        match socket.send_to(packet, peer).await {
+            Ok(_) => return Ok(()),
+            Err(err) if is_transient_udp_send_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Err(err) => return Err(err),
+        }
+    }
+}
+
+pub async fn recv_udp_retrying_transient(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+    wait: Duration,
+) -> std::io::Result<(usize, SocketAddr)> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        let now = Instant::now();
+        if now >= deadline {
+            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
+        }
+        match tokio::time::timeout(deadline - now, socket.recv_from(buf)).await {
+            Ok(Ok(value)) => return Ok(value),
+            Ok(Err(err)) if is_transient_udp_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Ok(Err(err)) => return Err(err),
+            Err(_) => return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)),
+        }
+    }
+}
+
+async fn sleep_until_retry(deadline: Instant) {
+    let now = Instant::now();
+    if now >= deadline {
+        return;
+    }
+    tokio::time::sleep((deadline - now).min(Duration::from_millis(20))).await;
+}
+
 #[cfg(unix)]
 fn is_transient_udp_os_error(code: i32) -> bool {
     matches!(
@@ -52,7 +104,12 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
 
 #[cfg(test)]
 mod tests {
-    use super::{is_transient_udp_error, is_transient_udp_send_error};
+    use super::{
+        is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+        send_udp_retrying_transient,
+    };
+    use std::time::Duration;
+    use tokio::net::UdpSocket;
 
     #[test]
     fn classifies_portable_transient_udp_errors() {
@@ -70,6 +127,35 @@ mod tests {
         assert!(!is_transient_udp_send_error(&fatal));
     }
 
+    #[tokio::test]
+    async fn retrying_udp_send_and_receive_round_trip() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let addr = receiver.local_addr().unwrap();
+
+        send_udp_retrying_transient(&sender, b"hello", addr, Duration::from_millis(50))
+            .await
+            .unwrap();
+        let mut buf = [0u8; 16];
+        let (n, peer) = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(50))
+            .await
+            .unwrap();
+
+        assert_eq!(&buf[..n], b"hello");
+        assert_eq!(peer, sender.local_addr().unwrap());
+    }
+
+    #[tokio::test]
+    async fn retrying_udp_receive_times_out_cleanly() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let mut buf = [0u8; 16];
+        let err = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(1))
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.kind(), std::io::ErrorKind::TimedOut);
+    }
+
     #[cfg(unix)]
     #[test]
     fn classifies_unix_network_churn_as_transient() {

tests/hostile_network.rs

@@ -29,7 +29,7 @@ use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, UdpSocket};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use std::time::{Duration, Instant};
@@ -49,9 +49,20 @@ use ed25519_dalek::{SigningKey, VerifyingKey};
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 
+const TEST_UDP_PORT_START: u16 = 31_000;
+const TEST_UDP_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_UDP_PORT_START as u32);
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
+    for _ in TEST_UDP_PORT_START..=TEST_UDP_PORT_END {
+        let offset = NEXT_UDP_PORT.fetch_add(1, Ordering::SeqCst) % span;
+        let port = TEST_UDP_PORT_START + offset as u16;
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_UDP_PORT_START}-{TEST_UDP_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {

tests/integration_smoke.rs

@@ -3,9 +3,11 @@
 use std::fs;
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream, UdpSocket};
+use std::os::unix::fs::PermissionsExt;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -23,14 +25,35 @@ use dosh::protocol::{
 };
 use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
 
+const TEST_PORT_START: u16 = 30_000;
+const TEST_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+static NEXT_TCP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+
+fn next_test_port(counter: &AtomicU32) -> u16 {
+    let span = u32::from(TEST_PORT_END - TEST_PORT_START) + 1;
+    let offset = counter.fetch_add(1, Ordering::SeqCst) % span;
+    TEST_PORT_START + offset as u16
+}
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_UDP_PORT);
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn free_tcp_port() -> u16 {
-    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
-    listener.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_TCP_PORT);
+        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free TCP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
@@ -980,6 +1003,8 @@ fn native_file_copy_recursive_round_trip() {
     fs::create_dir_all(src.join("nested")).unwrap();
     fs::write(src.join("root.txt"), b"root file\n").unwrap();
     fs::write(src.join("nested/child.txt"), b"child file\n").unwrap();
+    std::os::unix::fs::symlink("root.txt", src.join("root-link")).unwrap();
+    std::os::unix::fs::symlink("nested/child.txt", src.join("child-link")).unwrap();
 
     let client_bin = env!("CARGO_BIN_EXE_dosh-client");
     let upload = Command::new(client_bin)
@@ -1085,6 +1110,14 @@ fn native_file_copy_recursive_round_trip() {
         fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
         "child file\n"
     );
+    assert_eq!(
+        fs::read_link(downloaded.join("root-link")).unwrap(),
+        PathBuf::from("root.txt")
+    );
+    assert_eq!(
+        fs::read_link(downloaded.join("child-link")).unwrap(),
+        PathBuf::from("nested/child.txt")
+    );
 
     let remove = Command::new(client_bin)
         .arg("--dosh-host")
@@ -1219,6 +1252,80 @@ fn native_file_copy_resume_truncates_oversized_destinations() {
     let _ = server.wait();
 }
 
+#[test]
+fn native_file_copy_resume_prefix_mismatch_preserves_destinations() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+
+    let local_upload = dir.path().join("upload-prefix.txt");
+    let remote_upload = dir.path().join("remote-upload.txt");
+    fs::write(&local_upload, b"local-prefix\n").unwrap();
+    fs::write(&remote_upload, b"remote-prefix-with-tail\n").unwrap();
+    fs::set_permissions(&local_upload, fs::Permissions::from_mode(0o755)).unwrap();
+    fs::set_permissions(&remote_upload, fs::Permissions::from_mode(0o600)).unwrap();
+
+    let bad_upload_resume = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg(&local_upload)
+        .arg("local:remote-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_upload_resume.status.success(),
+        "bad upload resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_upload_resume.stdout),
+        String::from_utf8_lossy(&bad_upload_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&remote_upload).unwrap(),
+        "remote-prefix-with-tail\n"
+    );
+    assert_eq!(
+        fs::metadata(&remote_upload).unwrap().permissions().mode() & 0o777,
+        0o600
+    );
+
+    fs::write(dir.path().join("remote-download.txt"), b"remote-download\n").unwrap();
+    let local_download = dir.path().join("download-prefix.txt");
+    fs::write(&local_download, b"local-download-with-tail\n").unwrap();
+
+    let bad_download_resume = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg("local:remote-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_download_resume.status.success(),
+        "bad download resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_download_resume.stdout),
+        String::from_utf8_lossy(&bad_download_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&local_download).unwrap(),
+        "local-download-with-tail\n"
+    );
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
 #[test]
 fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();