Compare commits

...
27 Commits
Author SHA1 Message Date
DuProcess 3d5bb126c5 Trace ticket reattach recovery
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-12 12:15:25 -04:00
DuProcess 37632a96b7 Filter trace reports to latest run
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-12 12:09:37 -04:00
DuProcess e69be4fdf0 Reconnect before input after local sleep
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-12 12:01:46 -04:00
DuProcess fdbd58b628 Fetch remote server trace reports
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-12 11:55:23 -04:00
DuProcess c65aba9d7a Repaint terminal after local sleep
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-12 11:48:18 -04:00
DuProcess 818b481154 Cover terminal mouse input filtering
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-12 01:08:38 -04:00
DuProcess 70650e221b Flag trace terminal input anomalies
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-12 01:04:37 -04:00
DuProcess 01e8870578 Add trace report summarizer
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-12 00:59:15 -04:00
DuProcess 99d54899bd Add trace command for reconnect debugging
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-12 00:52:47 -04:00
DuProcess 80e922957e Trace reconnect mouse input handling
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-12 00:40:28 -04:00
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
21 changed files with 3485 additions and 299 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]] [[package]]
name = "dosh" name = "dosh"
version = "1.0.0-rc16" version = "1.0.0-rc37"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"base64", "base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "dosh" name = "dosh"
version = "1.0.0-rc16" version = "1.0.0-rc37"
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
+41
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:
@@ -67,6 +68,32 @@ dosh restart HOST
Agent forwarding is opt-in with `-A` and must also be enabled on the server. Agent forwarding is opt-in with `-A` and must also be enabled on the server.
File copy must be enabled by the server config. File copy must be enabled by the server config.
## Tracing
For terminal/reconnect bugs, run a client with:
```sh
dosh trace HOST
```
The client log path is printed before the session starts. To choose it:
```sh
dosh trace --client-log /tmp/dosh-client.log HOST
```
Summarize collected traces with:
```sh
dosh trace report HOST --client-log /tmp/dosh-client.log
```
When `HOST` is given, Dosh fetches `/tmp/dosh-server.log` over SSH before
building the report. Set `DOSH_TRACE=/tmp/dosh-server.log` on `dosh-server` for
matching server events. Reports default to the latest traced client/server
process run; add `--all-runs` to include older appended log entries. Trace byte
prefixes are enabled for `dosh trace`, so use it only for short reproductions.
## VS Code ## VS Code
Dosh can carry VS Code Remote-SSH through its transport: Dosh can carry VS Code Remote-SSH through its transport:
@@ -97,6 +124,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
+2491 -177
View File
File diff suppressed because it is too large Load Diff
+341 -70
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;
@@ -67,6 +67,15 @@ static AGENT_SOCK_COUNTER: AtomicU64 = AtomicU64::new(0);
/// before it is swept by the cleanup task. /// before it is swept by the cleanup task.
const NATIVE_HANDSHAKE_TTL_SECS: u64 = 30; const NATIVE_HANDSHAKE_TTL_SECS: u64 = 30;
fn hex_id(id: [u8; 16]) -> String {
let mut out = String::with_capacity(32);
for byte in id {
use std::fmt::Write as _;
let _ = write!(&mut out, "{byte:02x}");
}
out
}
#[derive(Debug, Parser)] #[derive(Debug, Parser)]
#[command( #[command(
name = "dosh-server", name = "dosh-server",
@@ -191,6 +200,13 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
.with_context(|| format!("bind {bind}"))?, .with_context(|| format!("bind {bind}"))?,
); );
eprintln!("dosh-server listening on {bind}"); eprintln!("dosh-server listening on {bind}");
dosh::trace::event(
"server.start",
&[
("version", dosh::build_info::VERSION.to_string()),
("bind", bind.clone()),
],
);
let (pty_tx, mut pty_rx) = mpsc::unbounded_channel(); let (pty_tx, mut pty_rx) = mpsc::unbounded_channel();
let state = Arc::new(Mutex::new(ServerState::new( let state = Arc::new(Mutex::new(ServerState::new(
@@ -262,7 +278,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:#}");
} }
@@ -357,6 +377,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>>,
@@ -383,6 +404,14 @@ struct Session {
last_screen_persist_at: Instant, last_screen_persist_at: Instant,
} }
#[derive(Clone)]
struct RestoredScreenSnapshot {
cols: u16,
rows: u16,
output_seq: u64,
snapshot: Vec<u8>,
}
#[derive(Clone)] #[derive(Clone)]
struct ClientState { struct ClientState {
endpoint: SocketAddr, endpoint: SocketAddr,
@@ -508,6 +537,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),
@@ -637,15 +667,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),
@@ -758,6 +796,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"
} }
@@ -1185,7 +1236,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,
@@ -1333,7 +1384,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(
@@ -1412,6 +1463,7 @@ async fn handle_ticket_attach(
peer: SocketAddr, peer: SocketAddr,
body: Vec<u8>, body: Vec<u8>,
) -> Result<()> { ) -> Result<()> {
dosh::trace::event("server.ticket_attach_start", &[("peer", peer.to_string())]);
let env: TicketAttachEnvelope = protocol::from_body(&body)?; let env: TicketAttachEnvelope = protocol::from_body(&body)?;
let opened = { let opened = {
let locked = state.lock().expect("server state poisoned"); let locked = state.lock().expect("server state poisoned");
@@ -1435,10 +1487,22 @@ async fn handle_ticket_attach(
}; };
let (ticket, request_plain) = match opened { let (ticket, request_plain) = match opened {
Ok(opened) => opened, Ok(opened) => opened,
Err(err) => return send_reject(socket, peer, &err.to_string()).await, Err(err) => {
let reason = err.to_string();
dosh::trace::event("server.ticket_attach_reject", &[("reason", reason.clone())]);
return send_reject(socket, peer, &reason).await;
}
}; };
let req: TicketAttachBody = protocol::from_body(&request_plain)?; let req: TicketAttachBody = protocol::from_body(&request_plain)?;
if req.session != ticket.session || req.mode != ticket.mode { if req.session != ticket.session || req.mode != ticket.mode {
dosh::trace::event(
"server.ticket_attach_reject",
&[
("session", req.session.clone()),
("mode", req.mode.clone()),
("reason", "ticket scope mismatch".to_string()),
],
);
return send_reject(socket, peer, "ticket scope mismatch").await; return send_reject(socket, peer, "ticket scope mismatch").await;
} }
@@ -1469,7 +1533,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(
@@ -1509,7 +1573,18 @@ async fn handle_ticket_attach(
}; };
let (client_id, output_seq, snapshot) = match attached { let (client_id, output_seq, snapshot) = match attached {
Ok(attached) => attached, Ok(attached) => attached,
Err(err) => return send_reject(socket, peer, &err.to_string()).await, Err(err) => {
let reason = err.to_string();
dosh::trace::event(
"server.ticket_attach_reject",
&[
("session", req.session.clone()),
("mode", req.mode.clone()),
("reason", reason.clone()),
],
);
return send_reject(socket, peer, &reason).await;
}
}; };
let ok = AttachOk { let ok = AttachOk {
@@ -1540,6 +1615,16 @@ async fn handle_ticket_attach(
let body = protocol::to_body(&envelope)?; let body = protocol::to_body(&envelope)?;
let out = protocol::encode_plain(PacketKind::AttachOk, client_id, 1, 0, &body)?; let out = protocol::encode_plain(PacketKind::AttachOk, client_id, 1, 0, &body)?;
let _ = send_udp(socket, &out, peer).await?; let _ = send_udp(socket, &out, peer).await?;
dosh::trace::event(
"server.ticket_attach_ok",
&[
("session", ok.session.clone()),
("mode", ok.mode.clone()),
("client", hex_id(client_id)),
("output_seq", output_seq.to_string()),
("bytes", ok.snapshot.len().to_string()),
],
);
Ok(()) Ok(())
} }
@@ -1567,9 +1652,22 @@ async fn handle_resume(
peer: SocketAddr, peer: SocketAddr,
packet: &protocol::Packet, packet: &protocol::Packet,
) -> Result<()> { ) -> Result<()> {
dosh::trace::event(
"server.resume_start",
&[
("peer", peer.to_string()),
("client", hex_id(packet.header.conn_id)),
("seq", packet.header.seq.to_string()),
("ack", packet.header.ack.to_string()),
],
);
let (key, session_name) = match find_client_decrypt_key(state, &packet.header) { let (key, session_name) = match find_client_decrypt_key(state, &packet.header) {
Ok(found) => found, Ok(found) => found,
Err(_) => { Err(_) => {
dosh::trace::event(
"server.resume_unknown_client",
&[("client", hex_id(packet.header.conn_id))],
);
return send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client") return send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client")
.await; .await;
} }
@@ -1582,20 +1680,40 @@ 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) {
} dosh::trace::event(
client.endpoint = peer; "server.resume_replay_drop",
client.last_acked = req.last_rendered_seq; &[
client.cols = req.cols; ("session", req.session.clone()),
client.rows = req.rows; ("seq", packet.header.seq.to_string()),
client.last_seen = Instant::now(); ],
client.send_seq += 1; );
if mode_allows_terminal_updates(&client.mode) { return Ok(());
}
if client.endpoint != peer {
dosh::trace::event(
"server.resume_roam",
&[
("session", req.session.clone()),
("from", client.endpoint.to_string()),
("to", peer.to_string()),
],
);
}
client.endpoint = peer;
client.last_acked = req.last_rendered_seq;
client.cols = req.cols;
client.rows = req.rows;
client.last_seen = Instant::now();
client.send_seq += 1;
(client.send_seq, mode_allows_terminal_updates(&client.mode))
};
if allow_terminal_updates {
apply_terminal_size( apply_terminal_size(
session.pty.as_ref(), session.pty.as_ref(),
&mut session.parser, &mut session.parser,
@@ -1603,8 +1721,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,
@@ -1624,6 +1742,14 @@ async fn handle_resume(
&body, &body,
)?; )?;
let _ = send_udp(socket, &out, peer).await?; let _ = send_udp(socket, &out, peer).await?;
dosh::trace::event(
"server.resume_ok",
&[
("session", frame.session),
("output_seq", frame.output_seq.to_string()),
("bytes", frame.bytes.len().to_string()),
],
);
Ok(()) Ok(())
} }
@@ -1635,6 +1761,7 @@ async fn handle_input(
let (key, session_name) = find_client_decrypt_key(state, &packet.header)?; let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?; let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
let input: Input = protocol::from_body(&body)?; let input: Input = protocol::from_body(&body)?;
let input_summary = dosh::trace::bytes_summary(&input.bytes);
let mut locked = state.lock().expect("server state poisoned"); let mut locked = state.lock().expect("server state poisoned");
let session = locked let session = locked
.sessions .sessions
@@ -1645,13 +1772,37 @@ async fn handle_input(
.get_mut(&packet.header.conn_id) .get_mut(&packet.header.conn_id)
.ok_or_else(|| anyhow!("unknown client"))?; .ok_or_else(|| anyhow!("unknown client"))?;
if !client.replay.accept(packet.header.seq) { if !client.replay.accept(packet.header.seq) {
dosh::trace::event(
"server.input_replay_drop",
&[
("session", session_name.clone()),
("seq", packet.header.seq.to_string()),
("summary", input_summary),
],
);
return Ok(()); return Ok(());
} }
if client.endpoint != peer { if client.endpoint != peer {
dosh::trace::event(
"server.input_roam",
&[
("session", session_name.clone()),
("from", client.endpoint.to_string()),
("to", peer.to_string()),
],
);
client.endpoint = peer; client.endpoint = peer;
} }
client.last_seen = Instant::now(); client.last_seen = Instant::now();
if !mode_allows_terminal_updates(&client.mode) { if !mode_allows_terminal_updates(&client.mode) {
dosh::trace::event(
"server.input_rejected_mode",
&[
("session", session_name),
("mode", client.mode.clone()),
("summary", input_summary),
],
);
return Ok(()); return Ok(());
} }
session session
@@ -1659,6 +1810,14 @@ async fn handle_input(
.as_ref() .as_ref()
.context("terminal session has no pty")? .context("terminal session has no pty")?
.write_all(&input.bytes)?; .write_all(&input.bytes)?;
dosh::trace::event(
"server.pty_write",
&[
("session", session_name),
("seq", packet.header.seq.to_string()),
("summary", input_summary),
],
);
Ok(()) Ok(())
} }
@@ -1743,10 +1902,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(())
@@ -2528,8 +2687,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(
@@ -2641,6 +2802,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 =
@@ -2678,32 +2879,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,
@@ -2711,23 +2921,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
} }
@@ -2749,8 +2956,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 {}",
@@ -2761,9 +2966,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()
@@ -2776,6 +2989,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,
@@ -2803,9 +3019,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,
@@ -2814,14 +3041,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 {
@@ -3003,6 +3226,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()
@@ -3206,7 +3440,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);
} }
@@ -3451,6 +3688,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());
@@ -4046,6 +4284,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();
@@ -4396,6 +4662,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(),
@@ -4460,6 +4727,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(),
@@ -4536,6 +4804,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(),
@@ -4676,6 +4945,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 {
@@ -4773,6 +5043,7 @@ mod tests {
Session { Session {
pty: None, pty: None,
parser: vt100::Parser::new(24, 80, 100), parser: vt100::Parser::new(24, 80, 100),
restored_screen: None,
clients: HashMap::from([( clients: HashMap::from([(
client_id, client_id,
ClientState { ClientState {
+5 -4
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)]
+1
View File
@@ -22,5 +22,6 @@ pub mod protocol;
pub mod pty; pub mod pty;
pub mod server; pub mod server;
pub mod ssh_agent; pub mod ssh_agent;
pub mod trace;
pub mod transport; pub mod transport;
pub mod udp; pub mod udp;
+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,
+161
View File
@@ -0,0 +1,161 @@
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::PathBuf;
use std::sync::{Mutex, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};
static TRACE_FILE: OnceLock<Option<Mutex<File>>> = OnceLock::new();
static TRACE_BYTES: OnceLock<bool> = OnceLock::new();
pub fn enabled() -> bool {
TRACE_FILE.get_or_init(open_trace_file).is_some()
}
pub fn bytes_enabled() -> bool {
*TRACE_BYTES.get_or_init(|| truthy_env("DOSH_TRACE_BYTES"))
}
pub fn event(name: &str, fields: &[(&str, String)]) {
let Some(file) = TRACE_FILE.get_or_init(open_trace_file) else {
return;
};
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.unwrap_or(0);
let mut line = format!(
"ts_ms={} pid={} event={}",
now,
std::process::id(),
shell_escape(name)
);
for (key, value) in fields {
line.push(' ');
line.push_str(key);
line.push('=');
line.push_str(&shell_escape(value));
}
line.push('\n');
if let Ok(mut file) = file.lock() {
let _ = file.write_all(line.as_bytes());
let _ = file.flush();
}
}
pub fn bytes_summary(bytes: &[u8]) -> String {
let mut parts = vec![
format!("len={}", bytes.len()),
format!(
"esc={}",
contains_byte(bytes, 0x1b) || contains_byte(bytes, 0x9b)
),
format!("focus={}", has_focus_report(bytes)),
format!("mouseish={}", looks_mouseish(bytes)),
format!("printable={}", printable_count(bytes)),
];
if bytes_enabled() {
parts.push(format!("hex={}", hex_prefix(bytes, 160)));
}
parts.join(",")
}
fn open_trace_file() -> Option<Mutex<File>> {
let raw = std::env::var_os("DOSH_TRACE")?;
let normalized = raw.to_string_lossy().to_ascii_lowercase();
if normalized.is_empty() || matches!(normalized.as_str(), "0" | "false" | "off") {
return None;
}
let path = if matches!(normalized.as_str(), "1" | "true" | "on") {
default_trace_path()
} else {
PathBuf::from(raw)
};
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.ok()
.map(Mutex::new)
}
fn default_trace_path() -> PathBuf {
let root = dirs::cache_dir()
.unwrap_or_else(std::env::temp_dir)
.join("dosh")
.join("trace");
let exe = std::env::current_exe()
.ok()
.and_then(|path| {
path.file_stem()
.map(|stem| stem.to_string_lossy().to_string())
})
.unwrap_or_else(|| "dosh".to_string());
root.join(format!("{}-{}.log", exe, std::process::id()))
}
fn truthy_env(name: &str) -> bool {
std::env::var_os(name).is_some_and(|value| {
let normalized = value.to_string_lossy().to_ascii_lowercase();
!normalized.is_empty() && !matches!(normalized.as_str(), "0" | "false" | "off")
})
}
fn shell_escape(value: &str) -> String {
if value.bytes().all(|byte| {
byte.is_ascii_alphanumeric()
|| matches!(byte, b'.' | b'-' | b'_' | b':' | b'/' | b',' | b'=')
}) {
return value.to_string();
}
let mut out = String::from("'");
for ch in value.chars() {
if ch == '\'' {
out.push_str("'\\''");
} else {
out.push(ch);
}
}
out.push('\'');
out
}
fn contains_byte(bytes: &[u8], needle: u8) -> bool {
bytes.contains(&needle)
}
fn has_focus_report(bytes: &[u8]) -> bool {
bytes
.windows(3)
.any(|window| matches!(window, b"\x1b[I" | b"\x1b[O"))
|| bytes
.windows(2)
.any(|window| matches!(window, b"\x9bI" | b"\x9bO"))
}
fn looks_mouseish(bytes: &[u8]) -> bool {
bytes.windows(3).any(|window| window == b"\x1b[<")
|| bytes.windows(2).any(|window| window == b"\x9b<")
|| bytes.iter().filter(|byte| **byte == b';').take(3).count() >= 2
}
fn printable_count(bytes: &[u8]) -> usize {
bytes
.iter()
.filter(|byte| byte.is_ascii_graphic() || **byte == b' ')
.count()
}
fn hex_prefix(bytes: &[u8], max: usize) -> String {
let mut out = String::with_capacity(bytes.len().min(max) * 2);
for byte in bytes.iter().take(max) {
use std::fmt::Write as _;
let _ = write!(&mut out, "{byte:02x}");
}
if bytes.len() > max {
out.push_str("...");
}
out
}
+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");