Compare commits

..
8 Commits
Author SHA1 Message Date
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
11 changed files with 591 additions and 63 deletions
Generated
+1 -1
View File
@@ -436,7 +436,7 @@ dependencies = [
[[package]]
name = "dosh"
version = "0.1.15"
version = "1.0.0-rc2"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "dosh"
version = "0.1.15"
version = "1.0.0-rc2"
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
+1 -1
View File
@@ -426,7 +426,7 @@ Wants=network-online.target
[Service]
Type=simple
ExecStart=$bindir/dosh-server serve
Restart=on-failure
Restart=always
RestartSec=1
KillMode=process
+1 -1
View File
@@ -6,7 +6,7 @@ Wants=network-online.target
[Service]
Type=simple
ExecStart=%h/.local/bin/dosh-server serve
Restart=on-failure
Restart=always
RestartSec=1
KillMode=process
+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"
+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"
+283 -33
View File
@@ -63,6 +63,8 @@ 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);
/// 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 +3526,60 @@ 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}")),
}
}
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
}
#[allow(clippy::too_many_arguments)]
async fn try_native_auth(
socket: &UdpSocket,
@@ -4322,6 +4378,7 @@ 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;
if let Some(frame) = first_frame {
if !forward_only {
render_frame(&frame)?;
@@ -4337,7 +4394,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;
}
@@ -4373,10 +4436,19 @@ 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;
}
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
@@ -4411,8 +4483,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),
@@ -4431,15 +4510,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;
}
@@ -4460,6 +4547,9 @@ 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);
@@ -4477,8 +4567,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,
)?;
}
}
}
}
@@ -4512,6 +4609,9 @@ 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);
@@ -4562,6 +4662,9 @@ 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);
@@ -4610,6 +4713,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
@@ -4651,7 +4770,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)?;
@@ -4667,6 +4786,9 @@ 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);
@@ -5059,6 +5181,9 @@ 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);
@@ -5086,7 +5211,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
@@ -5400,6 +5525,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>,
@@ -5439,6 +5581,26 @@ 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 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,
@@ -5462,8 +5624,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(())
}
@@ -5489,6 +5662,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)
@@ -5497,13 +5671,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,
}
}
@@ -5525,6 +5701,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;
@@ -5595,7 +5786,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,
@@ -5607,8 +5798,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(
@@ -5951,7 +6141,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(())
}
@@ -6715,7 +6905,7 @@ async fn send_resize(
&body,
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -6735,7 +6925,7 @@ async fn send_ack(
b"",
)?;
*send_seq += 1;
socket.send_to(&packet, addr).await?;
let _ = send_terminal_udp(socket, &packet, addr).await?;
Ok(())
}
@@ -6750,7 +6940,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(())
}
@@ -7088,18 +7278,19 @@ mod tests {
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,
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,
queue_pending_user_input, raw_contains_host_table, recv_response_until, refresh_live_addr,
release_tag_download_url, release_tag_from_effective_url, release_version_from_tag,
render_status_clear, 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,
is_local_status_target, is_resume_response_for_client, is_transient_udp_os_error,
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, queue_pending_user_input, raw_contains_host_table,
recv_response_until, refresh_live_addr, release_tag_download_url,
release_tag_from_effective_url, release_version_from_tag, render_status_clear,
render_status_overlay, requested_env, resolved_startup_command, retransmit_stream_opens,
rewrite_forward_command, selected_predict_mode, selected_udp_host, server_version_mismatch,
should_flush_terminal_input_after_contact, 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,
};
use dosh::config::{ClientConfig, CommandExtension, HostConfig};
use dosh::native::EnvVar;
@@ -7879,6 +8070,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();
@@ -8301,6 +8520,25 @@ 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_os_error(libc::ENETUNREACH));
assert!(is_transient_udp_os_error(libc::EHOSTUNREACH));
assert!(is_transient_udp_os_error(libc::EADDRNOTAVAIL));
assert!(!is_transient_udp_os_error(libc::EINVAL));
}
#[cfg(windows)]
{
assert!(is_transient_udp_os_error(10051)); // WSAENETUNREACH
assert!(is_transient_udp_os_error(10065)); // WSAEHOSTUNREACH
assert!(is_transient_udp_os_error(10049)); // WSAEADDRNOTAVAIL
assert!(!is_transient_udp_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";
@@ -8314,6 +8552,18 @@ 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 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 non_mouse_escape_input_survives_stale_filter() {
assert_eq!(strip_stale_mouse_reports(b"\x1b[A"), b"\x1b[A");
+91 -16
View File
@@ -980,7 +980,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(())
}
@@ -1058,7 +1058,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(());
}
@@ -1251,7 +1251,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(())
}
@@ -1359,7 +1359,7 @@ async fn handle_bootstrap_attach(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1494,7 +1494,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(())
}
@@ -1512,7 +1512,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(())
}
@@ -1578,7 +1578,7 @@ async fn handle_resume(
SERVER_TO_CLIENT,
&body,
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1690,7 +1690,7 @@ async fn handle_ping(
SERVER_TO_CLIENT,
b"",
)?;
socket.send_to(&out, peer).await?;
let _ = send_udp(socket, &out, peer).await?;
Ok(())
}
@@ -1705,6 +1705,60 @@ 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}")),
}
}
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
}
async fn handle_ack(
state: &Arc<Mutex<ServerState>>,
peer: SocketAddr,
@@ -3020,7 +3074,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(())
}
@@ -3114,7 +3168,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(())
}
@@ -3189,7 +3243,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?;
}
}
@@ -3333,7 +3387,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(())
}
@@ -3435,7 +3489,7 @@ async fn broadcast_output(
}
}
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3488,7 +3542,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(())
}
@@ -3595,7 +3649,7 @@ async fn retransmit_pending(
sends
};
for (endpoint, packet) in sends {
socket.send_to(&packet, endpoint).await?;
let _ = send_udp(socket, &packet, endpoint).await?;
}
Ok(())
}
@@ -3691,7 +3745,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(())
}
@@ -4536,4 +4590,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"
);
}
}
}
+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();
+4
View File
@@ -25,6 +25,10 @@ fn systemd_service_does_not_sandbox_remote_shells() {
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=",