diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 6f70560..f9de696 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -555,7 +555,10 @@ 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 last_size = size().unwrap_or((80, 24)); + let mut remote_terminal = vt100::Parser::new(last_size.1.max(1), last_size.0.max(1), 0); + let mut input_translator = UserInputTranslator::default(); if let Some(frame) = first_frame { + remote_terminal.process(&frame.bytes); render_frame(&frame)?; if frame.closed { return Ok(()); @@ -584,23 +587,32 @@ async fn run_terminal( let mut recv_buf = vec![0u8; 65535]; loop { tokio::select! { - Some(bytes) = stdin_rx.recv() => { - if bytes == [0x1d] { - break; - } - if cred.mode != "view-only" { - let body = protocol::to_body(&Input { bytes })?; - let packet = protocol::encode_encrypted( - PacketKind::Input, - cred.client_id, - send_seq, - cred.last_rendered_seq, - &cred.session_key, - CLIENT_TO_SERVER, - &body, - )?; - send_seq += 1; - socket.send_to(&packet, addr).await?; + stdin_msg = stdin_rx.recv() => { + match stdin_msg { + Some(bytes) => { + if bytes == [0x1d] { + break; + } + if cred.mode != "view-only" { + let bytes = input_translator.translate( + &bytes, + remote_terminal.screen().application_cursor(), + ); + let body = protocol::to_body(&Input { bytes })?; + let packet = protocol::encode_encrypted( + PacketKind::Input, + cred.client_id, + send_seq, + cred.last_rendered_seq, + &cred.session_key, + CLIENT_TO_SERVER, + &body, + )?; + send_seq += 1; + socket.send_to(&packet, addr).await?; + } + } + None => break, } } _ = resize_tick.tick() => { @@ -608,6 +620,7 @@ async fn run_terminal( && (cols, rows) != last_size { last_size = (cols, rows); + remote_terminal.set_size(rows.max(1), cols.max(1)); if cred.mode != "view-only" { send_resize(&socket, addr, &cred, &mut send_seq, cols, rows).await?; } @@ -625,6 +638,7 @@ async fn run_terminal( PacketKind::Frame | PacketKind::ResumeOk => { let plain = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT)?; let frame: Frame = protocol::from_body(&plain)?; + remote_terminal.process(&frame.bytes); render_frame(&frame)?; cred.last_rendered_seq = frame.output_seq; send_ack(&socket, addr, &cred, &mut send_seq).await?; @@ -664,9 +678,58 @@ async fn run_terminal( } } } + let _ = detach_once(&socket, &cred, send_seq).await; Ok(()) } +#[derive(Default)] +struct UserInputTranslator { + state: UserInputState, +} + +#[derive(Default)] +enum UserInputState { + #[default] + Ground, + Esc, + Ss3, +} + +impl UserInputTranslator { + fn translate(&mut self, bytes: &[u8], application_cursor: bool) -> Vec { + let mut out = Vec::with_capacity(bytes.len()); + for &byte in bytes { + match self.state { + UserInputState::Ground => { + if byte == 0x1b { + self.state = UserInputState::Esc; + } + out.push(byte); + } + UserInputState::Esc => { + if byte == b'O' { + self.state = UserInputState::Ss3; + } else { + self.state = UserInputState::Ground; + out.push(byte); + } + } + UserInputState::Ss3 => { + self.state = UserInputState::Ground; + if !application_cursor && matches!(byte, b'A'..=b'D') { + out.push(b'['); + out.push(byte); + } else { + out.push(b'O'); + out.push(byte); + } + } + } + } + out + } +} + async fn send_resize( socket: &UdpSocket, addr: SocketAddr, @@ -824,3 +887,31 @@ const TERMINAL_CLEANUP: &[u8] = concat!( "\x1b[?1049l" ) .as_bytes(); + +#[cfg(test)] +mod tests { + use super::UserInputTranslator; + + #[test] + fn translates_application_cursor_arrows_for_normal_remote_mode() { + let mut translator = UserInputTranslator::default(); + assert_eq!(translator.translate(b"\x1bOA", false), b"\x1b[A"); + assert_eq!(translator.translate(b"\x1bOB", false), b"\x1b[B"); + assert_eq!(translator.translate(b"\x1bOC", false), b"\x1b[C"); + assert_eq!(translator.translate(b"\x1bOD", false), b"\x1b[D"); + } + + #[test] + fn preserves_application_cursor_arrows_for_remote_application_mode() { + let mut translator = UserInputTranslator::default(); + assert_eq!(translator.translate(b"\x1bOA", true), b"\x1bOA"); + } + + #[test] + fn preserves_split_escape_state_between_reads() { + let mut translator = UserInputTranslator::default(); + assert_eq!(translator.translate(b"\x1b", false), b"\x1b"); + assert_eq!(translator.translate(b"O", false), b""); + assert_eq!(translator.translate(b"A", false), b"[A"); + } +} diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 9ba365f..2b7b8e5 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -125,6 +125,15 @@ async fn serve(config_path: Option) -> Result<()> { } }); + let cleanup_state = Arc::clone(&state); + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(5)); + loop { + interval.tick().await; + cleanup_disconnected_clients(&cleanup_state); + } + }); + let mut buf = vec![0u8; 65535]; loop { let (n, peer) = socket.recv_from(&mut buf).await?; @@ -781,6 +790,17 @@ async fn retransmit_pending( Ok(()) } +fn cleanup_disconnected_clients(state: &Arc>) { + let mut locked = state.lock().expect("server state poisoned"); + let timeout = Duration::from_secs(locked.config.client_timeout_secs.max(1)); + let now = Instant::now(); + for session in locked.sessions.values_mut() { + session + .clients + .retain(|_, client| now.duration_since(client.last_seen) <= timeout); + } +} + fn find_client_key( state: &Arc>, client_id: &[u8; 16], diff --git a/tests/integration_smoke.rs b/tests/integration_smoke.rs index 3f1ada6..ebe4207 100644 --- a/tests/integration_smoke.rs +++ b/tests/integration_smoke.rs @@ -18,6 +18,14 @@ fn free_udp_port() -> u16 { } fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf { + write_server_config_with_timeout(dir, port, 30) +} + +fn write_server_config_with_timeout( + dir: &tempfile::TempDir, + port: u16, + client_timeout_secs: u64, +) -> std::path::PathBuf { let config_dir = dir.path().join(".config/dosh"); fs::create_dir_all(&config_dir).unwrap(); let config = config_dir.join("server.toml"); @@ -31,7 +39,7 @@ scrollback = 5000 auth_ttl_secs = 30 attach_ticket_ttl_secs = 3600 allow_attach_tickets = true -client_timeout_secs = 30 +client_timeout_secs = {client_timeout_secs} retransmit_window = 256 default_input_mode = "read-write" prewarm_sessions = ["default"] @@ -215,6 +223,37 @@ fn unknown_live_client_gets_reject() { assert_eq!(reject.reason, "unknown client"); } +#[test] +fn disconnected_client_times_out_and_gets_reject() { + let dir = tempfile::tempdir().unwrap(); + let port = free_udp_port(); + let config = write_server_config_with_timeout(&dir, port, 1); + let mut server = start_server(&dir, &config); + let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); + thread::sleep(Duration::from_secs(6)); + send_encrypted( + &socket, + port, + PacketKind::Ping, + ok.client_id, + 2, + 0, + &bootstrap.session_key, + b"", + ); + let mut buf = [0u8; 65535]; + let (n, _) = socket.recv_from(&mut buf).unwrap(); + let packet = protocol::decode(&buf[..n]).unwrap(); + let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); + + let _ = server.kill(); + let _ = server.wait(); + + assert_eq!(packet.header.kind, PacketKind::AttachReject); + assert_eq!(packet.header.conn_id, ok.client_id); + assert_eq!(reject.reason, "unknown client"); +} + #[test] fn local_attach_only_smoke() { let dir = tempfile::tempdir().unwrap();