Compare commits

..

2 Commits

Author SHA1 Message Date
DuProcess bfe75dfbd2 Harden file resume prefix validation
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 17:59:41 -04:00
DuProcess 4ea05fc14f Clarify proxied SSH UDP requirements
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 17:44:43 -04:00
7 changed files with 336 additions and 76 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]]
name = "dosh"
version = "1.0.0-rc24"
version = "1.0.0-rc26"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dosh"
version = "1.0.0-rc24"
version = "1.0.0-rc26"
edition = "2024"
license = "MIT"
+4
View File
@@ -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.
+163 -23
View File
@@ -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:#}"),
@@ -2258,8 +2266,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:?}"),
};
@@ -2267,6 +2278,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;
@@ -2407,8 +2419,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:?}"),
};
@@ -2417,7 +2432,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:?}"),
}
@@ -2490,20 +2515,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)
@@ -2513,16 +2540,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:?}"),
}
@@ -2541,8 +2581,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 {
@@ -2577,6 +2625,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('\\')
@@ -3532,6 +3620,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() {
@@ -7756,20 +7863,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,
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,
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_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,
upsert_managed_block, valid_forward_host, vscode_safe_alias, windows_update_script,
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, upsert_managed_block, valid_forward_host,
vscode_safe_alias, windows_update_script,
};
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
use dosh::native::EnvVar;
@@ -8141,6 +8249,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()));
+66 -44
View File
@@ -2566,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(
@@ -2716,31 +2718,41 @@ 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 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)
} else {
let temp_path = upload_temp_path(&final_path, stream_id);
let file = fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&temp_path)
.with_context(|| format!("create {}", temp_path.display()))?;
(file, Some(temp_path), 0, true)
};
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()))?;
file.seek(SeekFrom::Start(written))?;
(
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()
.create(true)
.truncate(true)
.write(true)
.open(&temp_path)
.with_context(|| format!("create {}", temp_path.display()))?;
(file, Some(temp_path), 0, true, None, None)
};
*upload = Some(UploadState {
final_path,
temp_path,
@@ -2748,26 +2760,20 @@ async fn handle_file_request(
hasher,
written,
expected_size: size,
mode,
modified_secs,
atomic,
original_len,
});
if let Some(mode) = mode {
let Some(active_upload) = upload.as_ref() else {
return Err(anyhow!("upload state missing after start"));
};
let target = active_upload
.temp_path
.as_ref()
.unwrap_or(&active_upload.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
}
@@ -2789,8 +2795,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 {}",
@@ -2801,9 +2805,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()
@@ -2816,6 +2828,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,
@@ -2843,9 +2858,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,
@@ -2854,14 +2880,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 {
+26 -7
View File
@@ -46,13 +46,32 @@ 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>,
},
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)]
+75
View File
@@ -3,6 +3,7 @@
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};
@@ -1241,6 +1242,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();