Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bfe75dfbd2 | ||
|
|
4ea05fc14f | ||
|
|
bae0d0ed56 | ||
|
|
99e318ca6b |
Generated
+1
-1
@@ -436,7 +436,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "dosh"
|
name = "dosh"
|
||||||
version = "1.0.0-rc22"
|
version = "1.0.0-rc26"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"base64",
|
"base64",
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "dosh"
|
name = "dosh"
|
||||||
version = "1.0.0-rc22"
|
version = "1.0.0-rc26"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|
||||||
|
|||||||
@@ -108,6 +108,10 @@ dosh_host = "prod.example.com"
|
|||||||
port = 50000
|
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
|
## Rust Library
|
||||||
|
|
||||||
Dosh exposes a Rust transport for encrypted, reconnecting application streams.
|
Dosh exposes a Rust transport for encrypted, reconnecting application streams.
|
||||||
|
|||||||
+186
-60
@@ -36,7 +36,10 @@ use dosh::transport::{
|
|||||||
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
|
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
|
||||||
TransportEvent,
|
TransportEvent,
|
||||||
};
|
};
|
||||||
use dosh::udp::{is_transient_udp_error, 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 serde::{Deserialize, Serialize};
|
||||||
use sha2::{Digest, Sha256};
|
use sha2::{Digest, Sha256};
|
||||||
use std::collections::BTreeMap;
|
use std::collections::BTreeMap;
|
||||||
@@ -957,6 +960,14 @@ async fn run_doctor_for_host(
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
println!("[info] native endpoint: {udp_host}:{dosh_port}");
|
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) {
|
match resolve_addr(&udp_host, dosh_port) {
|
||||||
Ok(addr) => println!("[ok] udp resolve: {addr}"),
|
Ok(addr) => println!("[ok] udp resolve: {addr}"),
|
||||||
Err(err) => println!("[fail] udp resolve: {err:#}"),
|
Err(err) => println!("[fail] udp resolve: {err:#}"),
|
||||||
@@ -2255,8 +2266,11 @@ fn upload_file(
|
|||||||
overwrite: opts.overwrite,
|
overwrite: opts.overwrite,
|
||||||
resume: opts.resume,
|
resume: opts.resume,
|
||||||
})?;
|
})?;
|
||||||
let offset = match client.recv()? {
|
let (offset, prefix_sha256) = match client.recv()? {
|
||||||
FileResponse::Resume { offset } => offset.min(size),
|
FileResponse::Resume {
|
||||||
|
offset,
|
||||||
|
prefix_sha256,
|
||||||
|
} => (offset.min(size), prefix_sha256),
|
||||||
FileResponse::Error { message } => return Err(anyhow!(message)),
|
FileResponse::Error { message } => return Err(anyhow!(message)),
|
||||||
other => bail!("unexpected upload start response: {other:?}"),
|
other => bail!("unexpected upload start response: {other:?}"),
|
||||||
};
|
};
|
||||||
@@ -2264,6 +2278,7 @@ fn upload_file(
|
|||||||
let mut hasher = Sha256::new();
|
let mut hasher = Sha256::new();
|
||||||
if offset > 0 {
|
if offset > 0 {
|
||||||
hash_prefix(&mut file, offset, &mut hasher)?;
|
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))?;
|
file.seek(std::io::SeekFrom::Start(offset))?;
|
||||||
let mut sent = offset;
|
let mut sent = offset;
|
||||||
@@ -2404,8 +2419,11 @@ fn copy_remote_file(
|
|||||||
overwrite: opts.overwrite,
|
overwrite: opts.overwrite,
|
||||||
resume: opts.resume,
|
resume: opts.resume,
|
||||||
})?;
|
})?;
|
||||||
let offset = match dst_client.recv()? {
|
let (offset, dst_prefix_sha256) = match dst_client.recv()? {
|
||||||
FileResponse::Resume { offset } => offset.min(meta.len),
|
FileResponse::Resume {
|
||||||
|
offset,
|
||||||
|
prefix_sha256,
|
||||||
|
} => (offset.min(meta.len), prefix_sha256),
|
||||||
FileResponse::Error { message } => return Err(anyhow!(message)),
|
FileResponse::Error { message } => return Err(anyhow!(message)),
|
||||||
other => bail!("unexpected remote copy start response: {other:?}"),
|
other => bail!("unexpected remote copy start response: {other:?}"),
|
||||||
};
|
};
|
||||||
@@ -2414,7 +2432,17 @@ fn copy_remote_file(
|
|||||||
offset,
|
offset,
|
||||||
})?;
|
})?;
|
||||||
match src_client.recv()? {
|
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)),
|
FileResponse::Error { message } => return Err(anyhow!(message)),
|
||||||
other => bail!("unexpected remote copy source start response: {other:?}"),
|
other => bail!("unexpected remote copy source start response: {other:?}"),
|
||||||
}
|
}
|
||||||
@@ -2487,20 +2515,22 @@ fn download_file(
|
|||||||
{
|
{
|
||||||
fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
|
fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
|
||||||
}
|
}
|
||||||
let offset = if opts.resume && local.exists() {
|
let original_len = if local.exists() {
|
||||||
fs::metadata(local)?.len().min(meta.len)
|
Some(fs::metadata(local)?.len())
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
let offset = if opts.resume {
|
||||||
|
original_len.unwrap_or(0).min(meta.len)
|
||||||
} else {
|
} else {
|
||||||
0
|
0
|
||||||
};
|
};
|
||||||
let mut file = if offset > 0 {
|
let mut file = if offset > 0 {
|
||||||
let file = fs::OpenOptions::new()
|
fs::OpenOptions::new()
|
||||||
.read(true)
|
.read(true)
|
||||||
.append(true)
|
.append(true)
|
||||||
.open(local)
|
.open(local)
|
||||||
.with_context(|| format!("open {}", local.display()))?;
|
.with_context(|| format!("open {}", local.display()))?
|
||||||
file.set_len(offset)
|
|
||||||
.with_context(|| format!("truncate {}", local.display()))?;
|
|
||||||
file
|
|
||||||
} else {
|
} else {
|
||||||
fs::OpenOptions::new()
|
fs::OpenOptions::new()
|
||||||
.create(true)
|
.create(true)
|
||||||
@@ -2510,16 +2540,29 @@ fn download_file(
|
|||||||
.with_context(|| format!("create {}", local.display()))?
|
.with_context(|| format!("create {}", local.display()))?
|
||||||
};
|
};
|
||||||
let mut hasher = Sha256::new();
|
let mut hasher = Sha256::new();
|
||||||
if offset > 0 {
|
let local_prefix_sha256 = if offset > 0 {
|
||||||
let mut prefix = fs::File::open(local)?;
|
let mut prefix = fs::File::open(local)?;
|
||||||
hash_prefix(&mut prefix, offset, &mut hasher)?;
|
hash_prefix(&mut prefix, offset, &mut hasher)?;
|
||||||
}
|
Some(digest_prefix(&hasher))
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
client.send(FileRequest::Get {
|
client.send(FileRequest::Get {
|
||||||
path: remote.to_string(),
|
path: remote.to_string(),
|
||||||
offset,
|
offset,
|
||||||
})?;
|
})?;
|
||||||
match client.recv()? {
|
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)),
|
FileResponse::Error { message } => return Err(anyhow!(message)),
|
||||||
other => bail!("unexpected download start response: {other:?}"),
|
other => bail!("unexpected download start response: {other:?}"),
|
||||||
}
|
}
|
||||||
@@ -2538,8 +2581,16 @@ fn download_file(
|
|||||||
FileResponse::Done { sha256, bytes } => {
|
FileResponse::Done { sha256, bytes } => {
|
||||||
file.flush()?;
|
file.flush()?;
|
||||||
let digest: [u8; 32] = hasher.finalize().into();
|
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!(sha256 == digest, "download checksum mismatch for {remote}");
|
||||||
anyhow::ensure!(bytes == meta.len, "download size 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_mode(local, meta.mode, opts.preserve)?;
|
||||||
set_local_mtime(local, meta.modified_secs, opts.preserve)?;
|
set_local_mtime(local, meta.modified_secs, opts.preserve)?;
|
||||||
if opts.progress {
|
if opts.progress {
|
||||||
@@ -2574,6 +2625,46 @@ fn hash_prefix(file: &mut fs::File, bytes: u64, hasher: &mut Sha256) -> Result<(
|
|||||||
Ok(())
|
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 {
|
fn path_ends_separator(path: &Path) -> bool {
|
||||||
let raw = path.as_os_str().to_string_lossy();
|
let raw = path.as_os_str().to_string_lossy();
|
||||||
raw.ends_with('/') || raw.ends_with('\\')
|
raw.ends_with('/') || raw.ends_with('\\')
|
||||||
@@ -3529,6 +3620,25 @@ fn selected_udp_host(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn ssh_config_uses_proxy(ssh_config: &SshConfig) -> bool {
|
||||||
|
ssh_config.proxy_jump.is_some() || ssh_config.proxy_command.is_some()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn native_proxy_udp_warning(
|
||||||
|
ssh_config: &SshConfig,
|
||||||
|
requested_udp_host: Option<&str>,
|
||||||
|
udp_host: &str,
|
||||||
|
dosh_port: u16,
|
||||||
|
) -> Option<String> {
|
||||||
|
if ssh_config_uses_proxy(ssh_config) && requested_udp_host.is_none() {
|
||||||
|
Some(format!(
|
||||||
|
"SSH uses ProxyJump/ProxyCommand, but Dosh UDP still needs a directly reachable host; set dosh_host if {udp_host}:{dosh_port} is not reachable from this client"
|
||||||
|
))
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn ssh_username(server: &str) -> Option<String> {
|
fn ssh_username(server: &str) -> Option<String> {
|
||||||
let (user, _) = server.rsplit_once('@')?;
|
let (user, _) = server.rsplit_once('@')?;
|
||||||
if user.is_empty() {
|
if user.is_empty() {
|
||||||
@@ -3789,14 +3899,11 @@ async fn try_native_auth(
|
|||||||
hello: hello.clone(),
|
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 mut buf = vec![0u8; 65535];
|
||||||
let (n, _) = tokio::time::timeout(
|
let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
|
||||||
Duration::from_millis(config.native_auth_timeout_ms.max(1)),
|
|
||||||
socket.recv_from(&mut buf),
|
|
||||||
)
|
|
||||||
.await??;
|
|
||||||
let packet = protocol::decode(&buf[..n])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeServerHello {
|
if packet.header.kind != PacketKind::NativeServerHello {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
@@ -3859,13 +3966,9 @@ async fn try_native_auth(
|
|||||||
CLIENT_TO_SERVER,
|
CLIENT_TO_SERVER,
|
||||||
&auth_body,
|
&auth_body,
|
||||||
)?;
|
)?;
|
||||||
socket.send_to(&packet, addr).await?;
|
send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
|
||||||
|
|
||||||
let (n, _) = tokio::time::timeout(
|
let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
|
||||||
Duration::from_millis(config.native_auth_timeout_ms.max(1)),
|
|
||||||
socket.recv_from(&mut buf),
|
|
||||||
)
|
|
||||||
.await??;
|
|
||||||
let packet = protocol::decode(&buf[..n])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeAuthOk {
|
if packet.header.kind != PacketKind::NativeAuthOk {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
@@ -3947,14 +4050,11 @@ async fn try_native_auth_check(
|
|||||||
hello: hello.clone(),
|
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 mut buf = vec![0u8; 65535];
|
||||||
let (n, _) = tokio::time::timeout(
|
let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
|
||||||
Duration::from_millis(config.native_auth_timeout_ms.max(1)),
|
|
||||||
socket.recv_from(&mut buf),
|
|
||||||
)
|
|
||||||
.await??;
|
|
||||||
let packet = protocol::decode(&buf[..n])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeServerHello {
|
if packet.header.kind != PacketKind::NativeServerHello {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
@@ -4016,13 +4116,9 @@ async fn try_native_auth_check(
|
|||||||
CLIENT_TO_SERVER,
|
CLIENT_TO_SERVER,
|
||||||
&protocol::to_body(&NativeUserAuthBody { auth })?,
|
&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(
|
let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
|
||||||
Duration::from_millis(config.native_auth_timeout_ms.max(1)),
|
|
||||||
socket.recv_from(&mut buf),
|
|
||||||
)
|
|
||||||
.await??;
|
|
||||||
let packet = protocol::decode(&buf[..n])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeAuthCheckOk {
|
if packet.header.kind != PacketKind::NativeAuthCheckOk {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
@@ -4182,7 +4278,7 @@ async fn bootstrap_attach(
|
|||||||
let body = protocol::to_body(&req)?;
|
let body = protocol::to_body(&req)?;
|
||||||
let packet =
|
let packet =
|
||||||
protocol::encode_plain(PacketKind::BootstrapAttachRequest, [0u8; 16], 1, 0, &body)?;
|
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 expected_key_id = protocol::session_key_id(&bootstrap.session_key);
|
||||||
let ok = recv_response_until(socket, Duration::from_secs(5), |packet| {
|
let ok = recv_response_until(socket, Duration::from_secs(5), |packet| {
|
||||||
if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
|
if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
|
||||||
@@ -4262,7 +4358,7 @@ async fn try_ticket_attach(
|
|||||||
0,
|
0,
|
||||||
&protocol::to_body(&envelope)?,
|
&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| {
|
let ok = recv_response_until(socket, Duration::from_millis(700), |packet| {
|
||||||
if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
|
if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
|
||||||
@@ -4338,7 +4434,7 @@ async fn try_resume_fresh_process(
|
|||||||
CLIENT_TO_SERVER,
|
CLIENT_TO_SERVER,
|
||||||
&body,
|
&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| {
|
let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
|
||||||
if !is_resume_response_for_client(&packet, cached.client_id) {
|
if !is_resume_response_for_client(&packet, cached.client_id) {
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
@@ -4393,7 +4489,7 @@ async fn try_live_resume(
|
|||||||
&body,
|
&body,
|
||||||
)?;
|
)?;
|
||||||
*send_seq += 1;
|
*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| {
|
let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
|
||||||
if !is_resume_response_for_client(&packet, cached.client_id) {
|
if !is_resume_response_for_client(&packet, cached.client_id) {
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
@@ -4440,14 +4536,9 @@ where
|
|||||||
return Err(anyhow!("timed out waiting for server response"));
|
return Err(anyhow!("timed out waiting for server response"));
|
||||||
}
|
}
|
||||||
let remaining = wait - elapsed;
|
let remaining = wait - elapsed;
|
||||||
let (n, _) = match tokio::time::timeout(remaining, socket.recv_from(&mut buf))
|
let (n, _) = recv_udp_retrying_transient(socket, &mut buf, remaining)
|
||||||
.await
|
.await
|
||||||
.context("timed out waiting for server response")?
|
.context("timed out waiting for server response")?;
|
||||||
{
|
|
||||||
Ok(value) => value,
|
|
||||||
Err(err) if is_transient_udp_error(&err) => continue,
|
|
||||||
Err(err) => return Err(err.into()),
|
|
||||||
};
|
|
||||||
let Ok(packet) = protocol::decode(&buf[..n]) else {
|
let Ok(packet) = protocol::decode(&buf[..n]) else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
@@ -6269,7 +6360,9 @@ async fn flush_stream_pending_data(
|
|||||||
if credit < bytes.len() {
|
if credit < bytes.len() {
|
||||||
break;
|
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();
|
*stream_send_credit.entry(stream_id).or_default() -= bytes.len();
|
||||||
let offset = *stream_next_send_offset.entry(stream_id).or_default();
|
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));
|
stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));
|
||||||
@@ -7770,20 +7863,21 @@ mod tests {
|
|||||||
cleanup_stream_state, clear_cached_credentials, ensure_tui_safe_status_overlay,
|
cleanup_stream_state, clear_cached_credentials, ensure_tui_safe_status_overlay,
|
||||||
input_contains_focus_in, input_matches_escape, is_local_status_target,
|
input_contains_focus_in, input_matches_escape, is_local_status_target,
|
||||||
is_resume_response_for_client, latest_release_download_url,
|
is_resume_response_for_client, latest_release_download_url,
|
||||||
load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key,
|
load_first_native_identity_with_prompt, native_proxy_udp_warning, parse_dynamic_forward,
|
||||||
parse_local_forward, parse_remote_forward, parse_ssh_config, parse_update_options,
|
parse_escape_key, parse_local_forward, parse_remote_forward, parse_ssh_config,
|
||||||
post_submit_hold_duration, queue_pending_user_input, raw_contains_host_table,
|
parse_update_options, post_submit_hold_duration, queue_pending_user_input,
|
||||||
recv_response_until, refresh_live_addr, release_tag_download_url,
|
raw_contains_host_table, recv_response_until, refresh_live_addr, release_tag_download_url,
|
||||||
release_tag_from_effective_url, release_version_from_tag, render_status_clear,
|
release_tag_from_effective_url, release_version_from_tag, render_status_clear,
|
||||||
render_status_overlay, requested_env, resolved_startup_command, retire_stream_state,
|
render_status_overlay, requested_env, resolved_startup_command, retire_stream_state,
|
||||||
retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
|
retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
|
||||||
server_version_mismatch, should_flush_terminal_input_after_contact,
|
server_version_mismatch, should_flush_terminal_input_after_contact,
|
||||||
should_hold_during_startup_gate, should_hold_post_submit_input,
|
should_hold_during_startup_gate, should_hold_post_submit_input,
|
||||||
should_repaint_idle_alternate_screen, split_after_command_submit, ssh_command_target,
|
should_repaint_idle_alternate_screen, split_after_command_submit, ssh_command_target,
|
||||||
ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
|
ssh_config_uses_proxy, ssh_destination_host, ssh_username, ssh_with_user, startup_command,
|
||||||
strip_stale_mouse_reports, strip_terminal_focus_reports, terminal_private_mode_transition,
|
status_ssh_target, strip_stale_mouse_reports, strip_terminal_focus_reports,
|
||||||
toml_bare_key_or_quoted, unix_update_script, update_installer_url, update_version_status,
|
terminal_private_mode_transition, toml_bare_key_or_quoted, unix_update_script,
|
||||||
upsert_managed_block, valid_forward_host, vscode_safe_alias, windows_update_script,
|
update_installer_url, update_version_status, upsert_managed_block, valid_forward_host,
|
||||||
|
vscode_safe_alias, windows_update_script,
|
||||||
};
|
};
|
||||||
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
|
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
|
||||||
use dosh::native::EnvVar;
|
use dosh::native::EnvVar;
|
||||||
@@ -8155,6 +8249,38 @@ mod tests {
|
|||||||
assert!(selected_udp_host(Some("any"), "alias", &ssh).is_err());
|
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]
|
#[test]
|
||||||
fn parses_user_from_ssh_destination() {
|
fn parses_user_from_ssh_destination() {
|
||||||
assert_eq!(ssh_username("alice@example.com"), Some("alice".to_string()));
|
assert_eq!(ssh_username("alice@example.com"), Some("alice".to_string()));
|
||||||
|
|||||||
+157
-61
@@ -361,6 +361,7 @@ impl NativeAuthRateLimiter {
|
|||||||
struct Session {
|
struct Session {
|
||||||
pty: Option<PtyHandle>,
|
pty: Option<PtyHandle>,
|
||||||
parser: vt100::Parser,
|
parser: vt100::Parser,
|
||||||
|
restored_screen: Option<RestoredScreenSnapshot>,
|
||||||
clients: HashMap<[u8; 16], ClientState>,
|
clients: HashMap<[u8; 16], ClientState>,
|
||||||
output_seq: u64,
|
output_seq: u64,
|
||||||
recent: VecDeque<Vec<u8>>,
|
recent: VecDeque<Vec<u8>>,
|
||||||
@@ -387,6 +388,14 @@ struct Session {
|
|||||||
last_screen_persist_at: Instant,
|
last_screen_persist_at: Instant,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
struct RestoredScreenSnapshot {
|
||||||
|
cols: u16,
|
||||||
|
rows: u16,
|
||||||
|
output_seq: u64,
|
||||||
|
snapshot: Vec<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
struct ClientState {
|
struct ClientState {
|
||||||
endpoint: SocketAddr,
|
endpoint: SocketAddr,
|
||||||
@@ -512,6 +521,7 @@ impl ServerState {
|
|||||||
Session {
|
Session {
|
||||||
pty,
|
pty,
|
||||||
parser: vt100::Parser::new(rows.max(1), cols.max(1), self.config.scrollback),
|
parser: vt100::Parser::new(rows.max(1), cols.max(1), self.config.scrollback),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::new(),
|
clients: HashMap::new(),
|
||||||
output_seq: 0,
|
output_seq: 0,
|
||||||
recent: VecDeque::with_capacity(self.config.scrollback),
|
recent: VecDeque::with_capacity(self.config.scrollback),
|
||||||
@@ -641,15 +651,23 @@ impl ServerState {
|
|||||||
.unwrap_or((80, 24));
|
.unwrap_or((80, 24));
|
||||||
let mut parser = vt100::Parser::new(rows, cols, self.config.scrollback);
|
let mut parser = vt100::Parser::new(rows, cols, self.config.scrollback);
|
||||||
let mut output_seq = 0;
|
let mut output_seq = 0;
|
||||||
|
let mut restored_screen = None;
|
||||||
if let Some(saved) = saved {
|
if let Some(saved) = saved {
|
||||||
parser.process(&saved.snapshot);
|
parser.process(&saved.snapshot);
|
||||||
output_seq = saved.output_seq;
|
output_seq = saved.output_seq;
|
||||||
|
restored_screen = Some(RestoredScreenSnapshot {
|
||||||
|
cols,
|
||||||
|
rows,
|
||||||
|
output_seq,
|
||||||
|
snapshot: saved.snapshot,
|
||||||
|
});
|
||||||
}
|
}
|
||||||
self.sessions.insert(
|
self.sessions.insert(
|
||||||
name.clone(),
|
name.clone(),
|
||||||
Session {
|
Session {
|
||||||
pty: Some(pty),
|
pty: Some(pty),
|
||||||
parser,
|
parser,
|
||||||
|
restored_screen,
|
||||||
clients: HashMap::new(),
|
clients: HashMap::new(),
|
||||||
output_seq,
|
output_seq,
|
||||||
recent: VecDeque::with_capacity(self.config.scrollback),
|
recent: VecDeque::with_capacity(self.config.scrollback),
|
||||||
@@ -762,6 +780,19 @@ fn apply_terminal_size(
|
|||||||
Ok(())
|
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 {
|
fn mode_allows_terminal_updates(mode: &str) -> bool {
|
||||||
mode != "view-only" && mode != "forward-only"
|
mode != "view-only" && mode != "forward-only"
|
||||||
}
|
}
|
||||||
@@ -1189,7 +1220,7 @@ async fn handle_native_user_auth(
|
|||||||
apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?;
|
apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?;
|
||||||
}
|
}
|
||||||
let client_id = crypto::random_16();
|
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;
|
let output_seq = session.output_seq;
|
||||||
locked.insert_client(
|
locked.insert_client(
|
||||||
&session_name,
|
&session_name,
|
||||||
@@ -1337,7 +1368,7 @@ async fn handle_bootstrap_attach(
|
|||||||
)?;
|
)?;
|
||||||
}
|
}
|
||||||
let client_id = crypto::random_16();
|
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 output_seq = session.output_seq;
|
||||||
let bootstrap_session = req.bootstrap.session.clone();
|
let bootstrap_session = req.bootstrap.session.clone();
|
||||||
locked.insert_client(
|
locked.insert_client(
|
||||||
@@ -1473,7 +1504,7 @@ async fn handle_ticket_attach(
|
|||||||
)?;
|
)?;
|
||||||
}
|
}
|
||||||
let client_id = crypto::random_16();
|
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 output_seq = session.output_seq;
|
||||||
let ticket_session = req.session.clone();
|
let ticket_session = req.session.clone();
|
||||||
locked.insert_client(
|
locked.insert_client(
|
||||||
@@ -1586,20 +1617,23 @@ async fn handle_resume(
|
|||||||
.sessions
|
.sessions
|
||||||
.get_mut(&req.session)
|
.get_mut(&req.session)
|
||||||
.ok_or_else(|| anyhow!("unknown session"))?;
|
.ok_or_else(|| anyhow!("unknown session"))?;
|
||||||
let client = session
|
let (send_seq, allow_terminal_updates) = {
|
||||||
.clients
|
let client = session
|
||||||
.get_mut(&packet.header.conn_id)
|
.clients
|
||||||
.ok_or_else(|| anyhow!("unknown client"))?;
|
.get_mut(&packet.header.conn_id)
|
||||||
if !client.replay.accept(packet.header.seq) {
|
.ok_or_else(|| anyhow!("unknown client"))?;
|
||||||
return Ok(());
|
if !client.replay.accept(packet.header.seq) {
|
||||||
}
|
return Ok(());
|
||||||
client.endpoint = peer;
|
}
|
||||||
client.last_acked = req.last_rendered_seq;
|
client.endpoint = peer;
|
||||||
client.cols = req.cols;
|
client.last_acked = req.last_rendered_seq;
|
||||||
client.rows = req.rows;
|
client.cols = req.cols;
|
||||||
client.last_seen = Instant::now();
|
client.rows = req.rows;
|
||||||
client.send_seq += 1;
|
client.last_seen = Instant::now();
|
||||||
if mode_allows_terminal_updates(&client.mode) {
|
client.send_seq += 1;
|
||||||
|
(client.send_seq, mode_allows_terminal_updates(&client.mode))
|
||||||
|
};
|
||||||
|
if allow_terminal_updates {
|
||||||
apply_terminal_size(
|
apply_terminal_size(
|
||||||
session.pty.as_ref(),
|
session.pty.as_ref(),
|
||||||
&mut session.parser,
|
&mut session.parser,
|
||||||
@@ -1607,8 +1641,8 @@ async fn handle_resume(
|
|||||||
req.rows,
|
req.rows,
|
||||||
)?;
|
)?;
|
||||||
}
|
}
|
||||||
let snapshot = screen_snapshot(session.parser.screen());
|
let snapshot = attach_snapshot(session, req.cols, req.rows);
|
||||||
(client.send_seq, session.output_seq, snapshot)
|
(send_seq, session.output_seq, snapshot)
|
||||||
};
|
};
|
||||||
let frame = Frame {
|
let frame = Frame {
|
||||||
session: session_name,
|
session: session_name,
|
||||||
@@ -2532,8 +2566,10 @@ struct UploadState {
|
|||||||
hasher: Sha256,
|
hasher: Sha256,
|
||||||
written: u64,
|
written: u64,
|
||||||
expected_size: u64,
|
expected_size: u64,
|
||||||
|
mode: Option<u32>,
|
||||||
modified_secs: Option<u64>,
|
modified_secs: Option<u64>,
|
||||||
atomic: bool,
|
atomic: bool,
|
||||||
|
original_len: Option<u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn run_file_stream_service(
|
async fn run_file_stream_service(
|
||||||
@@ -2682,31 +2718,41 @@ async fn handle_file_request(
|
|||||||
bail!("destination exists: {}", final_path.display());
|
bail!("destination exists: {}", final_path.display());
|
||||||
}
|
}
|
||||||
let mut hasher = Sha256::new();
|
let mut hasher = Sha256::new();
|
||||||
let (file, temp_path, written, atomic) = if resume && final_path.exists() {
|
let (file, temp_path, written, atomic, original_len, prefix_sha256) =
|
||||||
let mut file = fs::OpenOptions::new()
|
if resume && final_path.exists() {
|
||||||
.read(true)
|
let original_len = fs::metadata(&final_path)
|
||||||
.append(true)
|
.with_context(|| format!("stat {}", final_path.display()))?
|
||||||
.open(&final_path)
|
.len();
|
||||||
.with_context(|| format!("open {}", final_path.display()))?;
|
let written = original_len.min(size);
|
||||||
let written = file.metadata()?.len().min(size);
|
if written > 0 {
|
||||||
file.set_len(written)
|
let mut prefix = fs::File::open(&final_path)?;
|
||||||
.with_context(|| format!("truncate {}", final_path.display()))?;
|
hash_exact_prefix(&mut prefix, written, &mut hasher)?;
|
||||||
if written > 0 {
|
}
|
||||||
let mut prefix = fs::File::open(&final_path)?;
|
let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
|
||||||
hash_exact_prefix(&mut prefix, written, &mut hasher)?;
|
let mut file = fs::OpenOptions::new()
|
||||||
}
|
.read(true)
|
||||||
file.seek(SeekFrom::Start(written))?;
|
.append(true)
|
||||||
(file, None, written, false)
|
.open(&final_path)
|
||||||
} else {
|
.with_context(|| format!("open {}", final_path.display()))?;
|
||||||
let temp_path = upload_temp_path(&final_path, stream_id);
|
file.seek(SeekFrom::Start(written))?;
|
||||||
let file = fs::OpenOptions::new()
|
(
|
||||||
.create(true)
|
file,
|
||||||
.truncate(true)
|
None,
|
||||||
.write(true)
|
written,
|
||||||
.open(&temp_path)
|
false,
|
||||||
.with_context(|| format!("create {}", temp_path.display()))?;
|
Some(original_len),
|
||||||
(file, Some(temp_path), 0, true)
|
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 {
|
*upload = Some(UploadState {
|
||||||
final_path,
|
final_path,
|
||||||
temp_path,
|
temp_path,
|
||||||
@@ -2714,23 +2760,20 @@ async fn handle_file_request(
|
|||||||
hasher,
|
hasher,
|
||||||
written,
|
written,
|
||||||
expected_size: size,
|
expected_size: size,
|
||||||
|
mode,
|
||||||
modified_secs,
|
modified_secs,
|
||||||
atomic,
|
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(
|
send_file_response_to_client(
|
||||||
state,
|
state,
|
||||||
socket,
|
socket,
|
||||||
client_id,
|
client_id,
|
||||||
stream_id,
|
stream_id,
|
||||||
FileResponse::Resume { offset: written },
|
FileResponse::Resume {
|
||||||
|
offset: written,
|
||||||
|
prefix_sha256,
|
||||||
|
},
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
@@ -2752,8 +2795,6 @@ async fn handle_file_request(
|
|||||||
let Some(mut active) = upload.take() else {
|
let Some(mut active) = upload.take() else {
|
||||||
bail!("no upload in progress");
|
bail!("no upload in progress");
|
||||||
};
|
};
|
||||||
active.file.flush()?;
|
|
||||||
active.file.sync_all()?;
|
|
||||||
anyhow::ensure!(
|
anyhow::ensure!(
|
||||||
active.written == active.expected_size,
|
active.written == active.expected_size,
|
||||||
"upload size mismatch: got {}, expected {}",
|
"upload size mismatch: got {}, expected {}",
|
||||||
@@ -2764,9 +2805,17 @@ async fn handle_file_request(
|
|||||||
if digest != sha256 {
|
if digest != sha256 {
|
||||||
if let Some(temp_path) = &active.temp_path {
|
if let Some(temp_path) = &active.temp_path {
|
||||||
let _ = fs::remove_file(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");
|
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);
|
drop(active.file);
|
||||||
if active.atomic
|
if active.atomic
|
||||||
&& let Some(temp_path) = active.temp_path.take()
|
&& let Some(temp_path) = active.temp_path.take()
|
||||||
@@ -2779,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 {
|
if let Some(modified_secs) = active.modified_secs {
|
||||||
filetime::set_file_mtime(
|
filetime::set_file_mtime(
|
||||||
&active.final_path,
|
&active.final_path,
|
||||||
@@ -2806,9 +2858,20 @@ async fn handle_file_request(
|
|||||||
"remote path is not a file: {}",
|
"remote path is not a file: {}",
|
||||||
meta.path
|
meta.path
|
||||||
);
|
);
|
||||||
|
anyhow::ensure!(
|
||||||
|
offset <= meta.len,
|
||||||
|
"download offset {offset} is beyond file size {}",
|
||||||
|
meta.len
|
||||||
|
);
|
||||||
let mut file =
|
let mut file =
|
||||||
fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
|
fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
|
||||||
file.seek(SeekFrom::Start(offset))?;
|
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(
|
send_file_response_to_client(
|
||||||
state,
|
state,
|
||||||
socket,
|
socket,
|
||||||
@@ -2817,14 +2880,10 @@ async fn handle_file_request(
|
|||||||
FileResponse::Start {
|
FileResponse::Start {
|
||||||
meta: meta.clone(),
|
meta: meta.clone(),
|
||||||
offset,
|
offset,
|
||||||
|
prefix_sha256,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.await?;
|
.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 sent = offset;
|
||||||
let mut buf = vec![0u8; CHUNK_SIZE];
|
let mut buf = vec![0u8; CHUNK_SIZE];
|
||||||
loop {
|
loop {
|
||||||
@@ -3209,7 +3268,10 @@ async fn flush_stream_pending_data_to_client(
|
|||||||
if credit < bytes.len() {
|
if credit < bytes.len() {
|
||||||
return Ok(());
|
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() {
|
if pending.is_empty() {
|
||||||
client.stream_pending_data.remove(&stream_id);
|
client.stream_pending_data.remove(&stream_id);
|
||||||
}
|
}
|
||||||
@@ -3454,6 +3516,7 @@ async fn broadcast_output(
|
|||||||
.get_mut(&output.session)
|
.get_mut(&output.session)
|
||||||
.ok_or_else(|| anyhow!("unknown session"))?;
|
.ok_or_else(|| anyhow!("unknown session"))?;
|
||||||
session.parser.process(&output.bytes);
|
session.parser.process(&output.bytes);
|
||||||
|
session.restored_screen = None;
|
||||||
session.output_seq += 1;
|
session.output_seq += 1;
|
||||||
let output_seq = session.output_seq;
|
let output_seq = session.output_seq;
|
||||||
session.recent.push_back(output.bytes.clone());
|
session.recent.push_back(output.bytes.clone());
|
||||||
@@ -4049,6 +4112,34 @@ mod tests {
|
|||||||
assert_eq!(ServerConfig::default().client_timeout_secs, 2_592_000);
|
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]
|
#[tokio::test]
|
||||||
async fn unknown_resume_reject_keeps_client_id() {
|
async fn unknown_resume_reject_keeps_client_id() {
|
||||||
let (pty_tx, _rx) = mpsc::unbounded_channel();
|
let (pty_tx, _rx) = mpsc::unbounded_channel();
|
||||||
@@ -4399,6 +4490,7 @@ mod tests {
|
|||||||
Session {
|
Session {
|
||||||
pty: None,
|
pty: None,
|
||||||
parser: vt100::Parser::new(24, 80, 100),
|
parser: vt100::Parser::new(24, 80, 100),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::from([(client_id, client)]),
|
clients: HashMap::from([(client_id, client)]),
|
||||||
output_seq: 0,
|
output_seq: 0,
|
||||||
recent: VecDeque::new(),
|
recent: VecDeque::new(),
|
||||||
@@ -4463,6 +4555,7 @@ mod tests {
|
|||||||
Session {
|
Session {
|
||||||
pty: None,
|
pty: None,
|
||||||
parser: vt100::Parser::new(24, 80, 100),
|
parser: vt100::Parser::new(24, 80, 100),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::from([(client_id, client)]),
|
clients: HashMap::from([(client_id, client)]),
|
||||||
output_seq: 0,
|
output_seq: 0,
|
||||||
recent: VecDeque::new(),
|
recent: VecDeque::new(),
|
||||||
@@ -4539,6 +4632,7 @@ mod tests {
|
|||||||
Session {
|
Session {
|
||||||
pty: None,
|
pty: None,
|
||||||
parser: vt100::Parser::new(24, 80, 100),
|
parser: vt100::Parser::new(24, 80, 100),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::from([(client_id, client)]),
|
clients: HashMap::from([(client_id, client)]),
|
||||||
output_seq: 0,
|
output_seq: 0,
|
||||||
recent: VecDeque::new(),
|
recent: VecDeque::new(),
|
||||||
@@ -4679,6 +4773,7 @@ mod tests {
|
|||||||
Session {
|
Session {
|
||||||
pty: None,
|
pty: None,
|
||||||
parser: vt100::Parser::new(24, 80, 100),
|
parser: vt100::Parser::new(24, 80, 100),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::from([(
|
clients: HashMap::from([(
|
||||||
client_id,
|
client_id,
|
||||||
ClientState {
|
ClientState {
|
||||||
@@ -4776,6 +4871,7 @@ mod tests {
|
|||||||
Session {
|
Session {
|
||||||
pty: None,
|
pty: None,
|
||||||
parser: vt100::Parser::new(24, 80, 100),
|
parser: vt100::Parser::new(24, 80, 100),
|
||||||
|
restored_screen: None,
|
||||||
clients: HashMap::from([(
|
clients: HashMap::from([(
|
||||||
client_id,
|
client_id,
|
||||||
ClientState {
|
ClientState {
|
||||||
|
|||||||
+5
-4
@@ -13,6 +13,7 @@ use crate::protocol::{
|
|||||||
};
|
};
|
||||||
use crate::ssh_agent;
|
use crate::ssh_agent;
|
||||||
use crate::transport::{DoshTransport, SessionRole, SessionTransportConfig, TransportConfig};
|
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 anyhow::{Context, Result, anyhow, bail};
|
||||||
use std::net::{SocketAddr, ToSocketAddrs};
|
use std::net::{SocketAddr, ToSocketAddrs};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
@@ -214,10 +215,10 @@ impl DoshClientBuilder {
|
|||||||
hello: hello.clone(),
|
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 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])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeServerHello {
|
if packet.header.kind != PacketKind::NativeServerHello {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
@@ -261,8 +262,8 @@ impl DoshClientBuilder {
|
|||||||
CLIENT_TO_SERVER,
|
CLIENT_TO_SERVER,
|
||||||
&protocol::to_body(&NativeUserAuthBody { auth })?,
|
&protocol::to_body(&NativeUserAuthBody { auth })?,
|
||||||
)?;
|
)?;
|
||||||
socket.send_to(&auth_packet, peer_addr).await?;
|
send_udp_retrying_transient(&socket, &auth_packet, peer_addr, timeout).await?;
|
||||||
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])?;
|
let packet = protocol::decode(&buf[..n])?;
|
||||||
if packet.header.kind != PacketKind::NativeAuthOk {
|
if packet.header.kind != PacketKind::NativeAuthOk {
|
||||||
if packet.header.kind == PacketKind::AttachReject {
|
if packet.header.kind == PacketKind::AttachReject {
|
||||||
|
|||||||
+26
-7
@@ -46,13 +46,32 @@ pub enum FileRequest {
|
|||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
pub enum FileResponse {
|
pub enum FileResponse {
|
||||||
Ok,
|
Ok,
|
||||||
Resume { offset: u64 },
|
Resume {
|
||||||
Stat { meta: FileMeta },
|
offset: u64,
|
||||||
List { entries: Vec<FileEntry> },
|
prefix_sha256: Option<[u8; 32]>,
|
||||||
Start { meta: FileMeta, offset: u64 },
|
},
|
||||||
Chunk { offset: u64, bytes: Vec<u8> },
|
Stat {
|
||||||
Done { sha256: [u8; 32], bytes: u64 },
|
meta: FileMeta,
|
||||||
Error { message: String },
|
},
|
||||||
|
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)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
|
|||||||
+6
-2
@@ -13,7 +13,7 @@ use crate::transport::{
|
|||||||
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
|
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
|
||||||
service_name_from_target,
|
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 anyhow::{Context, Result, anyhow, bail};
|
||||||
use ed25519_dalek::SigningKey;
|
use ed25519_dalek::SigningKey;
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
@@ -169,7 +169,11 @@ impl DoshServer {
|
|||||||
let mut buf = vec![0u8; 65535];
|
let mut buf = vec![0u8; 65535];
|
||||||
loop {
|
loop {
|
||||||
self.expire_pending();
|
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]) {
|
let packet = match protocol::decode(&buf[..n]) {
|
||||||
Ok(packet) => packet,
|
Ok(packet) => packet,
|
||||||
Err(_) => continue,
|
Err(_) => continue,
|
||||||
|
|||||||
+4
-2
@@ -596,11 +596,13 @@ impl StreamMux {
|
|||||||
if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
|
if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
let bytes = self
|
let Some(bytes) = self
|
||||||
.pending_data
|
.pending_data
|
||||||
.get_mut(&stream_id)
|
.get_mut(&stream_id)
|
||||||
.and_then(VecDeque::pop_front)
|
.and_then(VecDeque::pop_front)
|
||||||
.expect("pending front exists");
|
else {
|
||||||
|
break;
|
||||||
|
};
|
||||||
if self
|
if self
|
||||||
.pending_data
|
.pending_data
|
||||||
.get(&stream_id)
|
.get(&stream_id)
|
||||||
|
|||||||
+87
-1
@@ -1,5 +1,9 @@
|
|||||||
//! UDP error classification shared by Dosh clients, servers, and embedders.
|
//! UDP error classification shared by Dosh clients, servers, and embedders.
|
||||||
|
|
||||||
|
use std::net::SocketAddr;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
use tokio::net::UdpSocket;
|
||||||
|
|
||||||
pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
|
pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
|
||||||
matches!(
|
matches!(
|
||||||
err.kind(),
|
err.kind(),
|
||||||
@@ -13,6 +17,54 @@ pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
|
|||||||
is_transient_udp_error(err)
|
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)]
|
#[cfg(unix)]
|
||||||
fn is_transient_udp_os_error(code: i32) -> bool {
|
fn is_transient_udp_os_error(code: i32) -> bool {
|
||||||
matches!(
|
matches!(
|
||||||
@@ -52,7 +104,12 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{is_transient_udp_error, 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]
|
#[test]
|
||||||
fn classifies_portable_transient_udp_errors() {
|
fn classifies_portable_transient_udp_errors() {
|
||||||
@@ -70,6 +127,35 @@ mod tests {
|
|||||||
assert!(!is_transient_udp_send_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)]
|
#[cfg(unix)]
|
||||||
#[test]
|
#[test]
|
||||||
fn classifies_unix_network_churn_as_transient() {
|
fn classifies_unix_network_churn_as_transient() {
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ use std::io::{Read, Write};
|
|||||||
use std::net::{SocketAddr, TcpListener, UdpSocket};
|
use std::net::{SocketAddr, TcpListener, UdpSocket};
|
||||||
use std::process::{Child, Command, Stdio};
|
use std::process::{Child, Command, Stdio};
|
||||||
use std::sync::Arc;
|
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::sync::mpsc::{Receiver, Sender, channel};
|
||||||
use std::thread;
|
use std::thread;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
@@ -49,9 +49,20 @@ use ed25519_dalek::{SigningKey, VerifyingKey};
|
|||||||
use rand::rngs::StdRng;
|
use rand::rngs::StdRng;
|
||||||
use rand::{Rng, SeedableRng};
|
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 {
|
fn free_udp_port() -> u16 {
|
||||||
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
|
let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
|
||||||
socket.local_addr().unwrap().port()
|
for _ in TEST_UDP_PORT_START..=TEST_UDP_PORT_END {
|
||||||
|
let offset = NEXT_UDP_PORT.fetch_add(1, Ordering::SeqCst) % span;
|
||||||
|
let port = TEST_UDP_PORT_START + offset as u16;
|
||||||
|
if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
|
||||||
|
return port;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
panic!("no free UDP test port in {TEST_UDP_PORT_START}-{TEST_UDP_PORT_END}");
|
||||||
}
|
}
|
||||||
|
|
||||||
fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
|
fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
|
||||||
|
|||||||
+101
-4
@@ -3,9 +3,11 @@
|
|||||||
use std::fs;
|
use std::fs;
|
||||||
use std::io::{Read, Write};
|
use std::io::{Read, Write};
|
||||||
use std::net::{TcpListener, TcpStream, UdpSocket};
|
use std::net::{TcpListener, TcpStream, UdpSocket};
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
use std::os::unix::net::{UnixListener, UnixStream};
|
use std::os::unix::net::{UnixListener, UnixStream};
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::process::{Child, Command, Stdio};
|
use std::process::{Child, Command, Stdio};
|
||||||
|
use std::sync::atomic::{AtomicU32, Ordering};
|
||||||
use std::sync::mpsc;
|
use std::sync::mpsc;
|
||||||
use std::thread;
|
use std::thread;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
@@ -23,14 +25,35 @@ use dosh::protocol::{
|
|||||||
};
|
};
|
||||||
use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
|
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 {
|
fn free_udp_port() -> u16 {
|
||||||
let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
|
for _ in TEST_PORT_START..=TEST_PORT_END {
|
||||||
socket.local_addr().unwrap().port()
|
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 {
|
fn free_tcp_port() -> u16 {
|
||||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
for _ in TEST_PORT_START..=TEST_PORT_END {
|
||||||
listener.local_addr().unwrap().port()
|
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 {
|
fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
|
||||||
@@ -1219,6 +1242,80 @@ fn native_file_copy_resume_truncates_oversized_destinations() {
|
|||||||
let _ = server.wait();
|
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]
|
#[test]
|
||||||
fn native_exec_command_smoke() {
|
fn native_exec_command_smoke() {
|
||||||
let dir = tempfile::tempdir().unwrap();
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
|||||||
Reference in New Issue
Block a user