Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bfe75dfbd2 |
Generated
+1
-1
@@ -436,7 +436,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "dosh"
|
||||
version = "1.0.0-rc25"
|
||||
version = "1.0.0-rc26"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "dosh"
|
||||
version = "1.0.0-rc25"
|
||||
version = "1.0.0-rc26"
|
||||
edition = "2024"
|
||||
license = "MIT"
|
||||
|
||||
|
||||
+95
-15
@@ -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('\\')
|
||||
|
||||
+66
-44
@@ -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
@@ -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)]
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user