diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 0c784a2..ff0fb90 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -61,6 +61,7 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::sync::mpsc; const STREAM_INITIAL_WINDOW: usize = 1024 * 1024; +const STREAM_RETIRED_TOMBSTONES: usize = 16 * 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); @@ -4335,6 +4336,8 @@ async fn run_terminal( let mut stream_sent_data: HashMap> = HashMap::new(); let mut stream_next_recv_offset: HashMap = HashMap::new(); let mut stream_recv_pending: HashMap>> = HashMap::new(); + let mut stream_retired: HashSet = HashSet::new(); + let mut stream_retired_order: VecDeque = VecDeque::new(); let mut pending_user_input: VecDeque = VecDeque::new(); let mut pending_user_input_bytes = 0usize; let mut startup_input_hold_until: Option = None; @@ -4827,6 +4830,18 @@ async fn run_terminal( continue; }; last_packet_at = Instant::now(); + if stream_retired.contains(&open.stream_id) { + send_stream_open_reject( + &socket, + addr, + &cred, + &mut send_seq, + open.stream_id, + "stream closed".to_string(), + ) + .await?; + continue; + } if opened_streams.contains(&open.stream_id) { send_stream_open_ok( &socket, @@ -5024,7 +5039,7 @@ async fn run_terminal( { let _ = writer.write_all(&[5, 5, 0, 1, 0, 0, 0, 0, 0, 0]).await; } - cleanup_stream_state( + retire_stream_state( reject.stream_id, &mut opened_streams, &mut stream_send_credit, @@ -5034,6 +5049,8 @@ async fn run_terminal( &mut stream_sent_data, &mut stream_next_recv_offset, &mut stream_recv_pending, + &mut stream_retired, + &mut stream_retired_order, ); stream_writers.remove(&reject.stream_id); } @@ -5113,7 +5130,7 @@ async fn run_terminal( }; last_packet_at = Instant::now(); pending_socks_replies.remove(&close.stream_id); - cleanup_stream_state( + retire_stream_state( close.stream_id, &mut opened_streams, &mut stream_send_credit, @@ -5123,6 +5140,8 @@ async fn run_terminal( &mut stream_sent_data, &mut stream_next_recv_offset, &mut stream_recv_pending, + &mut stream_retired, + &mut stream_retired_order, ); stream_writers.remove(&close.stream_id); #[cfg(unix)] @@ -5175,7 +5194,7 @@ async fn run_terminal( } Some(ForwardEvent::Close { stream_id }) => { pending_socks_replies.remove(&stream_id); - cleanup_stream_state( + retire_stream_state( stream_id, &mut opened_streams, &mut stream_send_credit, @@ -5185,6 +5204,8 @@ async fn run_terminal( &mut stream_sent_data, &mut stream_next_recv_offset, &mut stream_recv_pending, + &mut stream_retired, + &mut stream_retired_order, ); stream_writers.remove(&stream_id); #[cfg(unix)] @@ -6188,6 +6209,77 @@ fn cleanup_stream_state( stream_recv_pending.remove(&stream_id); } +#[allow(clippy::too_many_arguments)] +fn stream_state_exists( + stream_id: u64, + opened_streams: &HashSet, + stream_send_credit: &HashMap, + stream_pending_data: &HashMap>>, + stream_pending_opens: &HashMap, + stream_next_send_offset: &HashMap, + stream_sent_data: &HashMap>, + stream_next_recv_offset: &HashMap, + stream_recv_pending: &HashMap>>, +) -> bool { + opened_streams.contains(&stream_id) + || stream_send_credit.contains_key(&stream_id) + || stream_pending_data.contains_key(&stream_id) + || stream_pending_opens.contains_key(&stream_id) + || stream_next_send_offset.contains_key(&stream_id) + || stream_sent_data.contains_key(&stream_id) + || stream_next_recv_offset.contains_key(&stream_id) + || stream_recv_pending.contains_key(&stream_id) +} + +#[allow(clippy::too_many_arguments)] +fn retire_stream_state( + stream_id: u64, + opened_streams: &mut HashSet, + stream_send_credit: &mut HashMap, + stream_pending_data: &mut HashMap>>, + stream_pending_opens: &mut HashMap, + stream_next_send_offset: &mut HashMap, + stream_sent_data: &mut HashMap>, + stream_next_recv_offset: &mut HashMap, + stream_recv_pending: &mut HashMap>>, + stream_retired: &mut HashSet, + stream_retired_order: &mut VecDeque, +) { + let had_state = stream_state_exists( + stream_id, + opened_streams, + stream_send_credit, + stream_pending_data, + stream_pending_opens, + stream_next_send_offset, + stream_sent_data, + stream_next_recv_offset, + stream_recv_pending, + ); + cleanup_stream_state( + stream_id, + opened_streams, + stream_send_credit, + stream_pending_data, + stream_pending_opens, + stream_next_send_offset, + stream_sent_data, + stream_next_recv_offset, + stream_recv_pending, + ); + if !had_state || STREAM_RETIRED_TOMBSTONES == 0 { + return; + } + if stream_retired.insert(stream_id) { + stream_retired_order.push_back(stream_id); + } + while stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES { + if let Some(expired) = stream_retired_order.pop_front() { + stream_retired.remove(&expired); + } + } +} + fn ack_stream_data( stream_sent_data: &mut HashMap>, stream_send_credit: &mut HashMap, @@ -7481,12 +7573,12 @@ mod tests { 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, 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, + render_status_overlay, requested_env, resolved_startup_command, retire_stream_state, + 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, + 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, terminal_private_mode_transition, toml_bare_key_or_quoted, update_check_requested, update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias, @@ -7496,7 +7588,7 @@ mod tests { use dosh::protocol::{self, Frame, PacketKind}; use dosh::udp::is_transient_udp_send_error; use ed25519_dalek::{SigningKey, VerifyingKey}; - use std::collections::{BTreeMap, HashMap, VecDeque}; + use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::fs; use std::time::{Duration, Instant}; @@ -7636,6 +7728,54 @@ mod tests { assert!(stream_recv_pending.contains_key(&other_stream_id)); } + #[test] + fn retire_stream_state_tombstones_only_real_stream_state() { + let stream_id = 42; + let mut opened_streams = HashSet::from([stream_id]); + let mut stream_send_credit = HashMap::from([(stream_id, 12)]); + let mut stream_pending_data = HashMap::new(); + let mut stream_pending_opens = HashMap::new(); + let mut stream_next_send_offset = HashMap::new(); + let mut stream_sent_data = HashMap::new(); + let mut stream_next_recv_offset = HashMap::new(); + let mut stream_recv_pending = HashMap::new(); + let mut stream_retired = HashSet::new(); + let mut stream_retired_order = VecDeque::new(); + + retire_stream_state( + stream_id, + &mut opened_streams, + &mut stream_send_credit, + &mut stream_pending_data, + &mut stream_pending_opens, + &mut stream_next_send_offset, + &mut stream_sent_data, + &mut stream_next_recv_offset, + &mut stream_recv_pending, + &mut stream_retired, + &mut stream_retired_order, + ); + retire_stream_state( + 99, + &mut opened_streams, + &mut stream_send_credit, + &mut stream_pending_data, + &mut stream_pending_opens, + &mut stream_next_send_offset, + &mut stream_sent_data, + &mut stream_next_recv_offset, + &mut stream_recv_pending, + &mut stream_retired, + &mut stream_retired_order, + ); + + assert!(!opened_streams.contains(&stream_id)); + assert!(!stream_send_credit.contains_key(&stream_id)); + assert!(stream_retired.contains(&stream_id)); + assert!(!stream_retired.contains(&99)); + assert_eq!(stream_retired_order, VecDeque::from([stream_id])); + } + #[test] fn managed_ssh_block_is_replaced_not_duplicated() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 2b4aba7..4c6721b 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -47,6 +47,7 @@ use tokio::process::Command as TokioCommand; use tokio::sync::mpsc; const STREAM_INITIAL_WINDOW: usize = 1024 * 1024; +const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024; /// Persist a persistent session's screen to disk after this many bytes of output /// have accumulated since the last mirror. Keeps the atomic write off the @@ -399,6 +400,8 @@ struct ClientState { stream_sent_data: HashMap>, stream_next_recv_offset: HashMap, stream_recv_pending: HashMap>>, + stream_retired: HashSet, + stream_retired_order: VecDeque, /// Current transport key epoch (0 = original handshake key). Bumped on rekey. epoch: u64, /// When the current epoch began, for the wall-clock rekey trigger. @@ -1201,6 +1204,8 @@ async fn handle_native_user_auth( stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1348,6 +1353,8 @@ async fn handle_bootstrap_attach( stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1482,6 +1489,8 @@ async fn handle_ticket_attach( stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -1831,7 +1840,9 @@ async fn handle_stream_open( } client.endpoint = peer; client.last_seen = Instant::now(); - if client.opened_streams.contains(&open.stream_id) { + if client.stream_retired.contains(&open.stream_id) { + StreamOpenCheck::New(Err(anyhow!("stream {} is closed", open.stream_id))) + } else if client.opened_streams.contains(&open.stream_id) { StreamOpenCheck::Duplicate } else { StreamOpenCheck::New(stream_open_allowed(&config, client, &open)) @@ -2138,7 +2149,7 @@ async fn handle_stream_open_reject( } client.endpoint = peer; client.last_seen = Instant::now(); - cleanup_client_stream(client, reject.stream_id); + retire_client_stream(client, reject.stream_id); Ok(()) } @@ -2196,7 +2207,7 @@ async fn handle_stream_close( // Connection migration on any authenticated, fresh packet (spec ยง11). client.endpoint = peer; client.last_seen = Instant::now(); - cleanup_client_stream(client, close.stream_id); + retire_client_stream(client, close.stream_id); Ok(()) } @@ -3288,6 +3299,34 @@ fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) { client.stream_recv_pending.remove(&stream_id); } +fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool { + client.stream_writers.contains_key(&stream_id) + || client.stream_pending_opens.contains_key(&stream_id) + || client.opened_streams.contains(&stream_id) + || client.stream_send_credit.contains_key(&stream_id) + || client.stream_pending_data.contains_key(&stream_id) + || client.stream_next_send_offset.contains_key(&stream_id) + || client.stream_sent_data.contains_key(&stream_id) + || client.stream_next_recv_offset.contains_key(&stream_id) + || client.stream_recv_pending.contains_key(&stream_id) +} + +fn retire_client_stream(client: &mut ClientState, stream_id: u64) { + let had_state = client_has_stream_state(client, stream_id); + cleanup_client_stream(client, stream_id); + if !had_state || STREAM_RETIRED_TOMBSTONES == 0 { + return; + } + if client.stream_retired.insert(stream_id) { + client.stream_retired_order.push_back(stream_id); + } + while client.stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES { + if let Some(expired) = client.stream_retired_order.pop_front() { + client.stream_retired.remove(&expired); + } + } +} + fn ack_stream_data(client: &mut ClientState, stream_id: u64, received_offset: u64) { let Some(sent) = client.stream_sent_data.get_mut(&stream_id) else { return; @@ -3348,7 +3387,7 @@ async fn send_stream_close_to_client( { let mut locked = state.lock().expect("server state poisoned"); if let Some(client) = locked.client_mut(&client_id) { - cleanup_client_stream(client, stream_id); + retire_client_stream(client, stream_id); } } let body = protocol::to_body(&StreamClose { stream_id })?; @@ -4047,6 +4086,8 @@ mod tests { stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -4154,6 +4195,25 @@ mod tests { assert!(client.stream_recv_pending.contains_key(&other_stream_id)); } + #[test] + fn retire_client_stream_tombstones_only_real_stream_state() { + let stream_id = 42; + let mut client = test_client_state([8u8; 32]); + client.opened_streams.insert(stream_id); + client + .stream_send_credit + .insert(stream_id, STREAM_INITIAL_WINDOW); + + retire_client_stream(&mut client, stream_id); + retire_client_stream(&mut client, 99); + + assert!(!client.opened_streams.contains(&stream_id)); + assert!(!client.stream_send_credit.contains_key(&stream_id)); + assert!(client.stream_retired.contains(&stream_id)); + assert!(!client.stream_retired.contains(&99)); + assert_eq!(client.stream_retired_order, VecDeque::from([stream_id])); + } + #[test] fn client_index_stays_in_sync_with_session_clients() { let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); @@ -4359,6 +4419,74 @@ mod tests { assert_eq!(ok.stream_id, stream_id); } + #[tokio::test] + async fn retired_stream_open_is_rejected_not_reopened() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx); + let client_id = [15u8; 16]; + let session_key = [16u8; 32]; + let stream_id = 100; + 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(); + client.stream_retired.insert(stream_id); + client.stream_retired_order.push_back(stream_id); + state.sessions.insert( + "test".to_string(), + Session { + pty: None, + parser: vt100::Parser::new(24, 80, 100), + clients: HashMap::from([(client_id, client)]), + output_seq: 0, + recent: VecDeque::new(), + empty_since: None, + holder_control: None, + persistent: false, + bytes_since_persist: 0, + last_persisted_seq: 0, + }, + ); + state.client_index.insert(client_id, "test".to_string()); + let state = Arc::new(Mutex::new(state)); + let raw = protocol::encode_encrypted( + PacketKind::StreamOpen, + client_id, + 2, + 0, + &session_key, + CLIENT_TO_SERVER, + &protocol::to_body(&StreamOpen { + stream_id, + target_host: "127.0.0.1".to_string(), + target_port: 9, + }) + .unwrap(), + ) + .unwrap(); + let packet = protocol::decode(&raw).unwrap(); + + handle_stream_open(&state, &sender, receiver.local_addr().unwrap(), &packet) + .await + .unwrap(); + + let mut buf = [0u8; 2048]; + let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf)) + .await + .unwrap() + .unwrap(); + let packet = protocol::decode(&buf[..n]).unwrap(); + assert_eq!(packet.header.kind, PacketKind::StreamOpenReject); + let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap(); + let reject: StreamOpenReject = protocol::from_body(&plain).unwrap(); + assert_eq!(reject.stream_id, stream_id); + assert!(reject.reason.contains("closed")); + + let locked = state.lock().unwrap(); + let client = locked.sessions["test"].clients.get(&client_id).unwrap(); + assert!(!client.opened_streams.contains(&stream_id)); + } + #[tokio::test] async fn pending_server_stream_open_is_retransmitted() { let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); @@ -4546,6 +4674,8 @@ mod tests { stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0, @@ -4640,6 +4770,8 @@ mod tests { stream_sent_data: HashMap::new(), stream_next_recv_offset: HashMap::new(), stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), epoch: 0, epoch_started: Instant::now(), epoch_packets: 0,