Compare commits

...
19 Commits
Author SHA1 Message Date
DuProcess 1dcb33550e Expand SSH identity path tokens
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 18:18:43 -04:00
DuProcess 2c5b9ab30d Preserve symlinks in file copy
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 18:07:13 -04:00
DuProcess bfe75dfbd2 Harden file resume prefix validation
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 17:59:41 -04:00
DuProcess 4ea05fc14f Clarify proxied SSH UDP requirements
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 17:44:43 -04:00
DuProcess bae0d0ed56 Harden stream queues and hostile network tests
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 17:33:49 -04:00
DuProcess 99e318ca6b Harden reconnect paths under UDP churn
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 12:34:18 -04:00
DuProcess b2be4c2cb7 Retry transport receive during UDP churn
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 12:15:25 -04:00
DuProcess dbf96b68d3 Keep sessions alive through UDP churn
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 12:08:32 -04:00
DuProcess d74457d198 Preserve OpenSSH alias config for hosts
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 11:47:45 -04:00
DuProcess 03f5b88889 Bump version to 1.0.0-rc19
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 11:34:42 -04:00
DuProcess 5a29925422 Stamp release builds from exported git metadata 2026-07-11 11:34:23 -04:00
DuProcess 630b8dab30 Bump version to 1.0.0-rc18
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-11 11:29:56 -04:00
DuProcess b87f277e2d Use native installer for Windows updates 2026-07-11 11:29:41 -04:00
DuProcess 732f53f2d0 Keep release clippy gate clean
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 23:11:16 -04:00
DuProcess 6aa03cca63 Make package artifact checks deterministic
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 23:07:50 -04:00
DuProcess 58d8e34828 Bump version to 1.0.0-rc17
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 22:55:45 -04:00
DuProcess 47e446b433 Let dosh update refresh server installs 2026-07-10 22:55:29 -04:00
DuProcess e11ba1a652 Bump version to 1.0.0-rc16
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 22:49:14 -04:00
DuProcess 07c4b321d1 Harden persistent screen mirroring
ci / test (push) Has been cancelled
ci / fuzz-smoke (push) Has been cancelled
ci / windows-client (push) Has been cancelled
ci / package-release (linux-x86_64, ubuntu-latest) (push) Has been cancelled
ci / package-release (macos-aarch64, macos-14) (push) Has been cancelled
ci / package-release (macos-x86_64, macos-13) (push) Has been cancelled
ci / package-release (windows-x86_64, windows-latest) (push) Has been cancelled
ci / remote-bench (push) Has been cancelled
2026-07-10 22:48:32 -04:00
19 changed files with 1553 additions and 280 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]] [[package]]
name = "dosh" name = "dosh"
version = "1.0.0-rc15" version = "1.0.0-rc28"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"base64", "base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "dosh" name = "dosh"
version = "1.0.0-rc15" version = "1.0.0-rc28"
edition = "2024" edition = "2024"
license = "MIT" license = "MIT"
+3 -1
View File
@@ -36,7 +36,9 @@ persistence-check:
package-check: package-check:
$(MAKE) package-release-linux $(MAKE) package-release-linux
$(MAKE) package-release-windows $(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: install:
sh install.sh --from-current sh install.sh --from-current
+15
View File
@@ -44,6 +44,7 @@ dosh setup HOST
dosh HOST dosh HOST
dosh HOST COMMAND dosh HOST COMMAND
dosh update dosh update
dosh update --client|--server|--both
``` ```
Useful commands: Useful commands:
@@ -97,6 +98,20 @@ cache_attach_tickets = true
disconnect_status = 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 ## Rust Library
Dosh exposes a Rust transport for encrypted, reconnecting application streams. Dosh exposes a Rust transport for encrypted, reconnecting application streams.
+23 -7
View File
@@ -1,3 +1,4 @@
use std::env;
use std::process::Command; use std::process::Command;
fn git(args: &[&str]) -> Option<String> { fn git(args: &[&str]) -> Option<String> {
@@ -12,14 +13,29 @@ fn git(args: &[&str]) -> Option<String> {
fn main() { fn main() {
println!("cargo:rerun-if-changed=.git/HEAD"); println!("cargo:rerun-if-changed=.git/HEAD");
println!("cargo:rerun-if-changed=.git/index"); 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 hash = env::var("DOSH_BUILD_GIT_HASH")
let date = git(&["show", "-s", "--format=%cs", "HEAD"]).unwrap_or_else(|| "unknown".into()); .ok()
let dirty = Command::new("git") .filter(|value| !value.is_empty())
.args(["diff", "--quiet", "--ignore-submodules", "--"]) .or_else(|| git(&["rev-parse", "--short=12", "HEAD"]))
.status() .unwrap_or_else(|| "unknown".into());
.map(|status| if status.success() { "" } else { "+dirty" }) let date = env::var("DOSH_BUILD_COMMIT_DATE")
.unwrap_or(""); .ok()
.filter(|value| !value.is_empty())
.or_else(|| git(&["show", "-s", "--format=%cs", "HEAD"]))
.unwrap_or_else(|| "unknown".into());
let dirty = match env::var("DOSH_BUILD_GIT_DIRTY").ok().as_deref() {
Some("1" | "true" | "yes" | "+dirty") => "+dirty",
Some(_) => "",
None => Command::new("git")
.args(["diff", "--quiet", "--ignore-submodules", "--"])
.status()
.map(|status| if status.success() { "" } else { "+dirty" })
.unwrap_or(""),
};
println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}"); println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}");
println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}"); println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}");
+16
View File
@@ -52,6 +52,22 @@ else
release_dir="target/release" release_dir="target/release"
fi 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 [ "$os" = "windows" ]; then
if [ -n "$target" ]; then if [ -n "$target" ]; then
cargo build --release --target "$target" --bin dosh-client --bin dosh-bench cargo build --release --target "$target" --bin dosh-client --bin dosh-bench
+7 -4
View File
@@ -22,11 +22,14 @@ artifact_version() {
esac 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 failed=0
for artifact in \ for artifact in $artifacts; do
"$out_dir/dosh-linux-x86_64.tar.gz" \
"$out_dir/dosh-macos-aarch64.tar.gz" \
"$out_dir/dosh-windows-x86_64.zip"; do
if [ ! -f "$artifact" ]; then if [ ! -f "$artifact" ]; then
echo "missing release artifact: $artifact" >&2 echo "missing release artifact: $artifact" >&2
failed=1 failed=1
+836 -153
View File
File diff suppressed because it is too large Load Diff
+252 -75
View File
@@ -28,7 +28,7 @@ use dosh::protocol::{
TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope, TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
}; };
use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session}; 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 sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::fs; 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 /// 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. /// per-packet hot path while bounding how much fresh output a crash can lose.
const SCREEN_PERSIST_BYTES: usize = 4096; const SCREEN_PERSIST_BYTES: usize = 4096;
const SCREEN_PERSIST_MAX_AGE: Duration = Duration::from_millis(250);
const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300; const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300;
/// Sentinel `target_host` sent to the client on a server-initiated `StreamOpen` /// 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 { if config.persist_sessions {
let flush_state = Arc::clone(&state); let flush_state = Arc::clone(&state);
tokio::spawn(async move { 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 { loop {
interval.tick().await; interval.tick().await;
flush_persistent_screens(&flush_state); 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]; let mut buf = vec![0u8; 65535];
loop { 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 { if let Err(err) = handle_packet(&state, &socket, peer, &buf[..n]).await {
eprintln!("packet from {peer}: {err:#}"); eprintln!("packet from {peer}: {err:#}");
} }
@@ -356,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>>,
@@ -376,6 +382,18 @@ struct Session {
/// `output_seq` at the last successful screen mirror, so the periodic flush /// `output_seq` at the last successful screen mirror, so the periodic flush
/// skips sessions whose screen has not changed. /// skips sessions whose screen has not changed.
last_persisted_seq: u64, 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)] #[derive(Clone)]
@@ -503,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),
@@ -511,6 +530,7 @@ impl ServerState {
persistent, persistent,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
}, },
); );
Ok(()) Ok(())
@@ -631,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),
@@ -648,6 +676,7 @@ impl ServerState {
persistent: true, persistent: true,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: output_seq, last_persisted_seq: output_seq,
last_screen_persist_at: Instant::now(),
}, },
); );
eprintln!( eprintln!(
@@ -751,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"
} }
@@ -1178,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,
@@ -1326,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(
@@ -1462,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(
@@ -1575,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,
@@ -1596,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,
@@ -1736,10 +1781,10 @@ async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packe
let (key, _) = find_client_decrypt_key(state, &packet.header)?; let (key, _) = find_client_decrypt_key(state, &packet.header)?;
let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?; let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
let mut locked = state.lock().expect("server state poisoned"); let mut locked = state.lock().expect("server state poisoned");
if let Some(client) = locked.client_mut(&packet.header.conn_id) { if let Some(client) = locked.client_mut(&packet.header.conn_id)
if !client.replay.accept(packet.header.seq) { && !client.replay.accept(packet.header.seq)
return Ok(()); {
} return Ok(());
} }
locked.remove_client_everywhere(&packet.header.conn_id); locked.remove_client_everywhere(&packet.header.conn_id);
Ok(()) Ok(())
@@ -2521,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(
@@ -2634,6 +2681,46 @@ async fn handle_file_request(
send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok) send_file_response_to_client(state, socket, client_id, stream_id, FileResponse::Ok)
.await .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 } => { FileRequest::Remove { path, recursive } => {
let path = clean_remote_path(&path, home)?; let path = clean_remote_path(&path, home)?;
let metadata = let metadata =
@@ -2671,32 +2758,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)
.write(true) .with_context(|| format!("stat {}", final_path.display()))?
.append(true) .len();
.open(&final_path) let written = original_len.min(size);
.with_context(|| format!("open {}", final_path.display()))?; if written > 0 {
let written = file.metadata()?.len().min(size); let mut prefix = fs::File::open(&final_path)?;
file.set_len(written) hash_exact_prefix(&mut prefix, written, &mut hasher)?;
.with_context(|| format!("truncate {}", final_path.display()))?; }
if written > 0 { let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
let mut prefix = fs::File::open(&final_path)?; let mut file = fs::OpenOptions::new()
hash_exact_prefix(&mut prefix, written, &mut hasher)?; .read(true)
} .append(true)
file.seek(SeekFrom::Start(written))?; .open(&final_path)
(file, None, written, false) .with_context(|| format!("open {}", final_path.display()))?;
} else { file.seek(SeekFrom::Start(written))?;
let temp_path = upload_temp_path(&final_path, stream_id); (
let file = fs::OpenOptions::new() file,
.create(true) None,
.truncate(true) written,
.write(true) false,
.open(&temp_path) Some(original_len),
.with_context(|| format!("create {}", temp_path.display()))?; prefix_sha256,
(file, Some(temp_path), 0, true) )
}; } 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,
@@ -2704,23 +2800,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
} }
@@ -2742,8 +2835,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 {}",
@@ -2754,9 +2845,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()
@@ -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 { if let Some(modified_secs) = active.modified_secs {
filetime::set_file_mtime( filetime::set_file_mtime(
&active.final_path, &active.final_path,
@@ -2796,9 +2898,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,
@@ -2807,14 +2920,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 {
@@ -2996,6 +3105,17 @@ fn set_mode(path: &Path, mode: u32) -> Result<()> {
fs::set_permissions(path, permissions).with_context(|| format!("chmod {}", path.display())) 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 { fn upload_temp_path(final_path: &Path, stream_id: u64) -> PathBuf {
let name = final_path let name = final_path
.file_name() .file_name()
@@ -3199,7 +3319,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);
} }
@@ -3444,6 +3567,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());
@@ -3451,16 +3575,20 @@ async fn broadcast_output(
session.recent.pop_front(); session.recent.pop_front();
} }
// For a persistent session, mirror the screen to disk periodically so a // For a persistent session, mirror the screen to disk periodically so a
// post-restart reattach repaints. Throttled by accumulated bytes to keep // post-restart reattach repaints. Throttled by accumulated bytes or a
// the (atomic) write off the per-packet hot path. The actual file write // short wall-clock age so low-throughput prompt/TUI changes persist
// happens after the lock is dropped. // quickly while large bursts still avoid per-packet disk writes. The
// actual file write happens after the lock is dropped.
if session.persistent { if session.persistent {
session.bytes_since_persist = session session.bytes_since_persist = session
.bytes_since_persist .bytes_since_persist
.saturating_add(output.bytes.len()); .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.bytes_since_persist = 0;
session.last_persisted_seq = output_seq; session.last_screen_persist_at = now;
let screen = session.parser.screen(); let screen = session.parser.screen();
persist_screen = Some(( persist_screen = Some((
output.session.clone(), output.session.clone(),
@@ -3519,6 +3647,8 @@ async fn broadcast_output(
persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot) persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
{ {
eprintln!("screen persist for {name} failed: {err:#}"); eprintln!("screen persist for {name} failed: {err:#}");
} else {
mark_screen_persisted(state, &name, output_seq);
} }
} }
for (endpoint, packet) in sends { 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 { if !session.persistent || session.output_seq == session.last_persisted_seq {
continue; continue;
} }
session.last_persisted_seq = session.output_seq;
session.bytes_since_persist = 0; session.bytes_since_persist = 0;
session.last_screen_persist_at = Instant::now();
let screen = session.parser.screen(); let screen = session.parser.screen();
to_write.push(( to_write.push((
name.clone(), 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) persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
{ {
eprintln!("screen flush for {name} failed: {err:#}"); 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>>) { fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
// Sessions removed here are dropped after the lock is released so their // Sessions removed here are dropped after the lock is released so their
// shells are killed (PtyHandle::drop) without holding up the event loop. // 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); 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();
@@ -4374,6 +4541,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(),
@@ -4382,6 +4550,7 @@ mod tests {
persistent: false, persistent: false,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now(),
}, },
); );
state.client_index.insert(client_id, "test".to_string()); state.client_index.insert(client_id, "test".to_string());
@@ -4437,6 +4606,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(),
@@ -4445,6 +4615,7 @@ mod tests {
persistent: false, persistent: false,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now(),
}, },
); );
state.client_index.insert(client_id, "test".to_string()); state.client_index.insert(client_id, "test".to_string());
@@ -4512,6 +4683,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(),
@@ -4520,6 +4692,7 @@ mod tests {
persistent: false, persistent: false,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now(),
}, },
); );
state.client_index.insert(client_id, "test".to_string()); state.client_index.insert(client_id, "test".to_string());
@@ -4651,6 +4824,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 {
@@ -4689,6 +4863,7 @@ mod tests {
persistent: false, persistent: false,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now(),
}, },
); );
// Keep the O(1) conn_id index in sync, matching `insert_client`. // Keep the O(1) conn_id index in sync, matching `insert_client`.
@@ -4747,6 +4922,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 {
@@ -4785,6 +4961,7 @@ mod tests {
persistent: false, persistent: false,
bytes_since_persist: 0, bytes_since_persist: 0,
last_persisted_seq: 0, last_persisted_seq: 0,
last_screen_persist_at: Instant::now(),
}, },
); );
state.client_index.insert(client_id, "test".to_string()); state.client_index.insert(client_id, "test".to_string());
+5 -4
View File
@@ -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 {
+6
View File
@@ -150,6 +150,12 @@ pub struct ClientConfig {
#[derive(Debug, Clone, Default, Serialize, Deserialize)] #[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct HostConfig { pub struct HostConfig {
pub ssh: Option<String>, 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 user: Option<String>,
pub dosh_host: Option<String>, pub dosh_host: Option<String>,
pub port: Option<u16>, pub port: Option<u16>,
+37 -7
View File
@@ -18,6 +18,14 @@ pub enum FileRequest {
path: String, path: String,
mode: Option<u32>, mode: Option<u32>,
}, },
Readlink {
path: String,
},
Symlink {
path: String,
target: String,
overwrite: bool,
},
Remove { Remove {
path: String, path: String,
recursive: bool, recursive: bool,
@@ -46,13 +54,35 @@ 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>,
},
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)] #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+32 -1
View File
@@ -23,12 +23,14 @@ use std::io::{Read, Write};
use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd}; use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
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::sync::Mutex;
use std::time::Duration; use std::time::Duration;
/// One-byte commands a server sends to a holder over its control socket after /// One-byte commands a server sends to a holder over its control socket after
/// the holder has handed back the master fd. /// the holder has handed back the master fd.
const HOLDER_CMD_SHUTDOWN: u8 = b'X'; const HOLDER_CMD_SHUTDOWN: u8 = b'X';
const HOLDER_STARTUP_TIMEOUT: Duration = Duration::from_millis(750); 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 /// 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. /// 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. // No holder runtime for this session (non-persistent): nothing to do.
return Ok(()); 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); let mut buf = Vec::with_capacity(snapshot.len() + 32);
buf.extend_from_slice(&cols.to_be_bytes()); buf.extend_from_slice(&cols.to_be_bytes());
buf.extend_from_slice(&rows.to_be_bytes()); buf.extend_from_slice(&rows.to_be_bytes());
buf.extend_from_slice(&output_seq.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.len() as u32).to_be_bytes());
buf.extend_from_slice(snapshot); 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. /// A restored screen for a re-adopted session.
@@ -619,6 +636,20 @@ mod tests {
assert_eq!(loaded.snapshot, snap); 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] #[test]
fn load_screen_absent_is_none() { fn load_screen_absent_is_none() {
let tmp = tempfile::tempdir().unwrap(); let tmp = tempfile::tempdir().unwrap();
+6 -2
View File
@@ -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,
+43 -4
View File
@@ -19,7 +19,7 @@ use crate::protocol::{
self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData, self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData,
StreamOpen, StreamOpenOk, StreamOpenReject, StreamWindowAdjust, 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 anyhow::{Result, anyhow, bail};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::net::SocketAddr; use std::net::SocketAddr;
@@ -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)
@@ -852,7 +854,11 @@ impl DoshTransport {
pub async fn recv(&mut self) -> Result<SessionEvent> { pub async fn recv(&mut self) -> Result<SessionEvent> {
let mut buf = vec![0u8; 65535]; let mut buf = vec![0u8; 65535];
loop { 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]) { let packet = match protocol::decode(&buf[..n]) {
Ok(packet) => packet, Ok(packet) => packet,
Err(_) => continue, 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 { fn initial_stream_id(role: SessionRole) -> u64 {
match role { match role {
SessionRole::Client => 1, SessionRole::Client => 1,
@@ -1033,6 +1043,35 @@ mod tests {
assert!(service_target("../bad").is_err()); 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] #[test]
fn service_registry_matches_registered_service_opens() { fn service_registry_matches_registered_service_opens() {
let mut registry = ServiceRegistry::new(); let mut registry = ServiceRegistry::new();
+105 -13
View File
@@ -1,6 +1,10 @@
//! UDP error classification shared by Dosh clients, servers, and embedders. //! 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!( matches!(
err.kind(), err.kind(),
std::io::ErrorKind::Interrupted 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) ) || 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)] #[cfg(unix)]
fn is_transient_udp_os_error(code: i32) -> bool { fn is_transient_udp_os_error(code: i32) -> bool {
matches!( matches!(
@@ -48,20 +104,56 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
#[cfg(test)] #[cfg(test)]
mod tests { 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] #[test]
fn classifies_portable_transient_send_errors() { fn classifies_portable_transient_udp_errors() {
for kind in [ for kind in [
std::io::ErrorKind::Interrupted, std::io::ErrorKind::Interrupted,
std::io::ErrorKind::TimedOut, std::io::ErrorKind::TimedOut,
std::io::ErrorKind::WouldBlock, 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( let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
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)] #[cfg(unix)]
@@ -78,9 +170,9 @@ mod tests {
libc::ENETUNREACH, libc::ENETUNREACH,
libc::ETIMEDOUT, libc::ETIMEDOUT,
] { ] {
assert!(is_transient_udp_send_error( let err = std::io::Error::from_raw_os_error(code);
&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 [ for code in [
10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065, 10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065,
] { ] {
assert!(is_transient_udp_send_error( let err = std::io::Error::from_raw_os_error(code);
&std::io::Error::from_raw_os_error(code) assert!(is_transient_udp_error(&err));
)); assert!(is_transient_udp_send_error(&err));
} }
} }
} }
+14 -3
View File
@@ -29,7 +29,7 @@ use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, UdpSocket}; use std::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 {
+111 -4
View File
@@ -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 {
@@ -980,6 +1003,8 @@ fn native_file_copy_recursive_round_trip() {
fs::create_dir_all(src.join("nested")).unwrap(); fs::create_dir_all(src.join("nested")).unwrap();
fs::write(src.join("root.txt"), b"root file\n").unwrap(); fs::write(src.join("root.txt"), b"root file\n").unwrap();
fs::write(src.join("nested/child.txt"), b"child 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 client_bin = env!("CARGO_BIN_EXE_dosh-client");
let upload = Command::new(client_bin) 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(), fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
"child file\n" "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) let remove = Command::new(client_bin)
.arg("--dosh-host") .arg("--dosh-host")
@@ -1219,6 +1252,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();
+40
View File
@@ -19,6 +19,46 @@ fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\"")); 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] #[test]
fn systemd_service_does_not_sandbox_remote_shells() { fn systemd_service_does_not_sandbox_remote_shells() {
let service = include_str!("../packaging/systemd/dosh-server.service"); let service = include_str!("../packaging/systemd/dosh-server.service");