diff --git a/tests/client_terminal_runtime.rs b/tests/client_terminal_runtime.rs index a5ee7a1..c223faa 100644 --- a/tests/client_terminal_runtime.rs +++ b/tests/client_terminal_runtime.rs @@ -50,6 +50,7 @@ struct CachedCredentialWire { enum ServerObservation { BulkSent, Input(Vec), + Reconnected, Resize(u16, u16), } @@ -224,12 +225,87 @@ fn terminal_output_backpressure_does_not_block_input_transport() { assert!(status.success(), "Dosh client exited with {status:?}"); } +#[test] +fn idle_reconnect_restores_snapshot_and_orders_reordered_frames() { + const PROBE: &[u8] = b"DOSH_RECONNECT_INPUT\r"; + + let dir = tempfile::tempdir().unwrap(); + let home = dir.path().join("home"); + let cache = dir.path().join("credentials"); + fs::create_dir_all(home.join(".config/dosh")).unwrap(); + fs::create_dir_all(&cache).unwrap(); + + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(8))) + .unwrap(); + let port = socket.local_addr().unwrap().port(); + write_client_fixture(&home, &cache, port); + let (observation_tx, observation_rx) = mpsc::channel(); + let server = thread::spawn(move || run_reconnect_server(socket, observation_tx, PROBE)); + + let pty = NativePtySystem::default(); + let pair = pty + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + let mut reader = pair.master.try_clone_reader().unwrap(); + let mut writer = pair.master.take_writer().unwrap(); + let mut command = client_command(dir.path(), port); + command.env("HOME", home.to_string_lossy().to_string()); + command.env("USERPROFILE", home.to_string_lossy().to_string()); + command.env("APPDATA", home.to_string_lossy().to_string()); + command.env("LOCALAPPDATA", home.to_string_lossy().to_string()); + command.env("TERM", "xterm-256color"); + let mut child = pair.slave.spawn_command(command).unwrap(); + drop(pair.slave); + + let output = Arc::new(Mutex::new(Vec::new())); + let reader_output = Arc::clone(&output); + let reader_thread = thread::spawn(move || { + let mut buf = [0u8; 4096]; + loop { + match reader.read(&mut buf) { + Ok(0) | Err(_) => break, + Ok(n) => reader_output.lock().unwrap().extend_from_slice(&buf[..n]), + } + } + }); + + wait_for_output(&output, b"DOSH_RECONNECT_READY", Duration::from_secs(3)); + wait_for_observation(&observation_rx, Duration::from_secs(4), |observation| { + matches!(observation, ServerObservation::Reconnected) + }); + wait_for_output(&output, b"DOSH_RECONNECT_SNAPSHOT", Duration::from_secs(3)); + + writer.write_all(PROBE).unwrap(); + writer.flush().unwrap(); + wait_for_input(&observation_rx, PROBE, Duration::from_secs(2)); + wait_for_output( + &output, + b"DOSH_ORDER_FIRST_DOSH_ORDER_SECOND_DOSH_RECONNECT_DONE", + Duration::from_secs(3), + ); + + let status = child.wait().unwrap(); + drop(writer); + drop(pair.master); + reader_thread.join().unwrap(); + server.join().unwrap(); + assert!(status.success(), "Dosh client exited with {status:?}"); +} + fn write_client_fixture(home: &Path, cache: &Path, port: u16) { let config = ClientConfig { server: "local".to_string(), dosh_host: Some("127.0.0.1".to_string()), dosh_port: port, cache_attach_tickets: false, + reconnect_timeout_secs: 1, credential_cache: cache.to_string_lossy().to_string(), auth_preference: "ssh".to_string(), predict: false, @@ -479,6 +555,89 @@ fn run_backpressure_server( } } +fn run_reconnect_server( + socket: UdpSocket, + observations: mpsc::Sender, + expected_input: &[u8], +) { + let mut resume_count = 0u64; + let mut received_input = Vec::new(); + let mut buf = [0u8; 65535]; + loop { + let (n, source) = socket.recv_from(&mut buf).unwrap(); + let packet = protocol::decode(&buf[..n]).unwrap(); + match packet.header.kind { + PacketKind::ResumeRequest => { + let plain = + protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap(); + let request: protocol::ResumeRequest = protocol::from_body(&plain).unwrap(); + assert_eq!(request.session, SESSION); + resume_count += 1; + let bytes = if resume_count == 1 { + b"DOSH_RECONNECT_READY".as_slice() + } else { + observations.send(ServerObservation::Reconnected).unwrap(); + b"DOSH_RECONNECT_SNAPSHOT".as_slice() + }; + send_frame( + &socket, + source, + PacketKind::ResumeOk, + resume_count, + 10, + bytes, + true, + false, + ); + } + PacketKind::Input => { + let plain = + protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap(); + let input: Input = protocol::from_body(&plain).unwrap(); + received_input.extend_from_slice(&input.bytes); + observations + .send(ServerObservation::Input(input.bytes)) + .unwrap(); + if contains(&received_input, expected_input) { + send_frame( + &socket, + source, + PacketKind::Frame, + 4, + 12, + b"DOSH_ORDER_SECOND_", + false, + false, + ); + send_frame( + &socket, + source, + PacketKind::Frame, + 3, + 11, + b"DOSH_ORDER_FIRST_", + false, + false, + ); + send_frame( + &socket, + source, + PacketKind::Frame, + 5, + 13, + b"DOSH_RECONNECT_DONE", + false, + true, + ); + break; + } + } + PacketKind::Ping | PacketKind::Ack => {} + _ => {} + } + } +} + #[allow(clippy::too_many_arguments)] fn send_frame( socket: &UdpSocket,