Compare commits
19
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1dcb33550e | ||
|
|
2c5b9ab30d | ||
|
|
bfe75dfbd2 | ||
|
|
4ea05fc14f | ||
|
|
bae0d0ed56 | ||
|
|
99e318ca6b | ||
|
|
b2be4c2cb7 | ||
|
|
dbf96b68d3 | ||
|
|
d74457d198 | ||
|
|
03f5b88889 | ||
|
|
5a29925422 | ||
|
|
630b8dab30 | ||
|
|
b87f277e2d | ||
|
|
732f53f2d0 | ||
|
|
6aa03cca63 | ||
|
|
58d8e34828 | ||
|
|
47e446b433 | ||
|
|
e11ba1a652 | ||
|
|
07c4b321d1 |
Generated
+1
-1
@@ -436,7 +436,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "dosh"
|
||||
version = "1.0.0-rc15"
|
||||
version = "1.0.0-rc28"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "dosh"
|
||||
version = "1.0.0-rc15"
|
||||
version = "1.0.0-rc28"
|
||||
edition = "2024"
|
||||
license = "MIT"
|
||||
|
||||
|
||||
@@ -36,7 +36,9 @@ persistence-check:
|
||||
package-check:
|
||||
$(MAKE) package-release-linux
|
||||
$(MAKE) package-release-windows
|
||||
sh scripts/verify-release-artifacts.sh
|
||||
sh scripts/verify-release-artifacts.sh \
|
||||
target/dosh-release/dosh-linux-x86_64.tar.gz \
|
||||
target/dosh-release/dosh-windows-x86_64.zip
|
||||
|
||||
install:
|
||||
sh install.sh --from-current
|
||||
|
||||
@@ -44,6 +44,7 @@ dosh setup HOST
|
||||
dosh HOST
|
||||
dosh HOST COMMAND
|
||||
dosh update
|
||||
dosh update --client|--server|--both
|
||||
```
|
||||
|
||||
Useful commands:
|
||||
@@ -97,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.
|
||||
|
||||
@@ -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")
|
||||
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("");
|
||||
.unwrap_or(""),
|
||||
};
|
||||
|
||||
println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}");
|
||||
println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}");
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -22,11 +22,14 @@ artifact_version() {
|
||||
esac
|
||||
}
|
||||
|
||||
if [ "$#" -gt 0 ]; then
|
||||
artifacts="$*"
|
||||
else
|
||||
artifacts="$out_dir/dosh-linux-x86_64.tar.gz $out_dir/dosh-macos-aarch64.tar.gz $out_dir/dosh-windows-x86_64.zip"
|
||||
fi
|
||||
|
||||
failed=0
|
||||
for artifact in \
|
||||
"$out_dir/dosh-linux-x86_64.tar.gz" \
|
||||
"$out_dir/dosh-macos-aarch64.tar.gz" \
|
||||
"$out_dir/dosh-windows-x86_64.zip"; do
|
||||
for artifact in $artifacts; do
|
||||
if [ ! -f "$artifact" ]; then
|
||||
echo "missing release artifact: $artifact" >&2
|
||||
failed=1
|
||||
|
||||
+831
-148
File diff suppressed because it is too large
Load Diff
+224
-47
@@ -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;
|
||||
@@ -53,6 +53,7 @@ const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
|
||||
/// have accumulated since the last mirror. Keeps the atomic write off the
|
||||
/// per-packet hot path while bounding how much fresh output a crash can lose.
|
||||
const SCREEN_PERSIST_BYTES: usize = 4096;
|
||||
const SCREEN_PERSIST_MAX_AGE: Duration = Duration::from_millis(250);
|
||||
const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300;
|
||||
|
||||
/// Sentinel `target_host` sent to the client on a server-initiated `StreamOpen`
|
||||
@@ -251,7 +252,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
|
||||
if config.persist_sessions {
|
||||
let flush_state = Arc::clone(&state);
|
||||
tokio::spawn(async move {
|
||||
let mut interval = tokio::time::interval(Duration::from_secs(2));
|
||||
let mut interval = tokio::time::interval(SCREEN_PERSIST_MAX_AGE);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
flush_persistent_screens(&flush_state);
|
||||
@@ -261,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:#}");
|
||||
}
|
||||
@@ -356,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>>,
|
||||
@@ -376,6 +382,18 @@ struct Session {
|
||||
/// `output_seq` at the last successful screen mirror, so the periodic flush
|
||||
/// skips sessions whose screen has not changed.
|
||||
last_persisted_seq: u64,
|
||||
/// Last time this session's screen was considered for a mirror. Combined
|
||||
/// with the byte threshold so low-throughput prompt/TUI updates persist
|
||||
/// quickly without writing on every packet in high-throughput bursts.
|
||||
last_screen_persist_at: Instant,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct RestoredScreenSnapshot {
|
||||
cols: u16,
|
||||
rows: u16,
|
||||
output_seq: u64,
|
||||
snapshot: Vec<u8>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -503,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),
|
||||
@@ -511,6 +530,7 @@ impl ServerState {
|
||||
persistent,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
@@ -631,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),
|
||||
@@ -648,6 +676,7 @@ impl ServerState {
|
||||
persistent: true,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: output_seq,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
eprintln!(
|
||||
@@ -751,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"
|
||||
}
|
||||
@@ -1178,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,
|
||||
@@ -1326,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(
|
||||
@@ -1462,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(
|
||||
@@ -1575,6 +1617,7 @@ async fn handle_resume(
|
||||
.sessions
|
||||
.get_mut(&req.session)
|
||||
.ok_or_else(|| anyhow!("unknown session"))?;
|
||||
let (send_seq, allow_terminal_updates) = {
|
||||
let client = session
|
||||
.clients
|
||||
.get_mut(&packet.header.conn_id)
|
||||
@@ -1588,7 +1631,9 @@ async fn handle_resume(
|
||||
client.rows = req.rows;
|
||||
client.last_seen = Instant::now();
|
||||
client.send_seq += 1;
|
||||
if mode_allows_terminal_updates(&client.mode) {
|
||||
(client.send_seq, mode_allows_terminal_updates(&client.mode))
|
||||
};
|
||||
if allow_terminal_updates {
|
||||
apply_terminal_size(
|
||||
session.pty.as_ref(),
|
||||
&mut session.parser,
|
||||
@@ -1596,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,
|
||||
@@ -1736,11 +1781,11 @@ async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packe
|
||||
let (key, _) = find_client_decrypt_key(state, &packet.header)?;
|
||||
let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
|
||||
let mut locked = state.lock().expect("server state poisoned");
|
||||
if let Some(client) = locked.client_mut(&packet.header.conn_id) {
|
||||
if !client.replay.accept(packet.header.seq) {
|
||||
if let Some(client) = locked.client_mut(&packet.header.conn_id)
|
||||
&& !client.replay.accept(packet.header.seq)
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
locked.remove_client_everywhere(&packet.header.conn_id);
|
||||
Ok(())
|
||||
}
|
||||
@@ -2521,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(
|
||||
@@ -2634,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 =
|
||||
@@ -2671,22 +2758,31 @@ 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)
|
||||
.write(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()))?;
|
||||
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)
|
||||
(
|
||||
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()
|
||||
@@ -2695,7 +2791,7 @@ async fn handle_file_request(
|
||||
.write(true)
|
||||
.open(&temp_path)
|
||||
.with_context(|| format!("create {}", temp_path.display()))?;
|
||||
(file, Some(temp_path), 0, true)
|
||||
(file, Some(temp_path), 0, true, None, None)
|
||||
};
|
||||
*upload = Some(UploadState {
|
||||
final_path,
|
||||
@@ -2704,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
|
||||
}
|
||||
@@ -2742,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 {}",
|
||||
@@ -2754,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()
|
||||
@@ -2769,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,
|
||||
@@ -2796,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,
|
||||
@@ -2807,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 {
|
||||
@@ -2996,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()
|
||||
@@ -3199,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);
|
||||
}
|
||||
@@ -3444,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());
|
||||
@@ -3451,16 +3575,20 @@ async fn broadcast_output(
|
||||
session.recent.pop_front();
|
||||
}
|
||||
// For a persistent session, mirror the screen to disk periodically so a
|
||||
// post-restart reattach repaints. Throttled by accumulated bytes to keep
|
||||
// the (atomic) write off the per-packet hot path. The actual file write
|
||||
// happens after the lock is dropped.
|
||||
// post-restart reattach repaints. Throttled by accumulated bytes or a
|
||||
// short wall-clock age so low-throughput prompt/TUI changes persist
|
||||
// quickly while large bursts still avoid per-packet disk writes. The
|
||||
// actual file write happens after the lock is dropped.
|
||||
if session.persistent {
|
||||
session.bytes_since_persist = session
|
||||
.bytes_since_persist
|
||||
.saturating_add(output.bytes.len());
|
||||
if session.bytes_since_persist >= SCREEN_PERSIST_BYTES {
|
||||
let now = Instant::now();
|
||||
if session.bytes_since_persist >= SCREEN_PERSIST_BYTES
|
||||
|| now.duration_since(session.last_screen_persist_at) >= SCREEN_PERSIST_MAX_AGE
|
||||
{
|
||||
session.bytes_since_persist = 0;
|
||||
session.last_persisted_seq = output_seq;
|
||||
session.last_screen_persist_at = now;
|
||||
let screen = session.parser.screen();
|
||||
persist_screen = Some((
|
||||
output.session.clone(),
|
||||
@@ -3519,6 +3647,8 @@ async fn broadcast_output(
|
||||
persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
|
||||
{
|
||||
eprintln!("screen persist for {name} failed: {err:#}");
|
||||
} else {
|
||||
mark_screen_persisted(state, &name, output_seq);
|
||||
}
|
||||
}
|
||||
for (endpoint, packet) in sends {
|
||||
@@ -3800,8 +3930,8 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
|
||||
if !session.persistent || session.output_seq == session.last_persisted_seq {
|
||||
continue;
|
||||
}
|
||||
session.last_persisted_seq = session.output_seq;
|
||||
session.bytes_since_persist = 0;
|
||||
session.last_screen_persist_at = Instant::now();
|
||||
let screen = session.parser.screen();
|
||||
to_write.push((
|
||||
name.clone(),
|
||||
@@ -3818,10 +3948,19 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
|
||||
persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
|
||||
{
|
||||
eprintln!("screen flush for {name} failed: {err:#}");
|
||||
} else {
|
||||
mark_screen_persisted(state, &name, output_seq);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn mark_screen_persisted(state: &Arc<Mutex<ServerState>>, name: &str, output_seq: u64) {
|
||||
let mut locked = state.lock().expect("server state poisoned");
|
||||
if let Some(session) = locked.sessions.get_mut(name) {
|
||||
session.last_persisted_seq = session.last_persisted_seq.max(output_seq);
|
||||
}
|
||||
}
|
||||
|
||||
fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
|
||||
// Sessions removed here are dropped after the lock is released so their
|
||||
// shells are killed (PtyHandle::drop) without holding up the event loop.
|
||||
@@ -4024,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();
|
||||
@@ -4374,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(),
|
||||
@@ -4382,6 +4550,7 @@ mod tests {
|
||||
persistent: false,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
state.client_index.insert(client_id, "test".to_string());
|
||||
@@ -4437,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(),
|
||||
@@ -4445,6 +4615,7 @@ mod tests {
|
||||
persistent: false,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
state.client_index.insert(client_id, "test".to_string());
|
||||
@@ -4512,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(),
|
||||
@@ -4520,6 +4692,7 @@ mod tests {
|
||||
persistent: false,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
state.client_index.insert(client_id, "test".to_string());
|
||||
@@ -4651,6 +4824,7 @@ mod tests {
|
||||
Session {
|
||||
pty: None,
|
||||
parser: vt100::Parser::new(24, 80, 100),
|
||||
restored_screen: None,
|
||||
clients: HashMap::from([(
|
||||
client_id,
|
||||
ClientState {
|
||||
@@ -4689,6 +4863,7 @@ mod tests {
|
||||
persistent: false,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
// Keep the O(1) conn_id index in sync, matching `insert_client`.
|
||||
@@ -4747,6 +4922,7 @@ mod tests {
|
||||
Session {
|
||||
pty: None,
|
||||
parser: vt100::Parser::new(24, 80, 100),
|
||||
restored_screen: None,
|
||||
clients: HashMap::from([(
|
||||
client_id,
|
||||
ClientState {
|
||||
@@ -4785,6 +4961,7 @@ mod tests {
|
||||
persistent: false,
|
||||
bytes_since_persist: 0,
|
||||
last_persisted_seq: 0,
|
||||
last_screen_persist_at: Instant::now(),
|
||||
},
|
||||
);
|
||||
state.client_index.insert(client_id, "test".to_string());
|
||||
|
||||
+5
-4
@@ -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 {
|
||||
|
||||
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
@@ -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();
|
||||
|
||||
@@ -19,6 +19,46 @@ fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
|
||||
assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn package_check_verifies_only_artifacts_it_builds() {
|
||||
let makefile = include_str!("../Makefile");
|
||||
let verify = include_str!("../scripts/verify-release-artifacts.sh");
|
||||
assert!(
|
||||
verify.contains("if [ \"$#\" -gt 0 ]; then"),
|
||||
"release verifier must accept an explicit artifact list"
|
||||
);
|
||||
let package_check = makefile
|
||||
.split("package-check:")
|
||||
.nth(1)
|
||||
.and_then(|tail| tail.split("\ninstall:").next())
|
||||
.expect("package-check target");
|
||||
assert!(package_check.contains("package-release-linux"));
|
||||
assert!(package_check.contains("package-release-windows"));
|
||||
assert!(package_check.contains("dosh-linux-x86_64.tar.gz"));
|
||||
assert!(package_check.contains("dosh-windows-x86_64.zip"));
|
||||
assert!(
|
||||
!package_check.contains("dosh-macos-aarch64.tar.gz"),
|
||||
"Linux package-check does not build macOS artifacts, so it must not depend on one"
|
||||
);
|
||||
}
|
||||
|
||||
#[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");
|
||||
|
||||
Reference in New Issue
Block a user