Compare commits

...

10 Commits

Author SHA1 Message Date
DuProcess 2c5b9ab30d Preserve symlinks in file copy
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 18:07:13 -04:00
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
DuProcess bae0d0ed56 Harden stream queues and hostile network tests
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:33:49 -04:00
DuProcess 99e318ca6b Harden reconnect paths under UDP churn
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 12:34:18 -04:00
DuProcess b2be4c2cb7 Retry transport receive during UDP churn
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 12:15:25 -04:00
DuProcess dbf96b68d3 Keep sessions alive through UDP churn
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 12:08:32 -04:00
DuProcess d74457d198 Preserve OpenSSH alias config for hosts
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 11:47:45 -04:00
DuProcess 03f5b88889 Bump version to 1.0.0-rc19
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 11:34:42 -04:00
DuProcess 5a29925422 Stamp release builds from exported git metadata 2026-07-11 11:34:23 -04:00
17 changed files with 958 additions and 193 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]]
name = "dosh"
version = "1.0.0-rc18"
version = "1.0.0-rc27"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dosh"
version = "1.0.0-rc18"
version = "1.0.0-rc27"
edition = "2024"
license = "MIT"
+14
View File
@@ -98,6 +98,20 @@ cache_attach_tickets = true
disconnect_status = true
```
Example host entry:
```toml
[prod]
ssh = "prod"
ssh_config = "prod" # OpenSSH alias for IdentityFile, ProxyJump, SendEnv, etc.
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.
+23 -7
View File
@@ -1,3 +1,4 @@
use std::env;
use std::process::Command;
fn git(args: &[&str]) -> Option<String> {
@@ -12,14 +13,29 @@ fn git(args: &[&str]) -> Option<String> {
fn main() {
println!("cargo:rerun-if-changed=.git/HEAD");
println!("cargo:rerun-if-changed=.git/index");
println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH");
println!("cargo:rerun-if-env-changed=DOSH_BUILD_COMMIT_DATE");
println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_DIRTY");
let hash = git(&["rev-parse", "--short=12", "HEAD"]).unwrap_or_else(|| "unknown".into());
let date = git(&["show", "-s", "--format=%cs", "HEAD"]).unwrap_or_else(|| "unknown".into());
let dirty = Command::new("git")
.args(["diff", "--quiet", "--ignore-submodules", "--"])
.status()
.map(|status| if status.success() { "" } else { "+dirty" })
.unwrap_or("");
let hash = env::var("DOSH_BUILD_GIT_HASH")
.ok()
.filter(|value| !value.is_empty())
.or_else(|| git(&["rev-parse", "--short=12", "HEAD"]))
.unwrap_or_else(|| "unknown".into());
let date = env::var("DOSH_BUILD_COMMIT_DATE")
.ok()
.filter(|value| !value.is_empty())
.or_else(|| git(&["show", "-s", "--format=%cs", "HEAD"]))
.unwrap_or_else(|| "unknown".into());
let dirty = match env::var("DOSH_BUILD_GIT_DIRTY").ok().as_deref() {
Some("1" | "true" | "yes" | "+dirty") => "+dirty",
Some(_) => "",
None => Command::new("git")
.args(["diff", "--quiet", "--ignore-submodules", "--"])
.status()
.map(|status| if status.success() { "" } else { "+dirty" })
.unwrap_or(""),
};
println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}");
println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}");
+16
View File
@@ -52,6 +52,22 @@ else
release_dir="target/release"
fi
if command -v git >/dev/null 2>&1 && git rev-parse --is-inside-work-tree >/dev/null 2>&1; then
if [ -z "${DOSH_BUILD_GIT_HASH:-}" ]; then
export DOSH_BUILD_GIT_HASH="$(git rev-parse --short=12 HEAD)"
fi
if [ -z "${DOSH_BUILD_COMMIT_DATE:-}" ]; then
export DOSH_BUILD_COMMIT_DATE="$(git show -s --format=%cs HEAD)"
fi
if [ -z "${DOSH_BUILD_GIT_DIRTY+x}" ]; then
if git diff --quiet --ignore-submodules --; then
export DOSH_BUILD_GIT_DIRTY=0
else
export DOSH_BUILD_GIT_DIRTY=1
fi
fi
fi
if [ "$os" = "windows" ]; then
if [ -n "$target" ]; then
cargo build --release --target "$target" --bin dosh-client --bin dosh-bench
+313 -83
View File
@@ -36,7 +36,10 @@ use dosh::transport::{
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
TransportEvent,
};
use dosh::udp::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;
@@ -302,8 +305,7 @@ async fn main() -> Result<()> {
.get(&requested_server)
.cloned()
.unwrap_or_default();
let raw_server = host.ssh.clone().unwrap_or_else(|| requested_server.clone());
let server = ssh_with_user(&raw_server, host.user.as_deref());
let server = ssh_command_target(&requested_server, &host);
let startup_command = resolved_startup_command(&args.command, &config, &host);
let local_forwards = parse_local_forwards(&args.local_forward)?;
let remote_forwards = parse_remote_forwards(&args.remote_forward)?;
@@ -655,8 +657,7 @@ fn run_trust_command(config: &dosh::config::ClientConfig, args: &Args) -> Result
let hosts = load_hosts_config(None).unwrap_or_default();
let host_config = hosts.hosts.get(host).cloned().unwrap_or_default();
let raw_server = host_config.ssh.clone().unwrap_or_else(|| host.to_string());
let server = ssh_with_user(&raw_server, host_config.user.as_deref());
let server = ssh_command_target(host, &host_config);
let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
let ssh_auth_command = args
.ssh_auth_command
@@ -701,8 +702,8 @@ async fn run_setup_command(config: &dosh::config::ClientConfig, args: &Args) ->
for entry in imported {
match entry.action {
ImportAction::Added => println!(
"[ok] host config: [{}] ssh={} dosh_host={} port={}",
entry.alias, entry.ssh, entry.dosh_host, entry.port
"[ok] host config: [{}] ssh={} ssh_config={} dosh_host={} port={}",
entry.alias, entry.ssh, entry.ssh_config, entry.dosh_host, entry.port
),
ImportAction::Skipped => println!("[ok] host config: [{}] already exists", entry.alias),
}
@@ -710,8 +711,7 @@ async fn run_setup_command(config: &dosh::config::ClientConfig, args: &Args) ->
let hosts = load_hosts_config(None).unwrap_or_default();
let host_config = hosts.hosts.get(&host).cloned().unwrap_or_default();
let raw_server = host_config.ssh.clone().unwrap_or_else(|| host.clone());
let server = ssh_with_user(&raw_server, host_config.user.as_deref());
let server = ssh_command_target(&host, &host_config);
let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
let ssh_auth_command = args
.ssh_auth_command
@@ -843,9 +843,15 @@ fn run_restart_command(config: &dosh::config::ClientConfig, args: &Args) -> Resu
}
fn status_ssh_target(requested: &str, host_config: &dosh::config::HostConfig) -> String {
ssh_command_target(requested, host_config)
}
fn ssh_command_target(requested: &str, host_config: &dosh::config::HostConfig) -> String {
let raw_server = host_config
.ssh
.clone()
.ssh_config
.as_ref()
.or(host_config.ssh.as_ref())
.cloned()
.unwrap_or_else(|| requested.to_string());
ssh_with_user(&raw_server, host_config.user.as_deref())
}
@@ -882,11 +888,7 @@ async fn run_doctor_for_host(
) -> Result<()> {
let hosts = load_hosts_config(None).unwrap_or_default();
let host_config = hosts.hosts.get(requested).cloned().unwrap_or_default();
let raw_server = host_config
.ssh
.clone()
.unwrap_or_else(|| requested.to_string());
let server = ssh_with_user(&raw_server, host_config.user.as_deref());
let server = ssh_command_target(requested, &host_config);
let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
let dosh_port = args
.dosh_port
@@ -958,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:#}"),
@@ -1393,8 +1403,7 @@ async fn run_proxy_stdio_command(config: &dosh::config::ClientConfig, args: &Arg
.get(&requested_server)
.cloned()
.unwrap_or_default();
let raw_server = host.ssh.clone().unwrap_or_else(|| requested_server.clone());
let server = ssh_with_user(&raw_server, host.user.as_deref());
let server = ssh_command_target(&requested_server, &host);
let ssh_port = args.ssh_port.or(host.ssh_port).or(config.ssh_port);
let resolved_ssh_config = ssh_config(&server, ssh_port).unwrap_or_default();
let requested_udp_host = args
@@ -1561,11 +1570,7 @@ fn ensure_vscode_ssh_host(
) -> Result<GeneratedVscodeHost> {
let hosts = load_hosts_config(None).unwrap_or_default();
let host_config = hosts.hosts.get(requested).cloned().unwrap_or_default();
let raw_server = host_config
.ssh
.clone()
.unwrap_or_else(|| requested.to_string());
let server = ssh_with_user(&raw_server, host_config.user.as_deref());
let server = ssh_command_target(requested, &host_config);
let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
let ssh_config = ssh_config(&server, ssh_port).unwrap_or_default();
let target_ssh_port = ssh_config.port.or(ssh_port).unwrap_or(22);
@@ -1861,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<()> {
@@ -2192,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,
@@ -2261,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:?}"),
};
@@ -2270,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;
@@ -2334,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}"
)),
}
}
@@ -2368,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}"
)),
}
}
@@ -2410,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:?}"),
};
@@ -2420,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:?}"),
}
@@ -2493,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)
@@ -2516,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:?}"),
}
@@ -2544,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 {
@@ -2564,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;
@@ -2580,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('\\')
@@ -3247,6 +3416,7 @@ enum ImportAction {
struct ImportedHost {
alias: String,
ssh: String,
ssh_config: String,
dosh_host: String,
port: u16,
action: ImportAction,
@@ -3260,8 +3430,8 @@ fn run_import_ssh(config: &dosh::config::ClientConfig, aliases: &[String]) -> Re
for entry in imported {
match entry.action {
ImportAction::Added => eprintln!(
"dosh import-ssh: added [{}] ssh={} dosh_host={}",
entry.alias, entry.ssh, entry.dosh_host
"dosh import-ssh: added [{}] ssh={} ssh_config={} dosh_host={}",
entry.alias, entry.ssh, entry.ssh_config, entry.dosh_host
),
ImportAction::Skipped => {
eprintln!(
@@ -3303,6 +3473,7 @@ fn import_ssh_aliases(
imported.push(ImportedHost {
alias: alias.clone(),
ssh: alias.clone(),
ssh_config: alias.clone(),
dosh_host: udp_host,
port: config.dosh_port,
action: ImportAction::Skipped,
@@ -3312,6 +3483,7 @@ fn import_ssh_aliases(
raw.push('\n');
raw.push_str(&format!("[{}]\n", toml_bare_key_or_quoted(alias)));
raw.push_str(&format!("ssh = {}\n", toml_string(alias)));
raw.push_str(&format!("ssh_config = {}\n", toml_string(alias)));
raw.push_str(&format!("dosh_host = {}\n", toml_string(&udp_host)));
raw.push_str(&format!("port = {}\n", config.dosh_port));
if let Some(user) = &ssh_config.user {
@@ -3326,6 +3498,7 @@ fn import_ssh_aliases(
imported.push(ImportedHost {
alias: alias.clone(),
ssh: alias.clone(),
ssh_config: alias.clone(),
dosh_host: udp_host,
port: config.dosh_port,
action: ImportAction::Added,
@@ -3531,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() {
@@ -3791,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 {
@@ -3861,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 {
@@ -3949,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 {
@@ -4018,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 {
@@ -4184,7 +4362,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] {
@@ -4264,7 +4442,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] {
@@ -4340,7 +4518,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);
@@ -4395,7 +4573,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);
@@ -4442,9 +4620,9 @@ where
return Err(anyhow!("timed out waiting for server response"));
}
let remaining = wait - elapsed;
let (n, _) = 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")??;
.context("timed out waiting for server response")?;
let Ok(packet) = protocol::decode(&buf[..n]) else {
continue;
};
@@ -4834,7 +5012,11 @@ async fn run_terminal(
}
}
recv = socket.recv_from(&mut recv_buf) => {
let (n, _) = recv?;
let (n, _) = match recv {
Ok(value) => value,
Err(err) if is_transient_udp_error(&err) => continue,
Err(err) => return Err(err.into()),
};
let Ok(packet) = protocol::decode(&recv_buf[..n]) else {
continue;
};
@@ -6262,7 +6444,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));
@@ -7763,20 +7947,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_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,
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, upsert_managed_block, valid_forward_host,
vscode_safe_alias, windows_update_script,
};
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
use dosh::native::EnvVar;
@@ -8148,6 +8333,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()));
@@ -9017,6 +9234,19 @@ mod tests {
);
}
#[test]
fn ssh_command_target_prefers_explicit_ssh_config_alias() {
let host = HostConfig {
ssh: Some("server.example.com".to_string()),
ssh_config: Some("work-alias".to_string()),
user: Some("alice".to_string()),
..HostConfig::default()
};
assert_eq!(ssh_command_target("work", &host), "alice@work-alias");
assert_eq!(status_ssh_target("work", &host), "alice@work-alias");
}
#[test]
fn status_detects_local_targets() {
assert!(is_local_status_target("localhost"));
+214 -63
View File
@@ -28,7 +28,7 @@ use dosh::protocol::{
TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
};
use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
use dosh::udp::is_transient_udp_send_error;
use dosh::udp::{is_transient_udp_error, is_transient_udp_send_error};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::fs;
@@ -262,7 +262,11 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
let mut buf = vec![0u8; 65535];
loop {
let (n, peer) = socket.recv_from(&mut buf).await?;
let (n, peer) = match socket.recv_from(&mut buf).await {
Ok(value) => value,
Err(err) if is_transient_udp_error(&err) => continue,
Err(err) => return Err(err.into()),
};
if let Err(err) = handle_packet(&state, &socket, peer, &buf[..n]).await {
eprintln!("packet from {peer}: {err:#}");
}
@@ -357,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>>,
@@ -383,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,
@@ -508,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),
@@ -637,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),
@@ -758,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"
}
@@ -1185,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,
@@ -1333,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(
@@ -1469,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(
@@ -1582,20 +1617,23 @@ async fn handle_resume(
.sessions
.get_mut(&req.session)
.ok_or_else(|| anyhow!("unknown session"))?;
let client = session
.clients
.get_mut(&packet.header.conn_id)
.ok_or_else(|| anyhow!("unknown client"))?;
if !client.replay.accept(packet.header.seq) {
return Ok(());
}
client.endpoint = peer;
client.last_acked = req.last_rendered_seq;
client.cols = req.cols;
client.rows = req.rows;
client.last_seen = Instant::now();
client.send_seq += 1;
if mode_allows_terminal_updates(&client.mode) {
let (send_seq, allow_terminal_updates) = {
let client = session
.clients
.get_mut(&packet.header.conn_id)
.ok_or_else(|| anyhow!("unknown client"))?;
if !client.replay.accept(packet.header.seq) {
return Ok(());
}
client.endpoint = peer;
client.last_acked = req.last_rendered_seq;
client.cols = req.cols;
client.rows = req.rows;
client.last_seen = Instant::now();
client.send_seq += 1;
(client.send_seq, mode_allows_terminal_updates(&client.mode))
};
if allow_terminal_updates {
apply_terminal_size(
session.pty.as_ref(),
&mut session.parser,
@@ -1603,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,
@@ -2528,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(
@@ -2641,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 =
@@ -2678,31 +2758,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,
@@ -2710,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
}
@@ -2748,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 {}",
@@ -2760,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()
@@ -2775,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,
@@ -2802,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,
@@ -2813,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 {
@@ -3002,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()
@@ -3205,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);
}
@@ -3450,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());
@@ -4045,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();
@@ -4395,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(),
@@ -4459,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(),
@@ -4535,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(),
@@ -4675,6 +4824,7 @@ mod tests {
Session {
pty: None,
parser: vt100::Parser::new(24, 80, 100),
restored_screen: None,
clients: HashMap::from([(
client_id,
ClientState {
@@ -4772,6 +4922,7 @@ mod tests {
Session {
pty: None,
parser: vt100::Parser::new(24, 80, 100),
restored_screen: None,
clients: HashMap::from([(
client_id,
ClientState {
+5 -4
View File
@@ -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 {
+6
View File
@@ -150,6 +150,12 @@ pub struct ClientConfig {
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct HostConfig {
pub ssh: Option<String>,
/// Optional OpenSSH config host name to use for SSH-backed bootstrap,
/// trust/setup/doctor checks, and native-auth identity selection. This lets
/// a Dosh host keep a separate UDP endpoint while still inheriting
/// alias-scoped OpenSSH settings like IdentityFile, ProxyJump, and SendEnv.
#[serde(default)]
pub ssh_config: Option<String>,
pub user: Option<String>,
pub dosh_host: Option<String>,
pub port: Option<u16>,
+37 -7
View File
@@ -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)]
+32 -1
View File
@@ -23,12 +23,14 @@ use std::io::{Read, Write};
use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::Duration;
/// One-byte commands a server sends to a holder over its control socket after
/// the holder has handed back the master fd.
const HOLDER_CMD_SHUTDOWN: u8 = b'X';
const HOLDER_STARTUP_TIMEOUT: Duration = Duration::from_millis(750);
static SCREEN_WRITE_LOCK: Mutex<()> = Mutex::new(());
/// Magic written into a holder's `meta` file, bumped if the on-disk layout
/// changes so a stale holder from an incompatible build is ignored.
@@ -140,13 +142,28 @@ pub fn save_screen(
// No holder runtime for this session (non-persistent): nothing to do.
return Ok(());
}
let _guard = SCREEN_WRITE_LOCK
.lock()
.map_err(|_| anyhow!("screen persistence lock poisoned"))?;
let path = screen_path(&dir);
if existing_screen_output_seq(&path).is_some_and(|existing| existing >= output_seq) {
return Ok(());
}
let mut buf = Vec::with_capacity(snapshot.len() + 32);
buf.extend_from_slice(&cols.to_be_bytes());
buf.extend_from_slice(&rows.to_be_bytes());
buf.extend_from_slice(&output_seq.to_be_bytes());
buf.extend_from_slice(&(snapshot.len() as u32).to_be_bytes());
buf.extend_from_slice(snapshot);
atomic_write(&screen_path(&dir), &buf)
atomic_write(&path, &buf)
}
fn existing_screen_output_seq(path: &Path) -> Option<u64> {
let data = std::fs::read(path).ok()?;
if data.len() < 12 {
return None;
}
Some(u64::from_be_bytes(data[4..12].try_into().ok()?))
}
/// A restored screen for a re-adopted session.
@@ -619,6 +636,20 @@ mod tests {
assert_eq!(loaded.snapshot, snap);
}
#[test]
fn save_screen_never_overwrites_newer_snapshot_with_older_seq() {
let tmp = tempfile::tempdir().unwrap();
let sessions_dir = tmp.path();
ensure_runtime_dir(sessions_dir, "work").unwrap();
save_screen(sessions_dir, "work", 80, 24, 10, b"new-screen").unwrap();
save_screen(sessions_dir, "work", 80, 24, 9, b"old-screen").unwrap();
let loaded = load_screen(sessions_dir, "work").expect("screen restored");
assert_eq!(loaded.output_seq, 10);
assert_eq!(loaded.snapshot, b"new-screen");
}
#[test]
fn load_screen_absent_is_none() {
let tmp = tempfile::tempdir().unwrap();
+6 -2
View File
@@ -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,
+43 -4
View File
@@ -19,7 +19,7 @@ use crate::protocol::{
self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData,
StreamOpen, StreamOpenOk, StreamOpenReject, StreamWindowAdjust,
};
use crate::udp::is_transient_udp_send_error;
use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
use anyhow::{Result, anyhow, bail};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::net::SocketAddr;
@@ -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)
@@ -852,7 +854,11 @@ impl DoshTransport {
pub async fn recv(&mut self) -> Result<SessionEvent> {
let mut buf = vec![0u8; 65535];
loop {
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 transport_recv_error_is_retryable(&err) => continue,
Err(err) => return Err(err.into()),
};
let packet = match protocol::decode(&buf[..n]) {
Ok(packet) => packet,
Err(_) => continue,
@@ -984,6 +990,10 @@ impl DoshTransport {
}
}
fn transport_recv_error_is_retryable(err: &std::io::Error) -> bool {
is_transient_udp_error(err)
}
fn initial_stream_id(role: SessionRole) -> u64 {
match role {
SessionRole::Client => 1,
@@ -1033,6 +1043,35 @@ mod tests {
assert!(service_target("../bad").is_err());
}
#[test]
fn transport_recv_retries_network_churn_errors() {
for kind in [
std::io::ErrorKind::Interrupted,
std::io::ErrorKind::TimedOut,
std::io::ErrorKind::WouldBlock,
] {
let err = std::io::Error::from(kind);
assert!(transport_recv_error_is_retryable(&err));
}
assert!(!transport_recv_error_is_retryable(&std::io::Error::from(
std::io::ErrorKind::PermissionDenied,
)));
#[cfg(unix)]
for code in [libc::ENETUNREACH, libc::EHOSTUNREACH, libc::ECONNRESET] {
assert!(transport_recv_error_is_retryable(
&std::io::Error::from_raw_os_error(code)
));
}
#[cfg(windows)]
for code in [10051, 10054, 10065] {
assert!(transport_recv_error_is_retryable(
&std::io::Error::from_raw_os_error(code)
));
}
}
#[test]
fn service_registry_matches_registered_service_opens() {
let mut registry = ServiceRegistry::new();
+105 -13
View File
@@ -1,6 +1,10 @@
//! UDP error classification shared by Dosh clients, servers, and embedders.
pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
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(),
std::io::ErrorKind::Interrupted
@@ -9,6 +13,58 @@ pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
) || err.raw_os_error().is_some_and(is_transient_udp_os_error)
}
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!(
@@ -48,20 +104,56 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
#[cfg(test)]
mod tests {
use super::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_send_errors() {
fn classifies_portable_transient_udp_errors() {
for kind in [
std::io::ErrorKind::Interrupted,
std::io::ErrorKind::TimedOut,
std::io::ErrorKind::WouldBlock,
] {
assert!(is_transient_udp_send_error(&std::io::Error::from(kind)));
let err = std::io::Error::from(kind);
assert!(is_transient_udp_error(&err));
assert!(is_transient_udp_send_error(&err));
}
assert!(!is_transient_udp_send_error(&std::io::Error::from(
std::io::ErrorKind::PermissionDenied,
)));
let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
assert!(!is_transient_udp_error(&fatal));
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)]
@@ -78,9 +170,9 @@ mod tests {
libc::ENETUNREACH,
libc::ETIMEDOUT,
] {
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(code)
));
let err = std::io::Error::from_raw_os_error(code);
assert!(is_transient_udp_error(&err));
assert!(is_transient_udp_send_error(&err));
}
}
@@ -90,9 +182,9 @@ mod tests {
for code in [
10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065,
] {
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(code)
));
let err = std::io::Error::from_raw_os_error(code);
assert!(is_transient_udp_error(&err));
assert!(is_transient_udp_send_error(&err));
}
}
}
+14 -3
View File
@@ -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 {
+111 -4
View File
@@ -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();
+17
View File
@@ -42,6 +42,23 @@ fn package_check_verifies_only_artifacts_it_builds() {
);
}
#[test]
fn package_release_stamps_git_build_info() {
let package = include_str!("../scripts/package-release.sh");
let build = include_str!("../build.rs");
for name in [
"DOSH_BUILD_GIT_HASH",
"DOSH_BUILD_COMMIT_DATE",
"DOSH_BUILD_GIT_DIRTY",
] {
assert!(package.contains(name), "package script must export {name}");
assert!(build.contains(name), "build.rs must consume {name}");
}
assert!(package.contains("git rev-parse --short=12 HEAD"));
assert!(package.contains("git show -s --format=%cs HEAD"));
assert!(build.contains("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH"));
}
#[test]
fn systemd_service_does_not_sandbox_remote_shells() {
let service = include_str!("../packaging/systemd/dosh-server.service");