Compare commits

...

28 Commits

Author SHA1 Message Date
DuProcess ff2b7fff2b Bump release candidate to rc7
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-07 14:58:13 -04:00
DuProcess f4879090f6 Keep terminal focus reports local 2026-07-07 14:57:53 -04:00
DuProcess 37ec9fc520 Bump release candidate to rc6
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-07 12:01:03 -04:00
DuProcess f3c7ec225d Repaint terminal after idle tab return 2026-07-07 12:00:40 -04:00
DuProcess d3c40a06ac Bump release candidate to rc5
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-07 11:49:41 -04:00
DuProcess 6b12b3dfe0 Harden terminal reconnect packet handling 2026-07-07 11:48:59 -04:00
DuProcess b90c34dc17 Bump release candidate to rc4
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-05 18:00:59 -04:00
DuProcess f205e0c128 Harden native auth and SDK datagram handling 2026-07-05 18:00:31 -04:00
DuProcess 92c94176bd Bump release candidate to rc3
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-05 09:39:21 -04:00
DuProcess 4b2e9a62d5 Harden embedded transport 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-05 09:37:44 -04:00
DuProcess 689d20f7e7 Bump release candidate to rc2
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-04 23:00:18 -04:00
DuProcess 252f081a61 Suppress stale terminal mouse reports after reconnect
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-04 22:56:32 -04:00
DuProcess b1e8326b0a Stop local benchmark holder leaks 2026-07-04 22:31:14 -04:00
DuProcess 3b4931aa8f Keep server alive on transient UDP send errors 2026-07-04 22:29:52 -04:00
DuProcess 5c54601cdf Prepare Dosh 1.0.0 rc1
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-04 16:58:15 -04:00
DuProcess c085137250 Add hostile network TUI soak gate
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-04 14:37:23 -04:00
DuProcess 237ad52bef Flush queued input after reconnect contact
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-03 16:35:22 -04:00
DuProcess a8ba852f16 Keep Dosh server service alive
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-03 14:51:47 -04:00
DuProcess f1a3de7730 Rate limit terminal gap resync
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-03 10:19:53 -04:00
DuProcess 80699dcf9d Stabilize terminal frame 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-03 10:06:23 -04:00
DuProcess 274c0f505e Stop sandboxing remote shells
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-02 20:03:42 -04:00
DuProcess b393c0e5d5 Allow netlink for terminal network monitors
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-02 19:40:30 -04:00
DuProcess d5bc8e3c64 Recover terminal output gaps with resume repaint
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-02 19:29:20 -04:00
DuProcess c40f5459ba Keep live TUI output raw under retransmit pressure
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-02 19:21:43 -04:00
DuProcess 601510687e Prepare Dosh 0.1.9 release
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-02 19:07:46 -04:00
DuProcess 60387e9222 Install Dosh binaries atomically
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-02 18:41:02 -04:00
DuProcess a48aa15308 Preserve raw terminal output for TUIs
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-02 18:34:09 -04:00
DuProcess 691b58e531 Harden update and release tooling
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-01 19:48:46 -04:00
23 changed files with 1680 additions and 321 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]]
name = "dosh"
version = "0.1.7"
version = "1.0.0-rc7"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dosh"
version = "0.1.7"
version = "1.0.0-rc7"
edition = "2024"
license = "MIT"
+20 -1
View File
@@ -1,4 +1,4 @@
.PHONY: build test fmt clippy release-check install package-release package-release-linux package-release-windows publish-release bench-report bench-local bench-local-json bench-docker-ssh bench-docker-mosh fuzz-smoke fuzz-deep soak-local tui-harness
.PHONY: build test fmt clippy release-check 1.0-check reconnect-check hostile-network-check persistence-check package-check install package-release package-release-linux package-release-windows publish-release bench-report bench-local bench-local-json bench-docker-ssh bench-docker-mosh fuzz-smoke fuzz-deep soak-local tui-harness
build:
cargo build --release
@@ -19,6 +19,25 @@ release-check:
cargo test
$(MAKE) tui-harness
1.0-check: release-check reconnect-check hostile-network-check persistence-check package-check
reconnect-check:
cargo test --test integration_smoke resume_updates_udp_endpoint_for_roaming -- --nocapture
DOSH_SOAK_SECONDS=$${DOSH_SOAK_SECONDS:-300} cargo test --test integration_smoke sleep_roaming_soak_30m -- --ignored --nocapture
hostile-network-check:
cargo test --test hostile_network -- --nocapture
DOSH_BADNET_SOAK_SECONDS=$${DOSH_BADNET_SOAK_SECONDS:-300} cargo test --test hostile_network bad_network_tui_work_soak_30m -- --ignored --nocapture
persistence-check:
cargo test --test integration_smoke session_survives_server_restart_same_shell_and_screen -- --nocapture
cargo test --test integration_smoke multiple_persistent_named_sessions_survive_restart_independently -- --nocapture
package-check:
$(MAKE) package-release-linux
$(MAKE) package-release-windows
sh scripts/verify-release-artifacts.sh
install:
sh install.sh --from-current
+19 -5
View File
@@ -141,6 +141,20 @@ function Verify-ArchiveVersion($ExtractDir, $Url) {
}
}
function Install-Binary($Source, $Destination) {
$dir = Split-Path -Parent $Destination
$name = Split-Path -Leaf $Destination
$tmp = Join-Path $dir ".$name.tmp.$PID"
Copy-Item $Source $tmp -Force
try {
Move-Item $tmp $Destination -Force
}
catch {
Remove-Item $tmp -Force -ErrorAction SilentlyContinue
throw
}
}
$bindir = Join-Path $Prefix "bin"
$configDir = Join-Path $HOME ".config\dosh"
New-Item -ItemType Directory -Force -Path $bindir, $configDir | Out-Null
@@ -168,9 +182,9 @@ function Install-Prebuilt {
if (-not $found) {
throw "prebuilt archive missing $bin"
}
Copy-Item $found.FullName (Join-Path $bindir $bin) -Force
Install-Binary $found.FullName (Join-Path $bindir $bin)
}
Copy-Item (Join-Path $bindir "dosh-client.exe") (Join-Path $bindir "dosh.exe") -Force
Install-Binary (Join-Path $bindir "dosh-client.exe") (Join-Path $bindir "dosh.exe")
return $true
}
catch {
@@ -202,9 +216,9 @@ function Install-FromSource {
try {
Push-Location $src
cargo build --release --bin dosh-client --bin dosh-bench
Copy-Item "target\release\dosh-client.exe" (Join-Path $bindir "dosh-client.exe") -Force
Copy-Item "target\release\dosh-client.exe" (Join-Path $bindir "dosh.exe") -Force
Copy-Item "target\release\dosh-bench.exe" (Join-Path $bindir "dosh-bench.exe") -Force
Install-Binary "target\release\dosh-client.exe" (Join-Path $bindir "dosh-client.exe")
Install-Binary "target\release\dosh-client.exe" (Join-Path $bindir "dosh.exe")
Install-Binary "target\release\dosh-bench.exe" (Join-Path $bindir "dosh-bench.exe")
}
finally {
Pop-Location
+21 -21
View File
@@ -12,11 +12,7 @@ start_server=1
force_config=0
update_cache="${DOSH_UPDATE_CACHE:-$HOME/.cache/dosh/source}"
quiet="${DOSH_UPDATE_QUIET:-0}"
if [ "${DOSH_UPDATE_QUIET:-0}" = "1" ] && [ "${DOSH_BINARY_REQUIRED:-0}" != "1" ]; then
use_prebuilt=0
else
use_prebuilt="${DOSH_USE_PREBUILT:-1}"
fi
use_prebuilt="${DOSH_USE_PREBUILT:-1}"
binary_url="${DOSH_BINARY_URL:-}"
binary_base="${DOSH_BINARY_BASE:-}"
binary_name="${DOSH_BINARY_NAME:-}"
@@ -230,13 +226,26 @@ find_extracted_binary() {
find "$1" -type f -name "$2" 2>/dev/null | sed -n '1p'
}
install_binary() {
src="$1"
dst="$2"
dst_dir="$(dirname "$dst")"
dst_base="$(basename "$dst")"
tmp="$dst_dir/.$dst_base.tmp.$$"
install -m 0755 "$src" "$tmp"
if ! mv -f "$tmp" "$dst"; then
rm -f "$tmp"
return 1
fi
}
install_extracted_binary() {
found="$(find_extracted_binary "$1" "$2")"
if [ -z "$found" ]; then
echo "prebuilt archive missing $2" >&2
return 1
fi
install -m 0755 "$found" "$3"
install_binary "$found" "$3"
}
sha256_file() {
@@ -333,7 +342,7 @@ try_install_prebuilt() {
install_extracted_binary "$tmpdir/extract" dosh-auth "$bindir/dosh-auth" || return 1
fi
if found_bench="$(find_extracted_binary "$tmpdir/extract" dosh-bench)" && [ -n "$found_bench" ]; then
install -m 0755 "$found_bench" "$bindir/dosh-bench"
install_binary "$found_bench" "$bindir/dosh-bench"
fi
return 0
}
@@ -392,14 +401,14 @@ install_from_source() {
fi
fi
install -m 0755 target/release/dosh-client "$bindir/dosh-client"
install_binary target/release/dosh-client "$bindir/dosh-client"
ln -sf dosh-client "$bindir/dosh"
if [ "$role" = "server" ] || [ "$role" = "both" ]; then
install -m 0755 target/release/dosh-server "$bindir/dosh-server"
install -m 0755 target/release/dosh-auth "$bindir/dosh-auth"
install_binary target/release/dosh-server "$bindir/dosh-server"
install_binary target/release/dosh-auth "$bindir/dosh-auth"
fi
if [ -f target/release/dosh-bench ]; then
install -m 0755 target/release/dosh-bench "$bindir/dosh-bench"
install_binary target/release/dosh-bench "$bindir/dosh-bench"
fi
}
@@ -417,18 +426,9 @@ Wants=network-online.target
[Service]
Type=simple
ExecStart=$bindir/dosh-server serve
Restart=on-failure
Restart=always
RestartSec=1
KillMode=process
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=$config_dir $data_dir
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
RestrictRealtime=true
LockPersonality=true
MemoryDenyWriteExecute=true
[Install]
WantedBy=default.target
+1 -10
View File
@@ -6,18 +6,9 @@ Wants=network-online.target
[Service]
Type=simple
ExecStart=%h/.local/bin/dosh-server serve
Restart=on-failure
Restart=always
RestartSec=1
KillMode=process
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=%h/.config/dosh %h/.local/share/dosh
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
RestrictRealtime=true
LockPersonality=true
MemoryDenyWriteExecute=true
[Install]
WantedBy=default.target
+2
View File
@@ -67,6 +67,7 @@ cleanup() {
kill "$server_pid" 2>/dev/null || true
wait "$server_pid" 2>/dev/null || true
fi
pkill -f "$server_bin hold --runtime-dir $home/sessions/run" 2>/dev/null || true
rm -rf "$home"
}
trap cleanup EXIT INT TERM
@@ -86,6 +87,7 @@ scrollback = 5000
auth_ttl_secs = 30
attach_ticket_ttl_secs = 3600
allow_attach_tickets = true
persist_sessions = false
client_timeout_secs = 60
retransmit_window = 256
default_input_mode = "read-write"
+20
View File
@@ -99,6 +99,21 @@ fi
web_repo="${base%/}/${repo}"
failed=0
artifact_version() {
artifact="$1"
case "$artifact" in
*.tar.gz)
tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*.zip)
unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*)
return 1
;;
esac
}
for artifact in target/dosh-release/dosh-linux-*.tar.gz target/dosh-release/dosh-macos-*.tar.gz target/dosh-release/dosh-windows-*.zip; do
[ -f "$artifact" ] || continue
name="$(basename "$artifact")"
@@ -107,6 +122,11 @@ for artifact in target/dosh-release/dosh-linux-*.tar.gz target/dosh-release/dosh
continue
;;
esac
actual="$(artifact_version "$artifact" || true)"
if [ "$actual" != "$version" ]; then
echo "skipping stale artifact $name: version ${actual:-unknown}, expected $version" >&2
continue
fi
url="$web_repo/releases/download/$tag/$name"
if curl -fsSI "$url" >/dev/null; then
echo "verified $url"
+2
View File
@@ -8,6 +8,8 @@ for test_name in \
live_output_forwards_terminal_control_sequences \
tui_control_sequences_survive_transport_verbatim \
unicode_output_survives_transport \
tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution \
btop_style_graph_output_stays_raw_when_retransmit_history_overflows \
large_tui_paint_is_delivered_in_mtu_safe_frames \
resume_snapshot_preserves_alternate_screen_mode
do
+20 -15
View File
@@ -32,6 +32,21 @@ token="${GITEA_TOKEN:-}"
title="${GITEA_TITLE:-$tag}"
expected_version="${tag#v}"
artifact_version() {
artifact="$1"
case "$artifact" in
*.tar.gz)
tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*.zip)
unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*)
return 1
;;
esac
}
if [ -z "$token" ] && [ -f "$HOME/.config/dosh/gitea-token" ]; then
token="$(tr -d '\r\n' <"$HOME/.config/dosh/gitea-token")"
fi
@@ -52,6 +67,11 @@ if [ "$#" -eq 0 ]; then
continue
;;
esac
actual="$(artifact_version "$artifact" || true)"
if [ "$actual" != "$expected_version" ]; then
echo "skipping $artifact: version ${actual:-unknown}, expected $expected_version" >&2
continue
fi
if [ -f "$artifact.sha256" ]; then
set -- "$@" "$artifact" "$artifact.sha256"
else
@@ -103,21 +123,6 @@ delete_existing_asset() {
done
}
artifact_version() {
artifact="$1"
case "$artifact" in
*.tar.gz)
tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*.zip)
unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*)
return 1
;;
esac
}
verify_release_artifact_version() {
artifact="$1"
name="$(basename "$artifact")"
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env sh
set -eu
repo_root="$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
cd "$repo_root"
version="$(sed -n 's/^version = "\(.*\)"/\1/p' Cargo.toml | sed -n '1p')"
out_dir="${DOSH_PACKAGE_DIR:-target/dosh-release}"
artifact_version() {
artifact="$1"
case "$artifact" in
*.tar.gz)
tar -xOf "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*.zip)
unzip -p "$artifact" dosh/VERSION 2>/dev/null | tr -d '\r\n'
;;
*)
return 1
;;
esac
}
failed=0
for artifact in \
"$out_dir/dosh-linux-x86_64.tar.gz" \
"$out_dir/dosh-macos-aarch64.tar.gz" \
"$out_dir/dosh-windows-x86_64.zip"; do
if [ ! -f "$artifact" ]; then
echo "missing release artifact: $artifact" >&2
failed=1
continue
fi
if [ ! -f "$artifact.sha256" ]; then
echo "missing release checksum: $artifact.sha256" >&2
failed=1
fi
actual="$(artifact_version "$artifact" || true)"
if [ "$actual" != "$version" ]; then
echo "artifact version mismatch: $artifact has ${actual:-unknown}, expected $version" >&2
failed=1
fi
done
exit "$failed"
+567 -37
View File
@@ -36,6 +36,7 @@ use dosh::transport::{
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
TransportEvent,
};
use dosh::udp::is_transient_udp_send_error;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
@@ -63,6 +64,10 @@ const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
const MAX_PENDING_USER_INPUT_BYTES: usize = 1024 * 1024;
const STARTUP_INPUT_HOLD: Duration = Duration::from_millis(750);
const POST_SUBMIT_ALL_INPUT_HOLD: Duration = Duration::from_millis(120);
const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2);
const POST_RECONNECT_STALE_INPUT_GRACE: Duration = Duration::from_secs(2);
const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1);
const ALT_SCREEN_IDLE_REPAINT_AFTER: Duration = Duration::from_secs(15);
/// Sentinel `target_host` the server uses on a server-initiated `StreamOpen` that
/// represents an SSH-agent connection (rather than a TCP target). The client
@@ -3524,6 +3529,14 @@ fn resolve_addr(host: &str, port: u16) -> Result<SocketAddr> {
.ok_or_else(|| anyhow!("no UDP address resolved for {host}:{port}"))
}
async fn send_terminal_udp(socket: &UdpSocket, packet: &[u8], addr: SocketAddr) -> Result<bool> {
match socket.send_to(packet, addr).await {
Ok(_) => Ok(true),
Err(err) if is_transient_udp_send_error(&err) => Ok(false),
Err(err) => Err(err).with_context(|| format!("send UDP packet to {addr}")),
}
}
#[allow(clippy::too_many_arguments)]
async fn try_native_auth(
socket: &UdpSocket,
@@ -4271,6 +4284,7 @@ async fn run_terminal(
let mut status_tick = tokio::time::interval(Duration::from_secs(1));
let mut resize_tick = tokio::time::interval(Duration::from_millis(250));
let mut stream_retransmit_tick = tokio::time::interval(Duration::from_millis(200));
let mut frame_gap_tick = tokio::time::interval(Duration::from_millis(250));
let mut last_size = terminal_size();
// React to terminal resize the instant it happens via SIGWINCH (mosh-style),
// instead of waiting up to one `resize_tick`. The 250ms poll below stays as a
@@ -4321,10 +4335,15 @@ async fn run_terminal(
let mut pending_user_input_bytes = 0usize;
let mut startup_input_hold_until: Option<Instant> = None;
let mut startup_gate_mode = StartupGateMode::HoldControl;
let mut stale_terminal_input_suppress_until: Option<Instant> = None;
let mut last_terminal_frame_at = Instant::now();
let mut last_focus_repaint_at = Instant::now() - FOCUS_REPAINT_COOLDOWN;
let mut last_idle_repaint_attempt_at = Instant::now() - ALT_SCREEN_IDLE_REPAINT_AFTER;
if let Some(frame) = first_frame {
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
if frame.closed {
return Ok(());
@@ -4336,7 +4355,13 @@ async fn run_terminal(
&& cred.mode != "view-only"
&& !forward_only
{
send_input(&socket, addr, &cred, &mut send_seq, bytes).await?;
if !send_input(&socket, addr, &cred, &mut send_seq, bytes.clone()).await? {
queue_stale_pending_user_input(
&mut pending_user_input,
&mut pending_user_input_bytes,
bytes,
)?;
}
startup_input_hold_until = Some(Instant::now() + STARTUP_INPUT_HOLD);
startup_gate_mode = StartupGateMode::HoldAll;
}
@@ -4372,10 +4397,59 @@ async fn run_terminal(
biased;
stdin_msg = stdin_rx.recv() => {
match stdin_msg {
Some(bytes) => {
Some(mut bytes) => {
if input_matches_escape(&bytes, escape_key.as_deref()) {
break;
}
let saw_focus_in = input_contains_focus_in(&bytes);
if !forward_only
&& saw_focus_in
&& last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN
{
last_focus_repaint_at = Instant::now();
if let Some(frame) = reconnect(
&socket,
&mut cred,
&mut send_seq,
last_size,
&mut frame_buffer,
&mut predictor,
)
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_packet_at = Instant::now();
flush_pending_user_input(
&socket,
addr,
&cred,
&mut send_seq,
&mut predictor,
&mut pending_user_input,
&mut pending_user_input_bytes,
)
.await?;
}
}
bytes = strip_terminal_focus_reports(&bytes);
if bytes.is_empty() {
continue;
}
if should_strip_stale_terminal_reports(
last_packet_at.elapsed(),
stale_terminal_input_suppress_until,
) {
bytes = strip_stale_mouse_reports(&bytes);
if bytes.is_empty() {
continue;
}
}
if cred.mode != "view-only" && cred.mode != "forward-only" {
if let Some(deadline) = startup_input_hold_until {
if !predictor.alternate_screen
@@ -4410,8 +4484,15 @@ async fn run_terminal(
&& let Some((send_now, mut hold_for_later)) =
split_after_command_submit(&bytes)
{
predictor.observe_input(&send_now)?;
send_input(&socket, addr, &cred, &mut send_seq, send_now).await?;
if send_input(&socket, addr, &cred, &mut send_seq, send_now.clone()).await? {
predictor.observe_input(&send_now)?;
} else {
queue_stale_pending_user_input(
&mut pending_user_input,
&mut pending_user_input_bytes,
send_now,
)?;
}
startup_input_hold_until = Some(
Instant::now()
+ post_submit_hold_duration(&hold_for_later),
@@ -4430,15 +4511,23 @@ async fn run_terminal(
)?;
continue;
} else if !hold_for_later.is_empty() {
predictor.observe_input(&hold_for_later)?;
send_input(
let sent = send_input(
&socket,
addr,
&cred,
&mut send_seq,
std::mem::take(&mut hold_for_later),
hold_for_later.clone(),
)
.await?;
if sent {
predictor.observe_input(&hold_for_later)?;
} else {
queue_stale_pending_user_input(
&mut pending_user_input,
&mut pending_user_input_bytes,
std::mem::take(&mut hold_for_later),
)?;
}
}
continue;
}
@@ -4459,9 +4548,13 @@ async fn run_terminal(
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
last_packet_at = Instant::now();
flush_pending_user_input(
@@ -4476,8 +4569,15 @@ async fn run_terminal(
.await?;
}
} else {
predictor.observe_input(&bytes)?;
send_input(&socket, addr, &cred, &mut send_seq, bytes).await?;
if send_input(&socket, addr, &cred, &mut send_seq, bytes.clone()).await? {
predictor.observe_input(&bytes)?;
} else {
queue_stale_pending_user_input(
&mut pending_user_input,
&mut pending_user_input_bytes,
bytes,
)?;
}
}
}
}
@@ -4498,6 +4598,40 @@ async fn run_terminal(
_ = resize_tick.tick() => {
maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?;
}
_ = frame_gap_tick.tick() => {
if frame_buffer.resync_due()
&& let Some(frame) = reconnect(
&socket,
&mut cred,
&mut send_seq,
last_size,
&mut frame_buffer,
&mut predictor,
)
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
last_packet_at = Instant::now();
flush_pending_user_input(
&socket,
addr,
&cred,
&mut send_seq,
&mut predictor,
&mut pending_user_input,
&mut pending_user_input_bytes,
)
.await?;
}
}
recv = socket.recv_from(&mut recv_buf) => {
let (n, _) = recv?;
let Ok(packet) = protocol::decode(&recv_buf[..n]) else {
@@ -4531,9 +4665,13 @@ async fn run_terminal(
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
last_packet_at = Instant::now();
flush_pending_user_input(
@@ -4559,6 +4697,7 @@ async fn run_terminal(
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
flush_startup_input_if_ready(
&socket,
addr,
@@ -4579,6 +4718,22 @@ async fn run_terminal(
}
PacketKind::Pong => {
last_packet_at = Instant::now();
if should_flush_terminal_input_after_contact(
forward_only,
&cred.mode,
pending_user_input.is_empty(),
) {
flush_pending_user_input(
&socket,
addr,
&cred,
&mut send_seq,
&mut predictor,
&mut pending_user_input,
&mut pending_user_input_bytes,
)
.await?;
}
}
PacketKind::Rekey => {
// Server-initiated transport rekey (spec §11). The Rekey is
@@ -4620,7 +4775,7 @@ async fn run_terminal(
CLIENT_TO_SERVER,
b"",
)?;
socket.send_to(&ack, addr).await?;
let _ = send_terminal_udp(&socket, &ack, addr).await?;
}
PacketKind::AttachReject => {
let reject: AttachReject = protocol::from_body(&packet.body)?;
@@ -4636,9 +4791,13 @@ async fn run_terminal(
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
last_packet_at = Instant::now();
flush_pending_user_input(
@@ -5028,9 +5187,13 @@ async fn run_terminal(
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
if !forward_only {
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
}
last_packet_at = Instant::now();
flush_pending_user_input(
@@ -5055,7 +5218,7 @@ async fn run_terminal(
b"",
)?;
send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(&socket, &packet, addr).await?;
}
// Re-evaluate the prediction display policy on every timer tick so
// a latency spike (or recovery) flips speculation on/off promptly
@@ -5072,6 +5235,44 @@ async fn run_terminal(
(&mut startup_input_hold_until, &mut startup_gate_mode),
)
.await?;
let now = Instant::now();
if should_repaint_idle_alternate_screen(
predictor.alternate_screen,
last_terminal_frame_at,
last_idle_repaint_attempt_at,
now,
) {
last_idle_repaint_attempt_at = now;
if let Some(frame) = reconnect(
&socket,
&mut cred,
&mut send_seq,
last_size,
&mut frame_buffer,
&mut predictor,
)
.await?
{
refresh_live_addr(&mut addr, &cred)?;
arm_stale_terminal_input_suppression(
&mut stale_terminal_input_suppress_until,
);
render_frame(&frame)?;
predictor.observe_output(&frame.bytes);
last_terminal_frame_at = Instant::now();
last_packet_at = Instant::now();
flush_pending_user_input(
&socket,
addr,
&cred,
&mut send_seq,
&mut predictor,
&mut pending_user_input,
&mut pending_user_input_bytes,
)
.await?;
}
}
}
// Refresh / clear the non-destructive disconnect status line based
// on how long the link has been silent (recomputed after any
@@ -5369,6 +5570,23 @@ fn queue_pending_user_input_with_filter(
Ok(())
}
fn requeue_pending_user_input_front(
pending: &mut VecDeque<PendingUserInput>,
pending_bytes: &mut usize,
input: PendingUserInput,
) -> Result<()> {
let next = pending_bytes
.checked_add(input.bytes.len())
.ok_or_else(|| anyhow!("pending input buffer overflow"))?;
anyhow::ensure!(
next <= MAX_PENDING_USER_INPUT_BYTES,
"pending input buffer full while disconnected"
);
*pending_bytes = next;
pending.push_front(input);
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct PendingUserInput {
bytes: Vec<u8>,
@@ -5408,6 +5626,61 @@ fn should_hold_during_startup_gate(
mode == StartupGateMode::HoldAll || already_queued || should_hold_post_submit_input(bytes)
}
fn should_flush_terminal_input_after_contact(
forward_only: bool,
mode: &str,
pending_empty: bool,
) -> bool {
!forward_only && !pending_empty && mode != "view-only" && mode != "forward-only"
}
fn should_strip_stale_terminal_reports(
silent_for: Duration,
suppress_until: Option<Instant>,
) -> bool {
silent_for >= STALE_TERMINAL_INPUT_AFTER
|| suppress_until.is_some_and(|deadline| Instant::now() < deadline)
}
fn input_contains_focus_in(bytes: &[u8]) -> bool {
contains_bytes(bytes, b"\x1b[I") || contains_bytes(bytes, b"\x9bI")
}
fn strip_terminal_focus_reports(bytes: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(bytes.len());
let mut offset = 0;
while offset < bytes.len() {
match bytes.get(offset..) {
Some([0x1b, b'[', b'I' | b'O', ..]) => {
offset += 3;
}
Some([0x9b, b'I' | b'O', ..]) => {
offset += 2;
}
_ => {
out.push(bytes[offset]);
offset += 1;
}
}
}
out
}
fn should_repaint_idle_alternate_screen(
alternate_screen: bool,
last_terminal_frame_at: Instant,
last_attempt_at: Instant,
now: Instant,
) -> bool {
alternate_screen
&& now.duration_since(last_terminal_frame_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
&& now.duration_since(last_attempt_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
}
fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) {
*suppress_until = Some(Instant::now() + POST_RECONNECT_STALE_INPUT_GRACE);
}
async fn flush_pending_user_input(
socket: &UdpSocket,
addr: SocketAddr,
@@ -5431,8 +5704,19 @@ async fn flush_pending_user_input(
if bytes.is_empty() {
continue;
}
predictor.observe_input(&bytes)?;
send_input(socket, addr, cred, send_seq, bytes).await?;
if send_input(socket, addr, cred, send_seq, bytes.clone()).await? {
predictor.observe_input(&bytes)?;
} else {
requeue_pending_user_input_front(
pending,
pending_bytes,
PendingUserInput {
bytes,
strip_mouse_reports: false,
},
)?;
break;
}
}
Ok(())
}
@@ -5458,6 +5742,7 @@ fn stale_mouse_report_len(bytes: &[u8], offset: usize) -> Option<usize> {
match bytes.get(offset).copied()? {
0x1b if bytes.get(offset + 1) == Some(&b'[') => match bytes.get(offset + 2).copied() {
Some(b'M') if offset + 6 <= bytes.len() => Some(6),
Some(b'I' | b'O') => Some(3),
Some(b'<') => mouse_report_params_len(bytes, offset + 3, 2).map(|len| len + 3),
Some(byte) if byte.is_ascii_digit() => {
mouse_report_params_len(bytes, offset + 2, 2).map(|len| len + 2)
@@ -5466,13 +5751,15 @@ fn stale_mouse_report_len(bytes: &[u8], offset: usize) -> Option<usize> {
},
0x9b => match bytes.get(offset + 1).copied() {
Some(b'M') if offset + 5 <= bytes.len() => Some(5),
Some(b'I' | b'O') => Some(2),
Some(b'<') => mouse_report_params_len(bytes, offset + 2, 2).map(|len| len + 2),
Some(byte) if byte.is_ascii_digit() => {
mouse_report_params_len(bytes, offset + 1, 2).map(|len| len + 1)
}
_ => None,
},
byte if byte.is_ascii_digit() => mouse_report_params_len(bytes, offset, 1),
byte if byte.is_ascii_digit() => mouse_report_params_len(bytes, offset, 1)
.or_else(|| orphan_mouse_suffix_len(bytes, offset)),
_ => None,
}
}
@@ -5494,6 +5781,21 @@ fn mouse_report_params_len(bytes: &[u8], offset: usize, min_semicolons: usize) -
None
}
fn orphan_mouse_suffix_len(bytes: &[u8], offset: usize) -> Option<usize> {
if offset != 0 {
return None;
}
let mut cursor = offset;
while let Some(byte) = bytes.get(cursor).copied() {
match byte {
b'0'..=b'9' => cursor += 1,
b'M' | b'm' if cursor > offset => return Some(cursor + 1 - offset),
_ => return None,
}
}
None
}
fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
let Some(rest) = bytes.get(offset..) else {
return false;
@@ -5502,19 +5804,21 @@ fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
[0x1b] | [0x1b, b'['] | [0x1b, b'[', b'<'] => true,
[0x1b, b'[', b'M', ..] if rest.len() < 6 => true,
[0x1b, b'[', b'<', params @ ..] | [0x1b, b'[', params @ ..]
if params_are_mouse_prefix(params) =>
if params_are_mouse_prefix(params, 2) =>
{
true
}
[0x9b] | [0x9b, b'<'] => true,
[0x9b, b'M', ..] if rest.len() < 5 => true,
[0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params) => true,
params if params_are_mouse_prefix(params) => true,
[0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params, 2) => {
true
}
params if params_are_mouse_prefix(params, 2) => true,
_ => false,
}
}
fn params_are_mouse_prefix(params: &[u8]) -> bool {
fn params_are_mouse_prefix(params: &[u8], min_semicolons: usize) -> bool {
let mut semicolons = 0usize;
let mut saw_digit = false;
for byte in params {
@@ -5524,7 +5828,7 @@ fn params_are_mouse_prefix(params: &[u8]) -> bool {
_ => return false,
}
}
saw_digit && semicolons >= 1
saw_digit && semicolons >= min_semicolons
}
async fn flush_startup_input_if_ready(
@@ -5564,7 +5868,7 @@ async fn send_input(
cred: &CachedCredential,
send_seq: &mut u64,
bytes: Vec<u8>,
) -> Result<()> {
) -> Result<bool> {
let body = protocol::to_body(&Input { bytes })?;
let packet = protocol::encode_encrypted(
PacketKind::Input,
@@ -5576,8 +5880,7 @@ async fn send_input(
&body,
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
Ok(())
send_terminal_udp(socket, &packet, addr).await
}
async fn send_stream_open(
@@ -5920,7 +6223,7 @@ async fn send_stream_packet(
body,
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -6561,14 +6864,22 @@ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
.any(|window| window == needle)
}
const FRAME_GAP_RESYNC_AFTER_MS: u64 = 2_500;
const FRAME_GAP_RESYNC_COOLDOWN_MS: u64 = 5_000;
const FRAME_GAP_PENDING_RESYNC_THRESHOLD: usize = 128;
#[derive(Default)]
struct FrameBuffer {
pending: BTreeMap<u64, Frame>,
gap_since: Option<Instant>,
last_resync_at: Option<Instant>,
}
impl FrameBuffer {
fn clear(&mut self) {
self.pending.clear();
self.gap_since = None;
self.last_resync_at = None;
}
fn accept(&mut self, frame: Frame, last_rendered_seq: &mut u64) -> Vec<Frame> {
@@ -6577,6 +6888,8 @@ impl FrameBuffer {
return Vec::new();
}
self.pending.clear();
self.gap_since = None;
self.last_resync_at = None;
*last_rendered_seq = frame.output_seq;
return vec![frame];
}
@@ -6590,8 +6903,47 @@ impl FrameBuffer {
*last_rendered_seq = frame.output_seq;
ready.push(frame);
}
self.refresh_gap(*last_rendered_seq);
ready
}
fn refresh_gap(&mut self, last_rendered_seq: u64) {
let waiting_on_missing = self
.pending
.first_key_value()
.is_some_and(|(&seq, _)| seq > last_rendered_seq.saturating_add(1));
if waiting_on_missing {
self.gap_since.get_or_insert_with(Instant::now);
} else {
self.gap_since = None;
self.last_resync_at = None;
}
}
fn resync_due(&mut self) -> bool {
let Some(gap_since) = self.gap_since else {
return false;
};
let now = Instant::now();
let waited = now.duration_since(gap_since);
if waited < Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS)
&& self.pending.len() < FRAME_GAP_PENDING_RESYNC_THRESHOLD
{
return false;
}
if self.last_resync_at.is_some_and(|last| {
now.duration_since(last) < Duration::from_millis(FRAME_GAP_RESYNC_COOLDOWN_MS)
}) {
return false;
}
self.last_resync_at = Some(now);
true
}
#[cfg(test)]
fn backdate_gap_for_test(&mut self, elapsed: Duration) {
self.gap_since = Some(Instant::now() - elapsed);
}
}
/// Read the current terminal size and, if it changed since `last_size`, send a
@@ -6635,7 +6987,7 @@ async fn send_resize(
&body,
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -6655,7 +7007,7 @@ async fn send_ack(
b"",
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -6670,7 +7022,7 @@ async fn detach_once(socket: &UdpSocket, cred: &CachedCredential, seq: u64) -> R
CLIENT_TO_SERVER,
b"",
)?;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -7003,11 +7355,12 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
#[cfg(test)]
mod tests {
use super::{
CachedCredential, DisconnectStatus, DynamicForward, FrameBuffer, LocalForward,
MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode,
Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig,
StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix,
clear_cached_credentials, ensure_tui_safe_status_overlay, input_matches_escape,
ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward,
FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode, Predictor,
RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig, StartupGateMode,
StatusAction, auth_allows, cache_key, cache_server_prefix, clear_cached_credentials,
ensure_tui_safe_status_overlay, input_contains_focus_in, input_matches_escape,
is_local_status_target, is_resume_response_for_client, latest_release_download_url,
load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key,
parse_local_forward, parse_remote_forward, parse_ssh_config, post_submit_hold_duration,
@@ -7015,19 +7368,21 @@ mod tests {
release_tag_download_url, release_tag_from_effective_url, release_version_from_tag,
render_status_clear, render_status_overlay, requested_env, resolved_startup_command,
retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
server_version_mismatch, should_hold_during_startup_gate, should_hold_post_submit_input,
split_after_command_submit, ssh_destination_host, ssh_username, ssh_with_user,
startup_command, status_ssh_target, strip_stale_mouse_reports, toml_bare_key_or_quoted,
update_check_requested, update_version_status, upsert_managed_block, valid_forward_host,
vscode_safe_alias,
server_version_mismatch, should_flush_terminal_input_after_contact,
should_hold_during_startup_gate, should_hold_post_submit_input,
should_repaint_idle_alternate_screen, split_after_command_submit, ssh_destination_host,
ssh_username, ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports,
strip_terminal_focus_reports, toml_bare_key_or_quoted, update_check_requested,
update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias,
};
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
use dosh::native::EnvVar;
use dosh::protocol::{self, Frame, PacketKind};
use dosh::udp::is_transient_udp_send_error;
use ed25519_dalek::{SigningKey, VerifyingKey};
use std::collections::{HashMap, VecDeque};
use std::fs;
use std::time::Duration;
use std::time::{Duration, Instant};
fn test_args(server: &str, command: &[&str]) -> super::Args {
super::Args {
@@ -7457,6 +7812,54 @@ mod tests {
assert!(buffer.accept(test_frame(12, false), &mut last).is_empty());
}
#[test]
fn frame_buffer_keeps_gap_buffered_without_snapshot_resync() {
let mut buffer = FrameBuffer::default();
let mut last = 10;
assert!(buffer.accept(test_frame(13, false), &mut last).is_empty());
assert_eq!(last, 10);
assert!(buffer.accept(test_frame(14, false), &mut last).is_empty());
assert_eq!(last, 10);
assert!(!buffer.resync_due());
let ready = buffer.accept(test_frame(11, false), &mut last);
assert_eq!(ready.len(), 1);
assert_eq!(ready[0].output_seq, 11);
assert_eq!(last, 11);
let ready = buffer.accept(test_frame(12, false), &mut last);
assert_eq!(ready.len(), 3);
assert_eq!(
ready
.iter()
.map(|frame| frame.output_seq)
.collect::<Vec<_>>(),
vec![12, 13, 14]
);
assert_eq!(last, 14);
}
#[test]
fn frame_buffer_resyncs_old_gap_with_cooldown() {
let mut buffer = FrameBuffer::default();
let mut last = 10;
assert!(buffer.accept(test_frame(13, false), &mut last).is_empty());
buffer.backdate_gap_for_test(Duration::from_millis(FRAME_GAP_RESYNC_AFTER_MS + 10));
assert!(buffer.resync_due());
assert!(
!buffer.resync_due(),
"resync must be rate-limited so snapshots cannot churn the TUI"
);
let ready = buffer.accept(test_frame(20, true), &mut last);
assert_eq!(ready.len(), 1);
assert_eq!(ready[0].output_seq, 20);
assert_eq!(last, 20);
assert!(!buffer.resync_due());
}
#[test]
fn frame_buffer_ignores_stale_snapshot() {
let mut buffer = FrameBuffer::default();
@@ -7751,6 +8154,34 @@ mod tests {
assert_eq!(post_submit_hold_duration(b"x"), POST_SUBMIT_ALL_INPUT_HOLD);
}
#[test]
fn contact_flushes_pending_terminal_input_for_interactive_sessions() {
assert!(should_flush_terminal_input_after_contact(
false, "normal", false
));
assert!(should_flush_terminal_input_after_contact(
false,
"read-write",
false
));
assert!(!should_flush_terminal_input_after_contact(
false, "normal", true
));
assert!(!should_flush_terminal_input_after_contact(
true, "normal", false
));
assert!(!should_flush_terminal_input_after_contact(
false,
"view-only",
false
));
assert!(!should_flush_terminal_input_after_contact(
false,
"forward-only",
false
));
}
#[test]
fn reconnect_refreshes_live_send_address_from_credentials() {
let mut addr = "127.0.0.1:50000".parse().unwrap();
@@ -8173,6 +8604,41 @@ mod tests {
assert_eq!(pending_bytes, MAX_PENDING_USER_INPUT_BYTES);
}
#[test]
fn transient_udp_send_errors_are_not_terminal_fatal() {
#[cfg(unix)]
{
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(libc::ENETUNREACH)
));
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(libc::EHOSTUNREACH)
));
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(libc::EADDRNOTAVAIL)
));
assert!(!is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(libc::EINVAL)
));
}
#[cfg(windows)]
{
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(10051)
)); // WSAENETUNREACH
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(10065)
)); // WSAEHOSTUNREACH
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(10049)
)); // WSAEADDRNOTAVAIL
assert!(!is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(10022)
)); // WSAEINVAL
}
}
#[test]
fn stale_mouse_reports_are_stripped_from_pending_input() {
let input = b"\x1b[<35;152;1Mhello\x1b[<0;107;10m";
@@ -8186,6 +8652,70 @@ mod tests {
assert_eq!(strip_stale_mouse_reports(b"152;1Mls\r"), b"ls\r");
}
#[test]
fn split_stale_mouse_suffixes_are_stripped_from_pending_input() {
assert_eq!(strip_stale_mouse_reports(b"1M35;149;1Mls\r"), b"ls\r");
assert_eq!(strip_stale_mouse_reports(b"10m"), b"");
}
#[test]
fn numeric_semicolon_input_survives_stale_filter() {
assert_eq!(strip_stale_mouse_reports(b"123;"), b"123;");
assert_eq!(strip_stale_mouse_reports(b"12;34"), b"12;34");
assert_eq!(strip_stale_mouse_reports(b"echo 1;"), b"echo 1;");
}
#[test]
fn incomplete_three_field_mouse_prefix_is_suppressed() {
assert_eq!(strip_stale_mouse_reports(b"35;152;"), b"");
assert_eq!(strip_stale_mouse_reports(b"\x1b[<35;152;"), b"");
assert_eq!(strip_stale_mouse_reports(b"\x9b35;152;"), b"");
}
#[test]
fn stale_focus_reports_are_stripped_from_pending_input() {
assert_eq!(strip_stale_mouse_reports(b"\x1b[Ihello\x1b[O"), b"hello");
assert_eq!(strip_stale_mouse_reports(b"\x9bIhello\x9bO"), b"hello");
}
#[test]
fn focus_in_reports_trigger_local_repaint_detection() {
assert!(input_contains_focus_in(b"\x1b[I"));
assert!(input_contains_focus_in(b"before\x9bIafter"));
assert!(!input_contains_focus_in(b"\x1b[O"));
assert!(!input_contains_focus_in(b"\x1b[A"));
}
#[test]
fn focus_reports_are_never_forwarded_as_user_input() {
assert_eq!(strip_terminal_focus_reports(b"\x1b[Ihello\x1b[O"), b"hello");
assert_eq!(strip_terminal_focus_reports(b"\x9bIhello\x9bO"), b"hello");
assert_eq!(strip_terminal_focus_reports(b"\x1b[A"), b"\x1b[A");
assert_eq!(
strip_terminal_focus_reports(b"before\x1b[Iafter"),
b"beforeafter"
);
}
#[test]
fn idle_repaint_only_runs_for_stale_alternate_screen() {
let now = Instant::now();
let stale = now - ALT_SCREEN_IDLE_REPAINT_AFTER - Duration::from_secs(1);
let recent = now - Duration::from_secs(1);
assert!(should_repaint_idle_alternate_screen(
true, stale, stale, now
));
assert!(!should_repaint_idle_alternate_screen(
false, stale, stale, now
));
assert!(!should_repaint_idle_alternate_screen(
true, recent, stale, now
));
assert!(!should_repaint_idle_alternate_screen(
true, stale, recent, now
));
}
#[test]
fn non_mouse_escape_input_survives_stale_filter() {
assert_eq!(strip_stale_mouse_reports(b"\x1b[A"), b"\x1b[A");
+191 -178
View File
@@ -28,6 +28,7 @@ use dosh::protocol::{
TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
};
use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
use dosh::udp::is_transient_udp_send_error;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::fs;
@@ -232,19 +233,6 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
}
});
let pacing_state = Arc::clone(&state);
let pacing_socket = Arc::clone(&socket);
let pacing_interval_ms = config.output_frame_interval_ms.max(1);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_millis(pacing_interval_ms));
loop {
interval.tick().await;
if let Err(err) = flush_paced_snapshots(&pacing_state, &pacing_socket).await {
eprintln!("paced snapshot error: {err:#}");
}
}
});
let cleanup_state = Arc::clone(&state);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(5));
@@ -400,8 +388,6 @@ struct ClientState {
rows: u16,
last_seen: Instant,
pending: VecDeque<PendingFrame>,
last_frame_sent: Instant,
paced_snapshot_due: bool,
allowed_forwardings: Vec<ForwardingRequest>,
stream_writers: HashMap<u64, mpsc::Sender<Vec<u8>>>,
opened_streams: HashSet<u64>,
@@ -447,6 +433,7 @@ struct PendingStreamOpen {
attempts: u32,
}
#[derive(Clone)]
struct PendingNativeAuth {
client: dosh::native::NativeClientHello,
server: NativeServerHello,
@@ -527,12 +514,13 @@ impl ServerState {
/// Whether a session named `name` should be backed by a persistent holder.
///
/// With persistence enabled, every PTY session gets a detached holder. That
/// includes implicit `term-<millis>-<pid>` sessions: a reconnecting client has
/// the exact session name in its ticket cache, so it can reattach after a
/// quick server restart without forcing users to name sessions manually.
fn should_persist_session(&self, _name: &str) -> bool {
self.config.persist_sessions
/// With persistence enabled, explicitly named sessions get a detached
/// holder. Implicit `term-<millis>-<pid>` sessions are one-off terminals
/// created by `dosh host`; users cannot intentionally reattach to them by
/// name, so persisting them only leaves stale holders that can be
/// accidentally re-adopted after restarts.
fn should_persist_session(&self, name: &str) -> bool {
self.config.persist_sessions && !protocol::is_implicit_session_name(name)
}
/// Spawn (or, in the persistent case, spawn-and-adopt) a PTY for a session.
@@ -620,6 +608,20 @@ impl ServerState {
if self.sessions.contains_key(&name) {
continue;
}
if !self.should_persist_session(&name) {
eprintln!(
"discarding stale implicit persistent session {name} (shell pid {})",
meta.shell_pid
);
persist::request_shutdown(&sessions_dir, &name, None);
if meta.shell_pid > 0 {
let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGHUP) };
let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGTERM) };
std::thread::sleep(Duration::from_millis(50));
let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGKILL) };
}
continue;
}
let (pty, control) = match self.adopt_persistent_pty(&name) {
Ok(pair) => pair,
Err(err) => {
@@ -863,6 +865,7 @@ async fn handle_packet(
| PacketKind::StreamEof
| PacketKind::StreamWindowAdjust
| PacketKind::RekeyAck
| PacketKind::Detach
if find_client_decrypt_key(state, &packet.header).is_err() =>
{
send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client").await
@@ -980,7 +983,7 @@ async fn handle_native_client_hello(
};
let body = protocol::to_body(&NativeServerHelloBody { hello })?;
let out = protocol::encode_plain(PacketKind::NativeServerHello, pending_id, 1, 0, &body)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -991,8 +994,8 @@ async fn handle_native_user_auth(
packet: &protocol::Packet,
) -> Result<()> {
let pending = {
let mut locked = state.lock().expect("server state poisoned");
locked.pending_native.remove(&packet.header.conn_id)
let locked = state.lock().expect("server state poisoned");
locked.pending_native.get(&packet.header.conn_id).cloned()
};
let Some(pending) = pending else {
return send_reject_to_client(socket, peer, packet.header.conn_id, "unknown native auth")
@@ -1008,11 +1011,38 @@ async fn handle_native_user_auth(
.await;
}
if pending.created_at.elapsed() > Duration::from_secs(30) {
state
.lock()
.expect("server state poisoned")
.pending_native
.remove(&packet.header.conn_id);
return send_reject_to_client(socket, peer, packet.header.conn_id, "native auth expired")
.await;
}
let body = protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER)?;
let req: NativeUserAuthBody = protocol::from_body(&body)?;
let body = match protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER) {
Ok(body) => body,
Err(_) => {
return send_reject_to_client(
socket,
peer,
packet.header.conn_id,
"native auth decrypt failed",
)
.await;
}
};
let req: NativeUserAuthBody = match protocol::from_body(&body) {
Ok(req) => req,
Err(_) => {
return send_reject_to_client(
socket,
peer,
packet.header.conn_id,
"invalid native auth body",
)
.await;
}
};
if pending.client.requested_mode == "doctor" {
let check_result = {
let locked = state.lock().expect("server state poisoned");
@@ -1048,6 +1078,11 @@ async fn handle_native_user_auth(
.await;
}
};
state
.lock()
.expect("server state poisoned")
.pending_native
.remove(&packet.header.conn_id);
let body = protocol::to_body(&check)?;
let out = protocol::encode_encrypted(
PacketKind::NativeAuthCheckOk,
@@ -1058,7 +1093,7 @@ async fn handle_native_user_auth(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
return Ok(());
}
@@ -1167,8 +1202,6 @@ async fn handle_native_user_auth(
rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: req.auth.requested_forwardings.clone(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -1230,6 +1263,11 @@ async fn handle_native_user_auth(
if let Some((listener, path)) = agent_listener {
spawn_agent_forward(state, socket, client_id, listener, path);
}
state
.lock()
.expect("server state poisoned")
.pending_native
.remove(&packet.header.conn_id);
let ok = NativeAuthOk {
client_id,
@@ -1253,7 +1291,7 @@ async fn handle_native_user_auth(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1311,8 +1349,6 @@ async fn handle_bootstrap_attach(
rows: req.rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -1363,7 +1399,7 @@ async fn handle_bootstrap_attach(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1447,8 +1483,6 @@ async fn handle_ticket_attach(
rows: req.rows,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -1500,7 +1534,7 @@ async fn handle_ticket_attach(
};
let body = protocol::to_body(&envelope)?;
let out = protocol::encode_plain(PacketKind::AttachOk, client_id, 1, 0, &body)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1518,7 +1552,7 @@ async fn send_reject_to_client(
reason: reason.to_string(),
})?;
let out = protocol::encode_plain(PacketKind::AttachReject, client_id, 0, 0, &body)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1584,7 +1618,7 @@ async fn handle_resume(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1696,12 +1730,19 @@ async fn handle_ping(
SERVER_TO_CLIENT,
b"",
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packet) -> Result<()> {
let (key, _) = find_client_decrypt_key(state, &packet.header)?;
let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
let mut locked = state.lock().expect("server state poisoned");
if let Some(client) = locked.client_mut(&packet.header.conn_id) {
if !client.replay.accept(packet.header.seq) {
return Ok(());
}
}
locked.remove_client_everywhere(&packet.header.conn_id);
Ok(())
}
@@ -1711,6 +1752,14 @@ fn remove_client(state: &Arc<Mutex<ServerState>>, client_id: [u8; 16]) {
locked.remove_client_everywhere(&client_id);
}
async fn send_udp(socket: &UdpSocket, packet: &[u8], peer: SocketAddr) -> Result<bool> {
match socket.send_to(packet, peer).await {
Ok(_) => Ok(true),
Err(err) if is_transient_udp_send_error(&err) => Ok(false),
Err(err) => Err(err).with_context(|| format!("send UDP packet to {peer}")),
}
}
async fn handle_ack(
state: &Arc<Mutex<ServerState>>,
peer: SocketAddr,
@@ -3026,7 +3075,7 @@ async fn send_stream_open_to_client(
found
};
if let Some((endpoint, packet)) = send {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3120,7 +3169,7 @@ async fn queue_or_send_stream_data_to_client(
send
};
if let Some((endpoint, packet)) = send {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3195,7 +3244,7 @@ async fn flush_stream_pending_data_to_client(
let Some((endpoint, packet)) = send else {
return Ok(());
};
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
}
@@ -3339,7 +3388,7 @@ async fn send_stream_packet_to_client(
found
};
if let Some((endpoint, packet)) = send {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3357,9 +3406,7 @@ async fn broadcast_output(
let sends = {
let mut locked = state.lock().expect("server state poisoned");
let scrollback = locked.config.scrollback;
let retransmit_window = locked.config.retransmit_window;
let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
let now = Instant::now();
let retransmit_window = locked.config.retransmit_window.max(1);
let session = locked
.sessions
.get_mut(&output.session)
@@ -3394,32 +3441,21 @@ async fn broadcast_output(
}
let mut sends = Vec::new();
for (client_id, client) in session.clients.iter_mut() {
let terminal_control =
output.bytes.contains(&0x1b) || session.parser.screen().alternate_screen();
if !terminal_control
&& !frame_interval.is_zero()
&& now.duration_since(client.last_frame_sent) < frame_interval
{
client.paced_snapshot_due = true;
continue;
}
client.send_seq += 1;
// Count traffic toward the packet-count rekey trigger (spec §11).
client.epoch_packets = client.epoch_packets.saturating_add(1);
// Only materialize a screen snapshot when this client has fallen too
// far behind; the common case sends the raw output bytes and must not
// clone the whole vt100 grid per client per packet.
let (bytes, snapshot) = if client.pending.len() >= retransmit_window {
client.pending.clear();
(screen_snapshot(session.parser.screen()), true)
} else {
(output.bytes.clone(), false)
};
// Live terminal bytes are never rewritten into vt100 snapshots. A
// snapshot is useful for attach/resume, but substituting it during a
// running TUI can lose graph/background-cell details that apps like
// btop rely on. If retransmit history is full, prune only history.
while client.pending.len() >= retransmit_window {
client.pending.pop_front();
}
let frame = Frame {
session: output.session.clone(),
output_seq,
bytes,
snapshot,
bytes: output.bytes.clone(),
snapshot: false,
closed: false,
};
let body = protocol::to_body(&frame)?;
@@ -3432,17 +3468,12 @@ async fn broadcast_output(
SERVER_TO_CLIENT,
&body,
)?;
while client.pending.len() >= retransmit_window {
client.pending.pop_front();
}
client.pending.push_back(PendingFrame {
output_seq,
packet: packet.clone(),
last_sent: Instant::now(),
attempts: 0,
});
client.last_frame_sent = now;
client.paced_snapshot_due = false;
sends.push((client.endpoint, packet));
}
sends
@@ -3459,7 +3490,7 @@ async fn broadcast_output(
}
}
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3512,7 +3543,7 @@ async fn broadcast_session_exit(
persist::remove_runtime_dir(&sessions_dir, &session_name);
}
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3543,11 +3574,9 @@ async fn retransmit_pending(
if pending.output_seq <= client.last_acked {
continue;
}
if now.duration_since(pending.last_sent) >= Duration::from_millis(200)
&& pending.attempts < 8
{
if now.duration_since(pending.last_sent) >= Duration::from_millis(200) {
pending.last_sent = now;
pending.attempts += 1;
pending.attempts = pending.attempts.saturating_add(1);
sends.push((client.endpoint, pending.packet.clone()));
}
}
@@ -3621,77 +3650,7 @@ async fn retransmit_pending(
sends
};
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
}
Ok(())
}
async fn flush_paced_snapshots(
state: &Arc<Mutex<ServerState>>,
socket: &Arc<UdpSocket>,
) -> Result<()> {
let sends = {
let mut locked = state.lock().expect("server state poisoned");
let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms);
if frame_interval.is_zero() {
return Ok(());
}
let retransmit_window = locked.config.retransmit_window;
let now = Instant::now();
let mut sends = Vec::new();
for (session_name, session) in locked.sessions.iter_mut() {
let due = session.clients.values().any(|client| {
client.paced_snapshot_due
&& now.duration_since(client.last_frame_sent) >= frame_interval
});
if !due {
continue;
}
let snapshot = screen_snapshot(session.parser.screen());
let output_seq = session.output_seq;
for (client_id, client) in session.clients.iter_mut() {
if !client.paced_snapshot_due
|| now.duration_since(client.last_frame_sent) < frame_interval
{
continue;
}
client.send_seq += 1;
client.epoch_packets = client.epoch_packets.saturating_add(1);
let frame = Frame {
session: session_name.clone(),
output_seq,
bytes: snapshot.clone(),
snapshot: true,
closed: false,
};
let body = protocol::to_body(&frame)?;
let packet = protocol::encode_encrypted(
PacketKind::Frame,
*client_id,
client.send_seq,
client.last_acked,
&client.session_key,
SERVER_TO_CLIENT,
&body,
)?;
while client.pending.len() >= retransmit_window {
client.pending.pop_front();
}
client.pending.push_back(PendingFrame {
output_seq,
packet: packet.clone(),
last_sent: now,
attempts: 0,
});
client.last_frame_sent = now;
client.paced_snapshot_due = false;
sends.push((client.endpoint, packet));
}
}
sends
};
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3787,7 +3746,7 @@ async fn maybe_rekey_clients(
sends
};
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3950,7 +3909,7 @@ mod tests {
use super::*;
#[test]
fn persists_all_sessions_when_enabled() {
fn persists_named_sessions_when_enabled() {
let (pty_tx, _rx) = mpsc::unbounded_channel();
let config = ServerConfig {
persist_sessions: true,
@@ -3960,7 +3919,7 @@ mod tests {
let state = ServerState::new(config, [0u8; 32], pty_tx);
assert!(state.should_persist_session("default")); // prewarmed
assert!(state.should_persist_session("work")); // explicitly named
assert!(state.should_persist_session("term-1781470634216-76685")); // implicit reconnect
assert!(!state.should_persist_session("term-1781470634216-76685")); // one-off terminal
}
#[test]
@@ -4077,8 +4036,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -4156,6 +4113,63 @@ mod tests {
assert!(locked.lookup_client(&client_id).is_none());
}
#[tokio::test]
async fn forged_plaintext_detach_does_not_remove_client() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
state
.ensure_session("work", 80, 24, "forward-only", &[])
.unwrap();
let client_id = [21u8; 16];
let session_key = [22u8; 32];
state.insert_client("work", client_id, test_client_state(session_key));
let state = Arc::new(Mutex::new(state));
let mut packet = protocol::decode(
&protocol::encode_plain(PacketKind::Detach, client_id, 1, 0, b"").unwrap(),
)
.unwrap();
packet.header.session_key_id = protocol::session_key_id(&session_key);
assert!(handle_detach(&state, &packet).await.is_err());
let locked = state.lock().unwrap();
assert!(
locked.lookup_client(&client_id).is_some(),
"plaintext detach with a copied key id must not remove a live client"
);
}
#[tokio::test]
async fn authenticated_detach_removes_client() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
state
.ensure_session("work", 80, 24, "forward-only", &[])
.unwrap();
let client_id = [23u8; 16];
let session_key = [24u8; 32];
state.insert_client("work", client_id, test_client_state(session_key));
let state = Arc::new(Mutex::new(state));
let packet = protocol::decode(
&protocol::encode_encrypted(
PacketKind::Detach,
client_id,
1,
0,
&session_key,
CLIENT_TO_SERVER,
b"",
)
.unwrap(),
)
.unwrap();
handle_detach(&state, &packet).await.unwrap();
let locked = state.lock().unwrap();
assert!(locked.lookup_client(&client_id).is_none());
}
#[tokio::test]
async fn duplicate_stream_open_for_open_stream_resends_ok() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4395,8 +4409,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::new(),
@@ -4491,8 +4503,6 @@ mod tests {
rows: 24,
last_seen: Instant::now(),
pending: VecDeque::new(),
last_frame_sent: Instant::now() - Duration::from_secs(3600),
paced_snapshot_due: false,
allowed_forwardings: Vec::new(),
stream_writers: HashMap::new(),
opened_streams: HashSet::from([42]),
@@ -4562,7 +4572,7 @@ mod tests {
}
#[tokio::test]
async fn output_pacing_coalesces_fast_frames_into_snapshot() {
async fn live_terminal_output_is_never_replaced_by_paced_snapshots() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
let config = ServerConfig {
output_frame_interval_ms: 1000,
@@ -4614,25 +4624,6 @@ mod tests {
let frame: Frame = protocol::from_body(&plain).unwrap();
assert!(!frame.snapshot);
assert_eq!(frame.bytes, b"first");
assert!(
tokio::time::timeout(Duration::from_millis(30), receiver.recv_from(&mut buf))
.await
.is_err(),
"second frame should be coalesced"
);
{
let mut locked = state.lock().unwrap();
let client = locked
.sessions
.get_mut("test")
.unwrap()
.clients
.get_mut(&client_id)
.unwrap();
client.last_frame_sent = Instant::now() - Duration::from_secs(2);
}
flush_paced_snapshots(&state, &sender).await.unwrap();
let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
.await
.unwrap()
@@ -4640,8 +4631,9 @@ mod tests {
let packet = protocol::decode(&buf[..n]).unwrap();
let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
let frame: Frame = protocol::from_body(&plain).unwrap();
assert!(frame.snapshot);
assert!(!frame.snapshot);
assert_eq!(frame.output_seq, 2);
assert_eq!(frame.bytes, b"second");
}
#[test]
@@ -4656,4 +4648,25 @@ mod tests {
};
remote_bind_allowed(&config, "0.0.0.0").unwrap();
}
#[cfg(unix)]
#[test]
fn server_udp_send_classifier_treats_network_roaming_errors_as_transient() {
for code in [
libc::EADDRNOTAVAIL,
libc::ECONNREFUSED,
libc::ECONNRESET,
libc::EHOSTDOWN,
libc::EHOSTUNREACH,
libc::ENETDOWN,
libc::ENETRESET,
libc::ENETUNREACH,
libc::ETIMEDOUT,
] {
assert!(
is_transient_udp_send_error(&std::io::Error::from_raw_os_error(code)),
"expected errno {code} to be transient"
);
}
}
}
+1
View File
@@ -23,3 +23,4 @@ pub mod pty;
pub mod server;
pub mod ssh_agent;
pub mod transport;
pub mod udp;
+4
View File
@@ -435,6 +435,10 @@ pub fn run_holder(
let mut cmd = [0u8; 1];
match stream.read(&mut cmd) {
Ok(1) if cmd[0] == HOLDER_CMD_SHUTDOWN => {
if shell_pid > 0 {
let _ = unsafe { libc::kill(shell_pid, libc::SIGHUP) };
let _ = unsafe { libc::kill(shell_pid, libc::SIGTERM) };
}
let _ = std::fs::remove_dir_all(runtime_dir);
std::process::exit(0);
}
+1 -1
View File
@@ -280,7 +280,7 @@ pub fn decode(input: &[u8]) -> Result<Packet> {
pub fn decrypt_body(packet: &Packet, key: &[u8; 32], direction: u32) -> Result<Vec<u8>> {
if packet.header.flags & 1 == 0 {
return Ok(packet.body.clone());
bail!("packet body is not encrypted");
}
let nonce = crypto::nonce_from(direction, packet.header.seq);
let aad = packet.header.aad();
+120 -6
View File
@@ -13,6 +13,7 @@ use crate::transport::{
DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
service_name_from_target,
};
use crate::udp::is_transient_udp_send_error;
use anyhow::{Context, Result, anyhow, bail};
use ed25519_dalek::SigningKey;
use std::collections::{HashMap, HashSet};
@@ -97,6 +98,7 @@ pub enum DoshServerEvent {
Ignored,
}
#[derive(Debug, Clone)]
struct PendingServerAuth {
client: native::NativeClientHello,
server: NativeServerHello,
@@ -255,7 +257,7 @@ impl DoshServer {
};
let body = protocol::to_body(&NativeServerHelloBody { hello })?;
let out = protocol::encode_plain(PacketKind::NativeServerHello, pending_id, 1, 0, &body)?;
self.socket.send_to(&out, peer).await?;
let _ = send_udp(&self.socket, &out, peer).await?;
Ok(())
}
@@ -327,7 +329,7 @@ impl DoshServer {
peer: SocketAddr,
packet: &protocol::Packet,
) -> Result<DoshServerEvent> {
let Some(pending) = self.pending.remove(&packet.header.conn_id) else {
let Some(pending) = self.pending.get(&packet.header.conn_id).cloned() else {
self.send_reject(peer, packet.header.conn_id, "unknown native auth")
.await?;
return Ok(DoshServerEvent::Ignored);
@@ -338,13 +340,28 @@ impl DoshServer {
return Ok(DoshServerEvent::Ignored);
}
if pending.created_at.elapsed() > self.config.auth_timeout {
self.pending.remove(&packet.header.conn_id);
self.send_reject(peer, packet.header.conn_id, "native auth expired")
.await?;
return Ok(DoshServerEvent::Ignored);
}
let body = protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER)?;
let req: NativeUserAuthBody = protocol::from_body(&body)?;
let body = match protocol::decrypt_body(packet, &pending.session_key, CLIENT_TO_SERVER) {
Ok(body) => body,
Err(_) => {
self.send_reject(peer, packet.header.conn_id, "native auth decrypt failed")
.await?;
return Ok(DoshServerEvent::Ignored);
}
};
let req: NativeUserAuthBody = match protocol::from_body(&body) {
Ok(req) => req,
Err(_) => {
self.send_reject(peer, packet.header.conn_id, "invalid native auth body")
.await?;
return Ok(DoshServerEvent::Ignored);
}
};
let services = match self.verify_auth_and_services(&pending, &req.auth) {
Ok(services) => services,
Err(err) => {
@@ -353,6 +370,7 @@ impl DoshServer {
return Ok(DoshServerEvent::Ignored);
}
};
self.pending.remove(&packet.header.conn_id);
let conn_id = crypto::random_16();
let key_id = protocol::session_key_id(&pending.session_key);
@@ -381,7 +399,7 @@ impl DoshServer {
SERVER_TO_CLIENT,
&body,
)?;
self.socket.send_to(&out, peer).await?;
let _ = send_udp(&self.socket, &out, peer).await?;
let transport = DoshTransport::new(
Arc::clone(&self.socket),
@@ -428,7 +446,7 @@ impl DoshServer {
reason: reason.to_string(),
})?;
let out = protocol::encode_plain(PacketKind::AttachReject, conn_id, 0, 0, &body)?;
self.socket.send_to(&out, peer).await?;
let _ = send_udp(&self.socket, &out, peer).await?;
Ok(())
}
@@ -439,6 +457,14 @@ impl DoshServer {
}
}
async fn send_udp(socket: &UdpSocket, packet: &[u8], peer: SocketAddr) -> Result<bool> {
match socket.send_to(packet, peer).await {
Ok(_) => Ok(true),
Err(err) if is_transient_udp_send_error(&err) => Ok(false),
Err(err) => Err(err).with_context(|| format!("send UDP packet to {peer}")),
}
}
fn validate_requested_services(
allowed_services: &HashSet<String>,
requests: &[ForwardingRequest],
@@ -477,6 +503,7 @@ mod tests {
use super::*;
use crate::client::DoshClient;
use crate::config::{ClientConfig, HostsConfig};
use crate::protocol::{CLIENT_TO_SERVER, NativeUserAuthBody};
use crate::transport::TransportEvent;
use ed25519_dalek::SigningKey;
@@ -588,4 +615,91 @@ mod tests {
}
}
}
#[tokio::test]
async fn bad_native_auth_does_not_consume_pending_challenge() {
let dir = tempfile::tempdir().unwrap();
let host_key = dir.path().join("host_key");
let authorized_keys = dir.path().join("authorized_keys");
let signing = SigningKey::from_bytes(&[93u8; 32]);
let keypair = ssh_key::private::Ed25519Keypair::from(&signing);
let private =
ssh_key::PrivateKey::new(ssh_key::private::KeypairData::from(keypair), "").unwrap();
std::fs::write(
&authorized_keys,
format!("{}\n", private.public_key().to_openssh().unwrap()),
)
.unwrap();
let server_config = ServerConfig {
host_key: host_key.to_string_lossy().to_string(),
authorized_keys: vec![authorized_keys.to_string_lossy().to_string()],
..ServerConfig::default()
};
let server_config = DoshServerConfig::new(server_config)
.bind_addr("127.0.0.1:0".parse().unwrap())
.require_current_user(false);
let mut server = DoshServer::bind(server_config).await.unwrap();
let peer: SocketAddr = "127.0.0.1:9".parse().unwrap();
let (client_secret, client_public) = native::generate_native_ephemeral();
let hello = native::NativeClientHello {
protocol_version: native::NATIVE_PROTOCOL_VERSION,
client_random: crypto::random_32(),
client_ephemeral_public: client_public,
requested_host: "127.0.0.1".to_string(),
requested_user: "sdk-user".to_string(),
requested_session: "test".to_string(),
requested_mode: "forward-only".to_string(),
terminal_size: (80, 24),
supported_aead: vec!["chacha20poly1305".to_string()],
supported_user_key_algorithms: vec!["ssh-ed25519".to_string()],
cached_host_key_fingerprint: None,
attach_ticket_envelope: None,
requested_env: Vec::new(),
};
let (pending_id, server_hello) = server.build_server_hello(hello.clone(), peer).unwrap();
let session_key = native::derive_native_session_key(
&client_secret,
server_hello.server_ephemeral_public,
&hello,
&server_hello,
)
.unwrap();
let auth = native::sign_user_auth(&signing, &hello, &server_hello, Vec::new()).unwrap();
let body = protocol::to_body(&NativeUserAuthBody { auth }).unwrap();
let bad = protocol::encode_encrypted(
PacketKind::NativeUserAuth,
pending_id,
2,
1,
&[7u8; 32],
CLIENT_TO_SERVER,
&body,
)
.unwrap();
let bad = protocol::decode(&bad).unwrap();
assert!(matches!(
server.handle_user_auth(peer, &bad).await.unwrap(),
DoshServerEvent::Ignored
));
assert!(server.pending.contains_key(&pending_id));
let valid = protocol::encode_encrypted(
PacketKind::NativeUserAuth,
pending_id,
2,
1,
&session_key,
CLIENT_TO_SERVER,
&body,
)
.unwrap();
let valid = protocol::decode(&valid).unwrap();
assert!(matches!(
server.handle_user_auth(peer, &valid).await.unwrap(),
DoshServerEvent::Accepted(_)
));
assert!(!server.pending.contains_key(&pending_id));
}
}
+114 -10
View File
@@ -19,6 +19,7 @@ use crate::protocol::{
self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData,
StreamOpen, StreamOpenOk, StreamOpenReject, StreamWindowAdjust,
};
use crate::udp::is_transient_udp_send_error;
use anyhow::{Result, anyhow, bail};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::net::SocketAddr;
@@ -348,8 +349,8 @@ impl StreamMux {
&mut self,
adjust: StreamWindowAdjust,
) -> Result<Vec<OutgoingStreamPacket>> {
self.ack_data(adjust.stream_id, adjust.received_offset);
self.add_credit(adjust.stream_id, adjust.bytes as usize);
let acked_bytes = self.ack_data(adjust.stream_id, adjust.received_offset);
self.add_credit(adjust.stream_id, acked_bytes);
self.flush_pending_data(adjust.stream_id)
}
@@ -556,9 +557,9 @@ impl StreamMux {
(chunks, consumed, *expected)
}
fn ack_data(&mut self, stream_id: u64, received_offset: u64) {
fn ack_data(&mut self, stream_id: u64, received_offset: u64) -> usize {
let Some(sent) = self.sent_data.get_mut(&stream_id) else {
return;
return 0;
};
let acked_offsets = sent
.iter()
@@ -567,12 +568,16 @@ impl StreamMux {
(end <= received_offset).then_some(*offset)
})
.collect::<Vec<_>>();
let mut acked_bytes = 0usize;
for offset in acked_offsets {
sent.remove(&offset);
if let Some(chunk) = sent.remove(&offset) {
acked_bytes = acked_bytes.saturating_add(chunk.bytes.len());
}
}
if sent.is_empty() {
self.sent_data.remove(&stream_id);
}
acked_bytes
}
fn add_credit(&mut self, stream_id: u64, bytes: usize) {
@@ -732,11 +737,17 @@ impl DoshTransport {
if packet.header.conn_id != self.conn_id {
continue;
}
let plain = match protocol::decrypt_body(
&packet,
&self.session_key,
self.role.recv_direction(),
) {
Ok(plain) => plain,
Err(_) => continue,
};
if !self.replay.accept(packet.header.seq) {
continue;
}
let plain =
protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction())?;
self.peer_addr = peer;
self.ack_seq = self.ack_seq.max(packet.header.seq);
self.last_contact = Instant::now();
@@ -751,14 +762,21 @@ impl DoshTransport {
datagram: &[u8],
peer: SocketAddr,
) -> Result<SessionEvent> {
let packet = protocol::decode(datagram)?;
let packet = match protocol::decode(datagram) {
Ok(packet) => packet,
Err(_) => return Ok(SessionEvent::Ignored),
};
if packet.header.conn_id != self.conn_id {
return Ok(SessionEvent::Ignored);
}
let plain =
match protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction()) {
Ok(plain) => plain,
Err(_) => return Ok(SessionEvent::Ignored),
};
if !self.replay.accept(packet.header.seq) {
return Ok(SessionEvent::Ignored);
}
let plain = protocol::decrypt_body(&packet, &self.session_key, self.role.recv_direction())?;
self.peer_addr = peer;
self.ack_seq = self.ack_seq.max(packet.header.seq);
self.last_contact = Instant::now();
@@ -815,7 +833,12 @@ impl DoshTransport {
async fn send_kind(&mut self, kind: PacketKind, body: &[u8]) -> Result<()> {
let encoded = self.encode_kind(kind, body)?;
self.socket.send_to(&encoded, self.peer_addr).await?;
if let Err(err) = self.socket.send_to(&encoded, self.peer_addr).await {
if is_transient_udp_send_error(&err) {
return Ok(());
}
return Err(err.into());
}
Ok(())
}
@@ -980,6 +1003,30 @@ mod tests {
assert_eq!(data.bytes, b"!");
}
#[test]
fn cumulative_window_adjust_restores_credit_even_if_delta_was_lost() {
let mut mux = StreamMux::new(TransportConfig {
initial_window: 5,
retransmit_after: Duration::from_secs(1),
..TransportConfig::default()
});
mux.open_stream(1, "@dosh-test", 0).unwrap();
mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
assert_eq!(mux.send_data(1, b"hello".to_vec()).unwrap().len(), 1);
assert_eq!(mux.send_credit[&1], 0);
let flushed = mux
.handle_window_adjust(StreamWindowAdjust {
stream_id: 1,
received_offset: 5,
bytes: 0,
})
.unwrap();
assert!(flushed.is_empty());
assert_eq!(mux.send_credit[&1], 5);
assert_eq!(mux.send_data(1, b"!".to_vec()).unwrap().len(), 1);
}
#[test]
fn handle_packet_decodes_and_dispatches_stream_packets() {
let mut mux = StreamMux::new(TransportConfig::default());
@@ -1096,4 +1143,61 @@ mod tests {
assert!(matches!(server.recv().await.unwrap(), SessionEvent::Ping));
assert_eq!(server.peer_addr(), roaming_addr);
}
#[tokio::test]
async fn unauthenticated_datagram_does_not_poison_replay_window() {
let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let peer: SocketAddr = "127.0.0.1:9".parse().unwrap();
let key = [11u8; 32];
let wrong_key = [12u8; 32];
let conn_id = [4u8; 16];
let mut transport = DoshTransport::new_owned(
socket,
SessionTransportConfig {
role: SessionRole::Client,
conn_id,
session_key: key,
peer_addr: peer,
initial_send_seq: 1,
initial_ack: 0,
stream: TransportConfig::default(),
},
);
let bad = protocol::encode_encrypted(
PacketKind::Pong,
conn_id,
9,
0,
&wrong_key,
SERVER_TO_CLIENT,
b"",
)
.unwrap();
let valid = protocol::encode_encrypted(
PacketKind::Pong,
conn_id,
9,
0,
&key,
SERVER_TO_CLIENT,
b"",
)
.unwrap();
assert!(matches!(
transport
.handle_datagram(b"not a dosh packet", peer)
.await
.unwrap(),
SessionEvent::Ignored
));
assert!(matches!(
transport.handle_datagram(&bad, peer).await.unwrap(),
SessionEvent::Ignored
));
assert!(matches!(
transport.handle_datagram(&valid, peer).await.unwrap(),
SessionEvent::Pong
));
}
}
+98
View File
@@ -0,0 +1,98 @@
//! UDP error classification shared by Dosh clients, servers, and embedders.
pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
matches!(
err.kind(),
std::io::ErrorKind::Interrupted
| std::io::ErrorKind::TimedOut
| std::io::ErrorKind::WouldBlock
) || err.raw_os_error().is_some_and(is_transient_udp_os_error)
}
#[cfg(unix)]
fn is_transient_udp_os_error(code: i32) -> bool {
matches!(
code,
libc::EADDRNOTAVAIL
| libc::ECONNREFUSED
| libc::ECONNRESET
| libc::EHOSTDOWN
| libc::EHOSTUNREACH
| libc::ENETDOWN
| libc::ENETRESET
| libc::ENETUNREACH
| libc::ETIMEDOUT
)
}
#[cfg(windows)]
fn is_transient_udp_os_error(code: i32) -> bool {
matches!(
code,
10049 // WSAEADDRNOTAVAIL
| 10050 // WSAENETDOWN
| 10051 // WSAENETUNREACH
| 10052 // WSAENETRESET
| 10054 // WSAECONNRESET
| 10060 // WSAETIMEDOUT
| 10061 // WSAECONNREFUSED
| 10064 // WSAEHOSTDOWN
| 10065 // WSAEHOSTUNREACH
)
}
#[cfg(not(any(unix, windows)))]
fn is_transient_udp_os_error(_code: i32) -> bool {
false
}
#[cfg(test)]
mod tests {
use super::is_transient_udp_send_error;
#[test]
fn classifies_portable_transient_send_errors() {
for kind in [
std::io::ErrorKind::Interrupted,
std::io::ErrorKind::TimedOut,
std::io::ErrorKind::WouldBlock,
] {
assert!(is_transient_udp_send_error(&std::io::Error::from(kind)));
}
assert!(!is_transient_udp_send_error(&std::io::Error::from(
std::io::ErrorKind::PermissionDenied,
)));
}
#[cfg(unix)]
#[test]
fn classifies_unix_network_churn_as_transient() {
for code in [
libc::EADDRNOTAVAIL,
libc::ECONNREFUSED,
libc::ECONNRESET,
libc::EHOSTDOWN,
libc::EHOSTUNREACH,
libc::ENETDOWN,
libc::ENETRESET,
libc::ENETUNREACH,
libc::ETIMEDOUT,
] {
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(code)
));
}
}
#[cfg(windows)]
#[test]
fn classifies_windows_network_churn_as_transient() {
for code in [
10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065,
] {
assert!(is_transient_udp_send_error(
&std::io::Error::from_raw_os_error(code)
));
}
}
}
+141 -9
View File
@@ -258,7 +258,8 @@ fn relay_loop(
let mut upstream = new_upstream();
let mut client_addr: Option<SocketAddr> = None;
let mut rng = StdRng::seed_from_u64(seed);
let mut held: Option<(Vec<u8>, bool)> = None; // (packet, is_c2s) held for reorder
let mut held_c2s: Option<Vec<u8>> = None;
let mut held_s2c: Option<Vec<u8>> = None;
let mut buf = [0u8; 65535];
loop {
@@ -288,10 +289,9 @@ fn relay_loop(
&upstream,
server_addr,
packet,
true,
&controls,
&mut rng,
&mut held,
&mut held_c2s,
);
}
}
@@ -310,7 +310,12 @@ fn relay_loop(
let drop_pct = controls.drop_s2c_percent.load(Ordering::SeqCst);
if drop_pct == 0 || rng.gen_range(0..100) >= drop_pct {
forward_with_effects(
&front, dst, packet, false, &controls, &mut rng, &mut held,
&front,
dst,
packet,
&controls,
&mut rng,
&mut held_s2c,
);
}
}
@@ -336,23 +341,22 @@ fn forward_with_effects(
out: &UdpSocket,
dst: SocketAddr,
packet: Vec<u8>,
is_c2s: bool,
controls: &RelayControls,
rng: &mut StdRng,
held: &mut Option<(Vec<u8>, bool)>,
held: &mut Option<Vec<u8>>,
) {
// Reorder: if armed, hold this packet and release the previously held one
// afterward (so two consecutive packets swap order).
if controls.reorder_next.swap(false, Ordering::SeqCst) {
if let Some((prev, _)) = held.take() {
if let Some(prev) = held.take() {
let _ = out.send_to(&packet, dst);
let _ = out.send_to(&prev, dst);
return;
}
*held = Some((packet, is_c2s));
*held = Some(packet);
return;
}
if let Some((prev, _)) = held.take() {
if let Some(prev) = held.take() {
let _ = out.send_to(&prev, dst);
}
@@ -767,6 +771,46 @@ fn wait_for_text(socket: &UdpSocket, key: &[u8; 32], needle: &str, millis: u64)
acc.contains(needle)
}
fn wait_for_text_with_ack(
socket: &UdpSocket,
relay: &Relay,
client_id: [u8; 16],
seq: &mut u64,
key: &[u8; 32],
needle: &str,
millis: u64,
) -> bool {
let prev = socket.read_timeout().unwrap();
socket
.set_read_timeout(Some(Duration::from_millis(100)))
.unwrap();
let deadline = Instant::now() + Duration::from_millis(millis);
let mut acc = String::new();
while Instant::now() < deadline {
if let Some((_header, frame)) = recv_frame(socket, key) {
acc.push_str(&String::from_utf8_lossy(&frame.bytes));
let ack = protocol::encode_encrypted(
PacketKind::Ack,
client_id,
*seq,
frame.output_seq,
key,
CLIENT_TO_SERVER,
b"",
)
.unwrap();
*seq += 1;
socket.send_to(&ack, relay.front_addr()).unwrap();
if acc.contains(needle) {
socket.set_read_timeout(prev).unwrap();
return true;
}
}
}
socket.set_read_timeout(prev).unwrap();
acc.contains(needle)
}
#[test]
fn session_survives_packet_loss_and_reorder() {
let dir = tempfile::tempdir().unwrap();
@@ -820,6 +864,94 @@ fn session_survives_packet_loss_and_reorder() {
);
}
#[test]
#[ignore = "30-minute hostile-network TUI soak; run with `DOSH_BADNET_SOAK_SECONDS=1800 cargo test --test hostile_network bad_network_tui_work_soak_30m -- --ignored --nocapture`"]
fn bad_network_tui_work_soak_30m() {
let soak_secs = std::env::var("DOSH_BADNET_SOAK_SECONDS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or(30 * 60);
let dir = tempfile::tempdir().unwrap();
let port = free_udp_port();
let config = write_server_config(&dir, port);
let mut server = start_server(&dir, &config);
let relay = Relay::spawn(port, 0xBADC0DEu64);
let (socket, bootstrap, ok) = attach_through_relay(&config, &relay);
relay.set_drop_c2s(15);
relay.set_drop_s2c(20);
relay.set_dup(10);
let deadline = Instant::now() + Duration::from_secs(soak_secs);
let mut seq = 2u64;
let mut iteration = 0u64;
while Instant::now() < deadline {
if iteration.is_multiple_of(2) {
relay.arm_reorder();
}
if iteration > 0 && iteration.is_multiple_of(7) {
let _ = relay.rebind_upstream();
}
if iteration > 0 && iteration.is_multiple_of(11) {
relay.set_drop_s2c(100);
thread::sleep(Duration::from_millis(750));
relay.set_drop_s2c(20);
}
let marker = format!("DOSH_BADNET_TUI_{iteration}");
let command = format!(
"stty -echo; \
printf '\\033[?1049h\\033[?2026h\\033[?25l'; \
for i in 1 2 3 4 5 6; do \
printf '\\033[%s;4H{} %s\\033[0m' \"$i\" \"$i\"; \
done; \
printf '\\033[?25h\\033[?2026l\\033[?1049l'\n",
marker
);
let step_deadline = Instant::now() + Duration::from_secs(8);
let mut seen = false;
let mut attempts = 0;
while Instant::now() < step_deadline && attempts < 12 {
send_input(
&socket,
&relay,
ok.client_id,
seq,
0,
&bootstrap.session_key,
command.as_bytes(),
);
seq += 1;
attempts += 1;
if wait_for_text_with_ack(
&socket,
&relay,
ok.client_id,
&mut seq,
&bootstrap.session_key,
&marker,
500,
) {
seen = true;
break;
}
}
assert!(
seen,
"TUI marker {marker} did not arrive during hostile-network soak"
);
iteration += 1;
thread::sleep(Duration::from_millis(500));
}
relay.clear_impairments();
drop(relay);
let _ = server.kill();
let _ = server.wait();
}
#[test]
fn duplicated_and_replayed_input_is_applied_at_most_once() {
let dir = tempfile::tempdir().unwrap();
+225 -26
View File
@@ -98,6 +98,35 @@ persist_sessions = false
config
}
fn write_server_config_with_output_interval(
dir: &tempfile::TempDir,
port: u16,
output_frame_interval_ms: u64,
) -> std::path::PathBuf {
let config = write_server_config(dir, port);
let mut raw = fs::read_to_string(&config).unwrap();
raw.push_str(&format!(
"output_frame_interval_ms = {output_frame_interval_ms}\n"
));
fs::write(&config, raw).unwrap();
config
}
fn write_server_config_with_retransmit_window(
dir: &tempfile::TempDir,
port: u16,
retransmit_window: u64,
) -> std::path::PathBuf {
let config = write_server_config(dir, port);
let mut raw = fs::read_to_string(&config).unwrap();
raw = raw.replace(
"retransmit_window = 256\n",
&format!("retransmit_window = {retransmit_window}\n"),
);
fs::write(&config, raw).unwrap();
config
}
fn start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child {
let server = env!("CARGO_BIN_EXE_dosh-server");
let child = Command::new(server)
@@ -1363,18 +1392,10 @@ fn server_retransmits_unacked_output_frames() {
);
let mut seen = Vec::new();
let mut duplicate = None;
let deadline = std::time::Instant::now() + Duration::from_secs(3);
let deadline = std::time::Instant::now() + Duration::from_secs(4);
while std::time::Instant::now() < deadline {
if let Some((header, frame)) = recv_frame(&socket, &bootstrap.session_key) {
if String::from_utf8_lossy(&frame.bytes).contains("DOSH_RETRANSMIT") {
if seen
.iter()
.any(|(_, output_seq)| *output_seq == frame.output_seq)
{
duplicate = Some((header.seq, frame.output_seq));
break;
}
seen.push((header.seq, frame.output_seq));
}
}
@@ -1383,9 +1404,18 @@ fn server_retransmits_unacked_output_frames() {
let _ = server.kill();
let _ = server.wait();
let max_same_output_seq = seen
.iter()
.map(|(_, output_seq)| {
seen.iter()
.filter(|(_, candidate)| candidate == output_seq)
.count()
})
.max()
.unwrap_or(0);
assert!(
duplicate.is_some(),
"expected retransmitted same output seq, seen={seen:?}"
max_same_output_seq >= 10,
"expected retransmission to continue past the old 8-attempt ceiling, seen={seen:?}"
);
}
@@ -1639,6 +1669,110 @@ fn unicode_output_survives_transport() {
);
}
#[test]
fn tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution() {
let dir = tempfile::tempdir().unwrap();
let port = free_udp_port();
let config = write_server_config_with_output_interval(&dir, port, 1000);
let mut server = start_server(&dir, &config);
let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write");
let graph = "DOSH_GRAPH ⣀⣤⣶⣿ █▇▆▅▄▃▂▁ ╭╮╯╰";
let input = Input {
bytes: format!(
"printf '\\033[?1049h\\033[?25l'; for i in 1 2 3 4 5; do printf '\\033[H{} %s\\n' \"$i\"; done; printf '\\033[?25h\\033[?1049l'\n",
graph
)
.into_bytes(),
};
send_encrypted(
&socket,
port,
PacketKind::Input,
ok.client_id,
2,
0,
&bootstrap.session_key,
&protocol::to_body(&input).unwrap(),
);
let text = collect_frame_text(&socket, &bootstrap.session_key, 3000);
let _ = server.kill();
let _ = server.wait();
assert!(
text.contains(graph) && text.matches("DOSH_GRAPH").count() >= 5,
"expected repeated raw graph glyph frames, got {text:?}"
);
assert!(
text.contains("\x1b[?25l") && text.contains("\x1b[?25h"),
"expected cursor visibility controls to survive around graph repaint, got {text:?}"
);
}
#[test]
fn btop_style_graph_output_stays_raw_when_retransmit_history_overflows() {
let dir = tempfile::tempdir().unwrap();
let port = free_udp_port();
let config = write_server_config_with_retransmit_window(&dir, port, 3);
let mut server = start_server(&dir, &config);
let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write");
let graph = "DOSH_BTOP_NET ⠀⠁⠃⠇⡇⣇⣧⣷⣿";
let input = Input {
bytes: format!(
"stty -echo; \
printf '\\033[?1049h\\033[?2026h\\033[?25l'; \
for i in 1 2 3 4 5 6 7 8 9 10; do \
printf '\\033[6;4H\\033[38;2;80;220;140m{} %s\\033[0m' \"$i\"; \
done; \
printf '\\033[?25h\\033[?2026l\\033[?1049l'\n",
graph
)
.into_bytes(),
};
send_encrypted(
&socket,
port,
PacketKind::Input,
ok.client_id,
2,
0,
&bootstrap.session_key,
&protocol::to_body(&input).unwrap(),
);
let mut text = String::new();
let mut snapshots = 0usize;
let deadline = std::time::Instant::now() + Duration::from_secs(3);
while std::time::Instant::now() < deadline
&& !(text.matches("DOSH_BTOP_NET").count() >= 10 && text.contains("\x1b[?2026l"))
{
if let Some((_header, frame)) = recv_frame(&socket, &bootstrap.session_key) {
if frame.snapshot {
snapshots += 1;
}
text.push_str(&String::from_utf8_lossy(&frame.bytes));
}
}
let _ = server.kill();
let _ = server.wait();
assert_eq!(
snapshots, 0,
"live btop-style graph output must not be replaced by snapshots; got {snapshots}"
);
assert!(
text.matches("DOSH_BTOP_NET").count() >= 10
&& text.contains("")
&& text.contains("\x1b[38;2;80;220;140m")
&& text.contains("\x1b[?2026h")
&& text.contains("\x1b[?2026l"),
"expected raw btop-style graph output to survive under retransmit pressure, got {text:?}"
);
}
#[test]
fn large_tui_paint_is_delivered_in_mtu_safe_frames() {
let dir = tempfile::tempdir().unwrap();
@@ -2385,6 +2519,9 @@ fn write_persistent_server_config(dir: &tempfile::TempDir, port: u16) -> std::pa
let mut raw = fs::read_to_string(&config).unwrap();
// The base writer hardcodes `persist_sessions = false`; flip it on.
raw = raw.replace("persist_sessions = false", "persist_sessions = true");
// Persistence tests attach the sessions they need explicitly. Leaving the
// base prewarm on here creates a detached default holder in every test.
raw = raw.replace("prewarm_sessions = [\"default\"]", "prewarm_sessions = []");
fs::write(&config, raw).unwrap();
config
}
@@ -2420,6 +2557,45 @@ fn kill_holder(sessions_dir: &Path, session: &str) {
.status();
}
/// Create a holder the way older Dosh builds did for implicit sessions, so
/// startup migration can be tested without depending on the new spawn policy.
fn spawn_legacy_holder(sessions_dir: &Path, session: &str) {
let server = env!("CARGO_BIN_EXE_dosh-server");
let runtime_dir = dosh::persist::ensure_runtime_dir(sessions_dir, session).unwrap();
let mut launcher = Command::new(server)
.arg("hold")
.arg("--runtime-dir")
.arg(&runtime_dir)
.arg("--session")
.arg(session)
.arg("--shell")
.arg("/bin/sh")
.arg("--cols")
.arg("80")
.arg("--rows")
.arg("24")
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while std::time::Instant::now() < deadline {
if holder_shell_pid(sessions_dir, session).is_some()
&& runtime_dir.join("holder.sock").exists()
{
let _ = launcher.wait();
return;
}
if let Some(status) = launcher.try_wait().unwrap() {
panic!("legacy holder launcher exited before ready: {status}");
}
thread::sleep(Duration::from_millis(20));
}
let _ = launcher.kill();
let _ = launcher.wait();
panic!("legacy holder did not become ready");
}
/// Send one encrypted Input line and give the shell a moment to act.
fn type_line(socket: &UdpSocket, port: u16, ok: &AttachOk, key: &[u8; 32], seq: u64, line: &str) {
let input = Input {
@@ -2557,7 +2733,7 @@ fn session_survives_server_restart_same_shell_and_screen() {
}
#[test]
fn implicit_session_survives_server_restart_for_same_ticket_holder() {
fn implicit_session_does_not_create_restart_holder() {
let dir = tempfile::tempdir().unwrap();
let sessions_dir = dir.path().join("sessions");
let port = free_udp_port();
@@ -2580,20 +2756,14 @@ fn implicit_session_survives_server_restart_for_same_ticket_holder() {
let shell_pid_before = holder_shell_pid(&sessions_dir, &session);
assert!(
shell_pid_before.is_some(),
"implicit sessions should get a persistent holder when persistence is enabled"
shell_pid_before.is_none(),
"implicit sessions must not get persistent holders when persistence is enabled"
);
let _ = server.kill();
let _ = server.wait();
thread::sleep(Duration::from_millis(300));
let pid = shell_pid_before.unwrap();
assert!(
unsafe { libc::kill(pid, 0) } == 0,
"implicit session shell pid {pid} must survive server restart"
);
let mut server = start_server(&dir, &config);
let (socket2, bootstrap2, ok2) = direct_attach_session(&config, port, "read-write", &session);
let key2 = bootstrap2.session_key;
@@ -2610,15 +2780,44 @@ fn implicit_session_survives_server_restart_for_same_ticket_holder() {
let _ = server.kill();
let _ = server.wait();
kill_holder(&sessions_dir, &session);
assert_eq!(
shell_pid_after, shell_pid_before,
"implicit session should re-adopt the same shell pid"
assert!(
shell_pid_after.is_none(),
"implicit restart attach must stay ephemeral"
);
assert!(
post.contains("IMPLICIT_MARK=implicit_restart_42"),
"implicit session state must survive restart, got {post:?}"
post.contains("IMPLICIT_MARK=") && !post.contains("implicit_restart_42"),
"implicit session must start fresh after server restart, got {post:?}"
);
}
#[test]
fn startup_discards_legacy_implicit_holders() {
let dir = tempfile::tempdir().unwrap();
let sessions_dir = dir.path().join("sessions");
let port = free_udp_port();
let config = write_persistent_server_config(&dir, port);
let session = protocol::generate_implicit_session_name();
spawn_legacy_holder(&sessions_dir, &session);
let shell_pid = holder_shell_pid(&sessions_dir, &session).expect("legacy holder shell pid");
assert!(
unsafe { libc::kill(shell_pid, 0) } == 0,
"legacy holder shell must be live before startup"
);
let mut server = start_server(&dir, &config);
thread::sleep(Duration::from_millis(300));
let _ = server.kill();
let _ = server.wait();
assert!(
holder_shell_pid(&sessions_dir, &session).is_none(),
"startup must remove old implicit holder metadata"
);
assert!(
unsafe { libc::kill(shell_pid, 0) } != 0,
"startup must shut down old implicit holder shell"
);
}
+13
View File
@@ -58,6 +58,19 @@ fn encrypted_packet_round_trips() {
assert_eq!(plain, b"hello");
}
#[test]
fn plaintext_packet_never_decrypts_as_authenticated_body() {
let key = crypto::random_32();
let mut decoded = protocol::decode(
&protocol::encode_plain(PacketKind::Input, crypto::random_16(), 1, 0, b"hello").unwrap(),
)
.unwrap();
decoded.header.session_key_id = protocol::session_key_id(&key);
let err = protocol::decrypt_body(&decoded, &key, CLIENT_TO_SERVER).unwrap_err();
assert!(err.to_string().contains("not encrypted"));
}
#[test]
fn encrypted_packet_rejects_wrong_session_key_id_before_decrypt() {
let key = crypto::random_32();
+52
View File
@@ -0,0 +1,52 @@
#[test]
fn quiet_update_keeps_prebuilt_enabled_by_default() {
let install = include_str!("../install.sh");
assert!(install.contains("use_prebuilt=\"${DOSH_USE_PREBUILT:-1}\""));
assert!(
!install.contains("use_prebuilt=0"),
"quiet updates must not force source builds"
);
}
#[test]
fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
let upload = include_str!("../scripts/upload-gitea-release.sh");
assert!(upload.contains("skipping $artifact: version"));
assert!(upload.contains("expected_version"));
let publish = include_str!("../scripts/publish-gitea-release.sh");
assert!(publish.contains("skipping stale artifact"));
assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\""));
}
#[test]
fn systemd_service_does_not_sandbox_remote_shells() {
let service = include_str!("../packaging/systemd/dosh-server.service");
let install = include_str!("../install.sh");
for raw in [service, install] {
assert!(raw.contains("KillMode=process"));
assert!(
raw.contains("Restart=always"),
"dosh-server should come back after accidental SIGTERM while preserving child shells"
);
for directive in [
"NoNewPrivileges=",
"PrivateTmp=",
"ProtectSystem=",
"ProtectHome=",
"RestrictAddressFamilies=",
"RestrictRealtime=",
"LockPersonality=",
"MemoryDenyWriteExecute=",
] {
assert!(
!raw.contains(directive),
"dosh sessions are user shells; systemd sandbox directive {directive} changes terminal/app behavior"
);
}
assert!(
!raw.contains("ReadWritePaths="),
"remote shells must not be restricted to dosh config/data paths"
);
}
}