Clarify proxied SSH UDP requirements

Cargo.lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc23"
+version = "1.0.0-rc25"
 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

@@ -960,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:#}"),
@@ -3532,6 +3540,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() {
@@ -6253,7 +6280,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));
@@ -7754,20 +7783,21 @@ mod tests {
         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,
+        load_first_native_identity_with_prompt, native_proxy_udp_warning, parse_dynamic_forward,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, parse_update_options,
+        parse_escape_key, parse_local_forward, parse_remote_forward, parse_ssh_config,
-        post_submit_hold_duration, queue_pending_user_input, raw_contains_host_table,
+        parse_update_options, post_submit_hold_duration, queue_pending_user_input,
-        recv_response_until, refresh_live_addr, release_tag_download_url,
+        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_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
+        ssh_config_uses_proxy, ssh_destination_host, ssh_username, ssh_with_user, startup_command,
-        strip_stale_mouse_reports, strip_terminal_focus_reports, terminal_private_mode_transition,
+        status_ssh_target, strip_stale_mouse_reports, strip_terminal_focus_reports,
-        toml_bare_key_or_quoted, unix_update_script, update_installer_url, update_version_status,
+        terminal_private_mode_transition, toml_bare_key_or_quoted, unix_update_script,
-        upsert_managed_block, valid_forward_host, vscode_safe_alias, windows_update_script,
+        update_installer_url, update_version_status, upsert_managed_block, valid_forward_host,
+        vscode_safe_alias, windows_update_script,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
@@ -8139,6 +8169,38 @@ 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 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()));

src/bin/dosh-server.rs

@@ -2752,10 +2752,13 @@ async fn handle_file_request(
                 atomic,
             });
             if let Some(mode) = mode {
-                let target = upload
+                let Some(active_upload) = upload.as_ref() else {
+                    return Err(anyhow!("upload state missing after start"));
+                };
+                let target = active_upload
+                    .temp_path
                     .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
+                    .unwrap_or(&active_upload.final_path)
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
                     .clone();
                 set_mode(&target, mode)?;
             }
@@ -3243,7 +3246,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);
                 }

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)

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();
+    let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
-    socket.local_addr().unwrap().port()
+    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 {