diff --git a/Cargo.lock b/Cargo.lock index c420660..2a2e825 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "1.0.0-rc25" +version = "1.0.0-rc26" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 225a906..f01a788 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "1.0.0-rc25" +version = "1.0.0-rc26" edition = "2024" license = "MIT" diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 011a7ef..539dd8b 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -2266,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:?}"), }; @@ -2275,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; @@ -2415,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:?}"), }; @@ -2425,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:?}"), } @@ -2498,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) @@ -2521,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:?}"), } @@ -2549,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 { @@ -2585,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('\\') diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 6f76014..cadd561 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -2566,8 +2566,10 @@ struct UploadState { hasher: Sha256, written: u64, expected_size: u64, + mode: Option, modified_secs: Option, atomic: bool, + original_len: Option, } 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 { diff --git a/src/file_transfer.rs b/src/file_transfer.rs index b939585..a2433b0 100644 --- a/src/file_transfer.rs +++ b/src/file_transfer.rs @@ -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 }, - Start { meta: FileMeta, offset: u64 }, - Chunk { offset: u64, bytes: Vec }, - Done { sha256: [u8; 32], bytes: u64 }, - Error { message: String }, + Resume { + offset: u64, + prefix_sha256: Option<[u8; 32]>, + }, + Stat { + meta: FileMeta, + }, + List { + entries: Vec, + }, + Start { + meta: FileMeta, + offset: u64, + prefix_sha256: Option<[u8; 32]>, + }, + Chunk { + offset: u64, + bytes: Vec, + }, + Done { + sha256: [u8; 32], + bytes: u64, + }, + Error { + message: String, + }, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] diff --git a/tests/integration_smoke.rs b/tests/integration_smoke.rs index 04a2812..5a667ae 100644 --- a/tests/integration_smoke.rs +++ b/tests/integration_smoke.rs @@ -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();