diff --git a/install.ps1 b/install.ps1 index 708a909..f03dab3 100644 --- a/install.ps1 +++ b/install.ps1 @@ -188,6 +188,7 @@ predict = true predict_mode = "experimental" cache_attach_tickets = true credential_cache = "~/.local/share/dosh/credentials" +escape_key = "^]" "@ | Set-Content -NoNewline -Encoding utf8 $clientConfig } diff --git a/install.sh b/install.sh index 43cc8fc..c083756 100755 --- a/install.sh +++ b/install.sh @@ -416,8 +416,9 @@ scrollback = 5000 auth_ttl_secs = 30 attach_ticket_ttl_secs = 3600 allow_attach_tickets = true -client_timeout_secs = 86400 +client_timeout_secs = 2592000 retransmit_window = 256 +output_frame_interval_ms = 16 default_input_mode = "read-write" prewarm_sessions = ["default"] create_on_attach = true @@ -492,6 +493,7 @@ known_hosts = "~/.config/dosh/known_hosts" identity_files = ["~/.ssh/id_ed25519"] use_ssh_agent = true forward_agent = false +escape_key = "^]" EOF fi diff --git a/scripts/tui-harness.sh b/scripts/tui-harness.sh index d06e1a9..c4be84f 100755 --- a/scripts/tui-harness.sh +++ b/scripts/tui-harness.sh @@ -7,6 +7,7 @@ cd "$repo_root" for test_name in \ live_output_forwards_terminal_control_sequences \ tui_control_sequences_survive_transport_verbatim \ + unicode_output_survives_transport \ large_tui_paint_is_delivered_in_mtu_safe_frames \ resume_snapshot_preserves_alternate_screen_mode do diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 777e9e1..efa5fee 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -110,6 +110,14 @@ struct Args { #[arg(long)] predict: bool, #[arg(long)] + predict_mode: Option, + #[arg(short = 'a', long = "predict-always")] + predict_always: bool, + #[arg(short = 'n', long = "predict-never")] + predict_never: bool, + #[arg(long)] + escape_key: Option, + #[arg(long)] ssh_port: Option, #[arg(long)] ssh_auth_command: Option, @@ -253,6 +261,8 @@ async fn main() -> Result<()> { || !dynamic_forwards.is_empty() || forward_agent; let predict = args.predict || host.predict.unwrap_or(config.predict); + let predict_mode = selected_predict_mode(&args, &config)?; + let escape_key = selected_escape_key(&args, &config)?; let session = select_session(args.session.as_deref(), args.new); let mode = if args.forward_only { "forward-only" @@ -302,22 +312,17 @@ async fn main() -> Result<()> { let started = Instant::now(); let mut resolved_ssh_config = None; - let target_udp_host = if let Some(host) = args + let requested_udp_host = args .dosh_host .clone() .or_else(|| host.dosh_host.clone()) - .or_else(|| config.dosh_host.clone()) - { - host - } else if args.local_auth { - "127.0.0.1".to_string() + .or_else(|| config.dosh_host.clone()); + let target_udp_host = if args.local_auth { + requested_udp_host.unwrap_or_else(|| "127.0.0.1".to_string()) } else { match ssh_config(&server, ssh_port) { Ok(parsed) => { - let hostname = parsed - .hostname - .clone() - .unwrap_or_else(|| ssh_destination_host(&server)); + let hostname = selected_udp_host(requested_udp_host.as_deref(), &server, &parsed)?; resolved_ssh_config = Some(parsed); hostname } @@ -328,7 +333,11 @@ async fn main() -> Result<()> { &format!("dosh could not resolve SSH config hostname, using {server}: {err:#}"), ); resolved_ssh_config = Some(SshConfig::default()); - ssh_destination_host(&server) + selected_udp_host( + requested_udp_host.as_deref(), + &server, + &SshConfig::default(), + )? } } }; @@ -362,6 +371,8 @@ async fn main() -> Result<()> { cred, Some(frame), predict, + predict_mode, + escape_key, startup_command, config.reconnect_timeout_secs, Vec::new(), @@ -398,6 +409,8 @@ async fn main() -> Result<()> { cred, Some(frame), predict, + predict_mode, + escape_key, startup_command, config.reconnect_timeout_secs, Vec::new(), @@ -464,6 +477,8 @@ async fn main() -> Result<()> { cred, Some(frame), predict, + predict_mode, + escape_key, startup_command, config.reconnect_timeout_secs, local_forwards, @@ -540,6 +555,8 @@ async fn main() -> Result<()> { cred, Some(first), predict, + predict_mode, + escape_key, startup_command, config.reconnect_timeout_secs, Vec::new(), @@ -779,17 +796,19 @@ async fn run_doctor_for_host( Err(err) => println!("[warn] ssh fallback: {err:#}"), } - let udp_host = args + let requested_udp_host = args .dosh_host .clone() .or_else(|| host_config.dosh_host.clone()) - .or_else(|| config.dosh_host.clone()) - .unwrap_or_else(|| { - parsed_ssh_config - .hostname - .clone() - .unwrap_or_else(|| ssh_destination_host(&server)) - }); + .or_else(|| config.dosh_host.clone()); + let udp_host = + match selected_udp_host(requested_udp_host.as_deref(), &server, &parsed_ssh_config) { + Ok(host) => host, + Err(err) => { + println!("[fail] native endpoint: {err:#}"); + ssh_destination_host(&server) + } + }; println!("[info] native endpoint: {udp_host}:{dosh_port}"); match resolve_addr(&udp_host, dosh_port) { Ok(addr) => println!("[ok] udp resolve: {addr}"), @@ -1717,6 +1736,30 @@ fn ssh_destination_host(server: &str) -> String { .to_string() } +fn selected_udp_host( + explicit: Option<&str>, + server: &str, + ssh_config: &SshConfig, +) -> Result { + let ssh_host = || { + ssh_config + .hostname + .clone() + .unwrap_or_else(|| ssh_destination_host(server)) + }; + let Some(raw) = explicit.map(str::trim).filter(|value| !value.is_empty()) else { + return Ok(ssh_host()); + }; + match raw.to_ascii_lowercase().as_str() { + "ssh" | "auto" => Ok(ssh_host()), + "localhost" => Ok("127.0.0.1".to_string()), + "any" => Err(anyhow!( + "dosh_host={raw:?} is a server bind policy, not a client destination; use ssh/auto or a host/IP" + )), + _ => Ok(raw.to_string()), + } +} + fn ssh_username(server: &str) -> Option { let (user, _) = server.rsplit_once('@')?; if user.is_empty() { @@ -2356,7 +2399,13 @@ async fn bootstrap_attach( }; Ok(protocol::from_body::(&plain).ok()) }) - .await?; + .await + .with_context(|| { + format!( + "no Dosh bootstrap response from {}:{}; check UDP reachability and server firewall", + bootstrap.udp_host, bootstrap.udp_port + ) + })?; let cred = CachedCredential { server: server_name.to_string(), session: ok.session.clone(), @@ -2443,7 +2492,13 @@ async fn try_ticket_attach( }; Ok(protocol::from_body::(&plain).ok()) }) - .await?; + .await + .with_context(|| { + format!( + "cached ticket attach to {}:{} timed out; check UDP reachability or run `dosh recover {}`", + cached.udp_host, cached.udp_port, cached.server + ) + })?; let frame = Frame { session: ok.session.clone(), output_seq: ok.initial_seq, @@ -2500,7 +2555,13 @@ async fn try_resume_fresh_process( }; Ok(protocol::from_body::(&plain).ok()) }) - .await?; + .await + .with_context(|| { + format!( + "cached resume to {}:{} timed out; check UDP reachability or run `dosh recover {}`", + cached.udp_host, cached.udp_port, cached.server + ) + })?; let mut next = cached.clone(); next.last_rendered_seq = frame.output_seq; Ok((frame, next)) @@ -2549,7 +2610,13 @@ async fn try_live_resume( }; Ok(protocol::from_body::(&plain).ok()) }) - .await?; + .await + .with_context(|| { + format!( + "live resume to {}:{} timed out; still waiting for the server to become reachable", + cached.udp_host, cached.udp_port + ) + })?; let mut next = cached.clone(); next.last_rendered_seq = frame.output_seq; Ok((frame, next)) @@ -2584,6 +2651,8 @@ async fn run_terminal( mut cred: CachedCredential, first_frame: Option, predict: bool, + predict_mode: PredictMode, + escape_key: Option>, startup_command: Option>, reconnect_timeout_secs: u64, local_forwards: Vec, @@ -2618,7 +2687,6 @@ async fn run_terminal( // env var wins for ad-hoc tuning; otherwise the client config's // `predict_mode` provides the persistent default. Predictions only run in a // read-write interactive session. - let predict_mode = resolve_predict_mode(); let mut predictor = Predictor::with_mode( predict && cred.mode != "view-only" && !forward_only, predict_mode, @@ -2703,7 +2771,7 @@ async fn run_terminal( stdin_msg = stdin_rx.recv() => { match stdin_msg { Some(bytes) => { - if bytes == [0x1d] { + if input_matches_escape(&bytes, escape_key.as_deref()) { break; } if cred.mode != "view-only" && cred.mode != "forward-only" { @@ -4444,17 +4512,81 @@ fn emit_overlay(_bytes: &[u8]) -> Result<()> { Ok(()) } -/// Resolve the prediction display policy. `DOSH_PREDICT_MODE` (off / experimental -/// / always) overrides for quick experiments; otherwise the client config's -/// `predict_mode` is used, defaulting to the adaptive (experimental) policy. -fn resolve_predict_mode() -> PredictMode { +/// Resolve the prediction display policy. `DOSH_PREDICT_MODE` (off / +/// experimental / always) overrides for quick experiments; otherwise the +/// already-selected CLI/config policy is used. +fn selected_predict_mode(args: &Args, config: &dosh::config::ClientConfig) -> Result { + if args.predict_always && args.predict_never { + return Err(anyhow!("--predict-always and --predict-never conflict")); + } + if args.predict_always { + return Ok(PredictMode::Always); + } + if args.predict_never { + return Ok(PredictMode::Off); + } + if let Some(mode) = args.predict_mode.as_deref() { + return Ok(PredictMode::parse(mode)); + } if let Ok(env) = std::env::var("DOSH_PREDICT_MODE") { - return PredictMode::parse(&env); + return Ok(PredictMode::parse(&env)); } - match load_client_config(None) { - Ok(cfg) => PredictMode::parse(&cfg.predict_mode), - Err(_) => PredictMode::Experimental, + Ok(PredictMode::parse(&config.predict_mode)) +} + +fn selected_escape_key( + args: &Args, + config: &dosh::config::ClientConfig, +) -> Result>> { + if let Some(raw) = args.escape_key.as_deref() { + return parse_escape_key(raw); } + if let Ok(raw) = std::env::var("DOSH_ESCAPE_KEY") { + return parse_escape_key(&raw); + } + parse_escape_key(&config.escape_key) +} + +fn parse_escape_key(raw: &str) -> Result>> { + let value = raw.trim(); + if value.eq_ignore_ascii_case("none") + || value.eq_ignore_ascii_case("off") + || value.eq_ignore_ascii_case("disabled") + { + return Ok(None); + } + if let Some(rest) = value.strip_prefix("^") { + anyhow::ensure!( + rest.chars().count() == 1, + "invalid control escape key {raw:?}" + ); + let ch = rest.as_bytes()[0]; + let byte = match ch { + b'?' => 0x7f, + b'@'..=b'_' => ch - b'@', + b'a'..=b'z' => ch - b'a' + 1, + _ => return Err(anyhow!("invalid control escape key {raw:?}")), + }; + return Ok(Some(vec![byte])); + } + if let Some(hex) = value.strip_prefix("0x") { + let byte = u8::from_str_radix(hex, 16) + .with_context(|| format!("invalid hex escape key {raw:?}"))?; + return Ok(Some(vec![byte])); + } + if value.len() == 1 { + return Ok(Some(value.as_bytes().to_vec())); + } + Err(anyhow!( + "escape key must be none, ^X, 0xNN, or one ASCII character" + )) +} + +fn input_matches_escape(bytes: &[u8], escape_key: Option<&[u8]>) -> bool { + let Some(escape_key) = escape_key else { + return false; + }; + !escape_key.is_empty() && bytes.windows(escape_key.len()).any(|w| w == escape_key) } /// Resolve whether the non-destructive disconnect status line is enabled. @@ -4870,10 +5002,11 @@ mod tests { DisconnectStatus, DynamicForward, FrameBuffer, LocalForward, MAX_PENDING_USER_INPUT_BYTES, PredictMode, Predictor, RemoteForward, SshConfig, StatusAction, auth_allows, cache_key, cache_server_prefix, clear_cached_credentials, ensure_tui_safe_status_overlay, - latest_release_download_url, load_first_native_identity_with_prompt, parse_dynamic_forward, - parse_local_forward, parse_remote_forward, parse_ssh_config, queue_pending_user_input, - raw_contains_host_table, recv_response_until, render_status_clear, render_status_overlay, - requested_env, resolved_startup_command, rewrite_forward_command, ssh_destination_host, + input_matches_escape, 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, queue_pending_user_input, raw_contains_host_table, recv_response_until, + render_status_clear, render_status_overlay, requested_env, resolved_startup_command, + rewrite_forward_command, selected_predict_mode, selected_udp_host, ssh_destination_host, ssh_username, ssh_with_user, startup_command, toml_bare_key_or_quoted, update_check_requested, valid_forward_host, }; @@ -4905,6 +5038,10 @@ mod tests { forward_only: false, background: false, predict: false, + predict_mode: None, + predict_always: false, + predict_never: false, + escape_key: None, ssh_port: None, ssh_auth_command: None, ssh_key: None, @@ -5046,6 +5183,31 @@ mod tests { assert_eq!(ssh_destination_host("alice@example.com"), "example.com"); } + #[test] + fn selected_udp_host_supports_ssh_auto_and_localhost() { + let ssh = SshConfig { + hostname: Some("server.lan".to_string()), + ..SshConfig::default() + }; + assert_eq!( + selected_udp_host(None, "alias", &ssh).unwrap(), + "server.lan" + ); + assert_eq!( + selected_udp_host(Some("ssh"), "alias", &ssh).unwrap(), + "server.lan" + ); + assert_eq!( + selected_udp_host(Some("auto"), "alias", &ssh).unwrap(), + "server.lan" + ); + assert_eq!( + selected_udp_host(Some("localhost"), "alias", &ssh).unwrap(), + "127.0.0.1" + ); + assert!(selected_udp_host(Some("any"), "alias", &ssh).is_err()); + } + #[test] fn parses_user_from_ssh_destination() { assert_eq!(ssh_username("alice@example.com"), Some("alice".to_string())); @@ -5269,6 +5431,8 @@ mod tests { assert!(!Predictor::predictable(b"\n")); assert!(!Predictor::predictable(b"\x1b[A")); assert!(!Predictor::predictable("snowman: \u{2603}".as_bytes())); + assert!(!Predictor::predictable("wide: 你".as_bytes())); + assert!(!Predictor::predictable("combine: e\u{301}".as_bytes())); } #[test] @@ -5446,6 +5610,42 @@ mod tests { assert_eq!(PredictMode::parse("anything"), PredictMode::Experimental); } + #[test] + fn cli_prediction_flags_override_config() { + let mut args = test_args("host", &[]); + let mut config = ClientConfig { + predict_mode: "experimental".to_string(), + ..ClientConfig::default() + }; + + args.predict_always = true; + assert_eq!( + selected_predict_mode(&args, &config).unwrap(), + PredictMode::Always + ); + + args.predict_always = false; + args.predict_never = true; + assert_eq!( + selected_predict_mode(&args, &config).unwrap(), + PredictMode::Off + ); + + args.predict_never = false; + args.predict_mode = Some("always".to_string()); + assert_eq!( + selected_predict_mode(&args, &config).unwrap(), + PredictMode::Always + ); + + args.predict_mode = None; + config.predict_mode = "off".to_string(); + assert_eq!( + selected_predict_mode(&args, &config).unwrap(), + PredictMode::Off + ); + } + #[test] fn startup_command_joins_args_for_remote_shell() { assert_eq!(startup_command(&[]), None); @@ -5699,6 +5899,24 @@ mod tests { assert_eq!(pending_bytes, MAX_PENDING_USER_INPUT_BYTES); } + #[test] + fn escape_key_parser_accepts_mosh_style_controls() { + assert_eq!(parse_escape_key("^]").unwrap(), Some(vec![0x1d])); + assert_eq!(parse_escape_key("^C").unwrap(), Some(vec![0x03])); + assert_eq!(parse_escape_key("^?").unwrap(), Some(vec![0x7f])); + assert_eq!(parse_escape_key("0x1d").unwrap(), Some(vec![0x1d])); + assert_eq!(parse_escape_key("~").unwrap(), Some(vec![b'~'])); + assert_eq!(parse_escape_key("none").unwrap(), None); + assert!(parse_escape_key("too-long").is_err()); + } + + #[test] + fn escape_key_matches_inside_stdin_chunk() { + assert!(input_matches_escape(b"abc\x1d", Some(&[0x1d]))); + assert!(!input_matches_escape(b"abc", Some(&[0x1d]))); + assert!(!input_matches_escape(b"abc\x1d", None)); + } + // --- Item 1: disconnect status line state machine --- /// The status line stays hidden while the link is fresh and only appears once diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index d4c3eb6..7d44a7e 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -203,12 +203,15 @@ async fn serve(config_path: Option) -> Result<()> { let retransmit_state = Arc::clone(&state); let retransmit_socket = Arc::clone(&socket); tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_millis(100)); + let mut interval = tokio::time::interval(Duration::from_millis(20)); loop { interval.tick().await; if let Err(err) = retransmit_pending(&retransmit_state, &retransmit_socket).await { eprintln!("retransmit error: {err:#}"); } + if let Err(err) = flush_paced_snapshots(&retransmit_state, &retransmit_socket).await { + eprintln!("paced snapshot error: {err:#}"); + } if let Err(err) = maybe_rekey_clients(&retransmit_state, &retransmit_socket).await { eprintln!("rekey error: {err:#}"); } @@ -370,6 +373,8 @@ struct ClientState { rows: u16, last_seen: Instant, pending: VecDeque, + last_frame_sent: Instant, + paced_snapshot_due: bool, allowed_forwardings: Vec, stream_writers: HashMap>>, opened_streams: HashSet, @@ -1122,6 +1127,8 @@ 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(), @@ -1271,6 +1278,8 @@ 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(), @@ -1394,6 +1403,8 @@ 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(), @@ -2630,6 +2641,8 @@ async fn broadcast_output( 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 session = locked .sessions .get_mut(&output.session) @@ -2664,6 +2677,15 @@ 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); @@ -2702,6 +2724,8 @@ async fn broadcast_output( last_sent: Instant::now(), attempts: 0, }); + client.last_frame_sent = now; + client.paced_snapshot_due = false; sends.push((client.endpoint, packet)); } sends @@ -2852,6 +2876,76 @@ async fn retransmit_pending( Ok(()) } +async fn flush_paced_snapshots( + state: &Arc>, + socket: &Arc, +) -> 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?; + } + Ok(()) +} + /// How long the previous epoch's key is retained after a rekey so in-flight /// pre-rekey packets still decrypt instead of being dropped as fatal. const REKEY_PREVIOUS_KEY_GRACE_SECS: u64 = 5; @@ -3185,6 +3279,11 @@ mod tests { assert!(limiter.buckets.is_empty()); } + #[test] + fn default_client_timeout_supports_long_sleep() { + assert_eq!(ServerConfig::default().client_timeout_secs, 2_592_000); + } + fn test_client_state(session_key: [u8; 32]) -> ClientState { ClientState { endpoint: "127.0.0.1:9".parse().unwrap(), @@ -3197,6 +3296,8 @@ 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(), @@ -3397,6 +3498,8 @@ 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(), @@ -3490,6 +3593,8 @@ 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]), @@ -3557,6 +3662,89 @@ mod tests { assert_eq!(frame.bytes, b"terminal-priority"); } + #[tokio::test] + async fn output_pacing_coalesces_fast_frames_into_snapshot() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let config = ServerConfig { + output_frame_interval_ms: 1000, + ..ServerConfig::default() + }; + let mut state = ServerState::new(config, [0u8; 32], pty_tx); + state + .ensure_session("test", 80, 24, "forward-only", &[]) + .unwrap(); + let client_id = [42u8; 16]; + let session_key = [7u8; 32]; + let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let sender = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap()); + let mut client = test_client_state(session_key); + client.endpoint = receiver.local_addr().unwrap(); + state.insert_client("test", client_id, client); + let state = Arc::new(Mutex::new(state)); + + broadcast_output( + &state, + &sender, + PtyOutput { + session: "test".to_string(), + bytes: b"first".to_vec(), + exited: false, + }, + ) + .await + .unwrap(); + broadcast_output( + &state, + &sender, + PtyOutput { + session: "test".to_string(), + bytes: b"second".to_vec(), + exited: false, + }, + ) + .await + .unwrap(); + + let mut buf = [0u8; 4096]; + let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf)) + .await + .unwrap() + .unwrap(); + 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_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() + .unwrap(); + 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_eq!(frame.output_seq, 2); + } + #[test] fn remote_non_loopback_bind_requires_explicit_config() { let config = ServerConfig::default(); diff --git a/src/config.rs b/src/config.rs index 3eaa75f..c32f421 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,6 +14,8 @@ pub struct ServerConfig { pub allow_attach_tickets: bool, pub client_timeout_secs: u64, pub retransmit_window: usize, + #[serde(default = "default_output_frame_interval_ms")] + pub output_frame_interval_ms: u64, pub default_input_mode: String, pub prewarm_sessions: Vec, pub create_on_attach: bool, @@ -68,8 +70,9 @@ impl Default for ServerConfig { auth_ttl_secs: 30, attach_ticket_ttl_secs: 3600, allow_attach_tickets: true, - client_timeout_secs: 86400, + client_timeout_secs: 2_592_000, retransmit_window: 256, + output_frame_interval_ms: default_output_frame_interval_ms(), default_input_mode: "read-write".to_string(), prewarm_sessions: vec!["default".to_string()], create_on_attach: true, @@ -132,6 +135,8 @@ pub struct ClientConfig { /// full-screen TUI, and cleared the instant packets resume. Default on. #[serde(default = "default_true")] pub disconnect_status: bool, + #[serde(default = "default_escape_key")] + pub escape_key: String, #[serde(default = "default_send_env")] pub send_env: Vec, #[serde(default)] @@ -205,6 +210,7 @@ impl Default for ClientConfig { use_ssh_agent: true, forward_agent: false, disconnect_status: true, + escape_key: default_escape_key(), send_env: default_send_env(), set_env: HashMap::new(), extensions: HashMap::new(), @@ -231,6 +237,10 @@ fn default_native_auth_rate_limit_per_minute() -> u32 { 30 } +fn default_output_frame_interval_ms() -> u64 { + 16 +} + fn default_rekey_after_packets() -> u64 { // Rotate well before any AEAD nonce-reuse concern; ChaCha20-Poly1305 with a // per-direction monotonic seq is safe far beyond this, but a bounded epoch @@ -259,6 +269,10 @@ fn default_known_hosts() -> String { "~/.config/dosh/known_hosts".to_string() } +fn default_escape_key() -> String { + "^]".to_string() +} + fn default_identity_files() -> Vec { vec!["~/.ssh/id_ed25519".to_string()] } diff --git a/tests/integration_smoke.rs b/tests/integration_smoke.rs index dd9ef02..673de67 100644 --- a/tests/integration_smoke.rs +++ b/tests/integration_smoke.rs @@ -1330,13 +1330,16 @@ fn tui_control_sequences_survive_transport_verbatim() { let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let sequences = concat!( - "\x1b[?1049h", // alternate screen on - "\x1b[?2004h", // bracketed paste on - "\x1b[?1000h", // mouse tracking on - "\x1b[?1006h", // SGR mouse encoding on - "\x1b[?25l", // cursor hidden - "\x1b[12;34H", // absolute cursor movement + "\x1b[?1049h", // alternate screen on + "\x1b[?2004h", // bracketed paste on + "\x1b[?1000h", // mouse tracking on + "\x1b[?1006h", // SGR mouse encoding on + "\x1b]0;DOSH_TITLE\x07", // OSC title + "\x1b[?25l", // cursor hidden + "\x1b[12;34H", // absolute cursor movement + "\x1b[1;31m", // bold red "DOSH_TUI_VERBATIM", + "\x1b[0m", "\x1b[?25h", "\x1b[?1006l", "\x1b[?1000l", @@ -1366,9 +1369,12 @@ fn tui_control_sequences_survive_transport_verbatim() { "\x1b[?2004h", "\x1b[?1000h", "\x1b[?1006h", + "\x1b]0;DOSH_TITLE\x07", "\x1b[?25l", "\x1b[12;34H", + "\x1b[1;31m", "DOSH_TUI_VERBATIM", + "\x1b[0m", "\x1b[?25h", "\x1b[?1006l", "\x1b[?1000l", @@ -1382,6 +1388,39 @@ fn tui_control_sequences_survive_transport_verbatim() { } } +#[test] +fn unicode_output_survives_transport() { + 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 (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); + + let marker = "DOSH_UNICODE 你 e\u{301} 🚀"; + let input = Input { + bytes: format!("printf '{}\\n'\n", marker).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, 2000); + + let _ = server.kill(); + let _ = server.wait(); + + assert!( + text.contains(marker), + "expected UTF-8 output marker to survive transport, got {text:?}" + ); +} + #[test] fn large_tui_paint_is_delivered_in_mtu_safe_frames() { let dir = tempfile::tempdir().unwrap();