diff --git a/Cargo.lock b/Cargo.lock index 98925e3..69943c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "1.0.0-rc10" +version = "1.0.0-rc11" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index f13c09c..122c007 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "1.0.0-rc10" +version = "1.0.0-rc11" edition = "2024" license = "MIT" diff --git a/src/transport.rs b/src/transport.rs index a522b1d..58c455e 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -121,6 +121,10 @@ pub struct AcceptedService { #[derive(Debug, Clone, PartialEq, Eq)] pub enum TransportEvent { Open(IncomingStreamOpen), + DuplicateOpen { + stream_id: u64, + response: OutgoingStreamPacket, + }, OpenOk { stream_id: u64, flushed: Vec, @@ -258,6 +262,12 @@ impl StreamMux { } pub fn accept_open(&mut self, open: StreamOpen) -> Result { + if self.pending_opens.contains_key(&open.stream_id) { + bail!( + "stream {} collides with a local pending open", + open.stream_id + ); + } self.opened_streams.insert(open.stream_id); self.send_credit .entry(open.stream_id) @@ -309,8 +319,17 @@ impl StreamMux { if bytes.is_empty() { return Ok(Vec::new()); } - if !self.opened_streams.contains(&stream_id) - || self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len() + if !self.opened_streams.contains(&stream_id) { + if self.pending_opens.contains_key(&stream_id) { + self.pending_data + .entry(stream_id) + .or_default() + .push_back(bytes); + return Ok(Vec::new()); + } + bail!("stream {stream_id} is not open"); + } + if self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len() || self .pending_data .get(&stream_id) @@ -367,6 +386,12 @@ impl StreamMux { match kind { PacketKind::StreamOpen => { let open: StreamOpen = protocol::from_body(body)?; + if self.opened_streams.contains(&open.stream_id) { + return Ok(TransportEvent::DuplicateOpen { + stream_id: open.stream_id, + response: Self::open_ok(open.stream_id)?, + }); + } self.accept_open(open).map(TransportEvent::Open) } PacketKind::StreamOpenOk => { @@ -625,7 +650,7 @@ impl DoshTransport { ack_seq: config.initial_ack, replay: ReplayWindow::default(), mux: StreamMux::new(config.stream), - next_stream_id: 1, + next_stream_id: initial_stream_id(config.role), last_contact: now, last_keepalive: now, } @@ -669,7 +694,10 @@ impl DoshTransport { pub fn allocate_stream_id(&mut self) -> u64 { let id = self.next_stream_id; - self.next_stream_id = self.next_stream_id.saturating_add(1).max(1); + self.next_stream_id = self + .next_stream_id + .wrapping_add(1) + .max(initial_stream_id(self.role)); id } @@ -811,6 +839,9 @@ impl DoshTransport { async fn send_followups(&mut self, event: &TransportEvent) -> Result<()> { match event { + TransportEvent::DuplicateOpen { response, .. } => { + self.send_outgoing(response.clone()).await?; + } TransportEvent::OpenOk { flushed, .. } | TransportEvent::WindowAdjust { flushed, .. } => { for packet in flushed.clone() { @@ -857,6 +888,13 @@ impl DoshTransport { } } +fn initial_stream_id(role: SessionRole) -> u64 { + match role { + SessionRole::Client => 1, + SessionRole::Server => 1u64 << 63, + } +} + pub fn service_target(name: &str) -> Result { if name.is_empty() || !name @@ -1047,6 +1085,152 @@ mod tests { ); } + #[test] + fn duplicate_stream_open_reacks_without_second_open_event() { + let mut mux = StreamMux::new(TransportConfig::default()); + let open = StreamOpen { + stream_id: 9, + target_host: "@dosh-app".to_string(), + target_port: 0, + }; + let body = protocol::to_body(&open).unwrap(); + assert!(matches!( + mux.handle_packet(PacketKind::StreamOpen, &body).unwrap(), + TransportEvent::Open(_) + )); + + let duplicate = mux.handle_packet(PacketKind::StreamOpen, &body).unwrap(); + match duplicate { + TransportEvent::DuplicateOpen { + stream_id, + response, + } => { + assert_eq!(stream_id, 9); + assert_eq!(response.kind, PacketKind::StreamOpenOk); + let ok: StreamOpenOk = decode(&response); + assert_eq!(ok.stream_id, 9); + } + other => panic!("unexpected duplicate open event {other:?}"), + } + } + + #[test] + fn send_data_to_closed_or_unknown_stream_fails() { + let mut mux = StreamMux::new(TransportConfig::default()); + mux.open_stream(1, "@dosh-test", 0).unwrap(); + assert!(mux.send_data(1, b"queued".to_vec()).unwrap().is_empty()); + mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap(); + mux.close_stream(1).unwrap(); + + assert!(mux.send_data(1, b"lost".to_vec()).is_err()); + assert!(mux.send_data(99, b"lost".to_vec()).is_err()); + } + + #[tokio::test] + async fn session_transport_partitions_client_and_server_stream_ids() { + let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let client_addr = client_socket.local_addr().unwrap(); + let server_addr = server_socket.local_addr().unwrap(); + let key = [44u8; 32]; + let conn_id = [8u8; 16]; + let mut client = DoshTransport::new_owned( + client_socket, + SessionTransportConfig { + role: SessionRole::Client, + conn_id, + session_key: key, + peer_addr: server_addr, + initial_send_seq: 1, + initial_ack: 0, + stream: TransportConfig::default(), + }, + ); + let mut server = DoshTransport::new_owned( + server_socket, + SessionTransportConfig { + role: SessionRole::Server, + conn_id, + session_key: key, + peer_addr: client_addr, + initial_send_seq: 1, + initial_ack: 0, + stream: TransportConfig::default(), + }, + ); + + assert_eq!(client.allocate_stream_id(), 1); + assert_eq!(client.allocate_stream_id(), 2); + assert_eq!(server.allocate_stream_id(), 1u64 << 63); + assert_eq!(server.allocate_stream_id(), (1u64 << 63) + 1); + } + + #[tokio::test] + async fn session_transport_handles_simultaneous_opens_without_id_collision() { + let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let client_addr = client_socket.local_addr().unwrap(); + let server_addr = server_socket.local_addr().unwrap(); + let key = [45u8; 32]; + let conn_id = [9u8; 16]; + let mut client = DoshTransport::new_owned( + client_socket, + SessionTransportConfig { + role: SessionRole::Client, + conn_id, + session_key: key, + peer_addr: server_addr, + initial_send_seq: 1, + initial_ack: 0, + stream: TransportConfig::default(), + }, + ); + let mut server = DoshTransport::new_owned( + server_socket, + SessionTransportConfig { + role: SessionRole::Server, + conn_id, + session_key: key, + peer_addr: client_addr, + initial_send_seq: 1, + initial_ack: 0, + stream: TransportConfig::default(), + }, + ); + + let client_stream = client.open_service("client_app").await.unwrap(); + let server_stream = server.open_service("server_app").await.unwrap(); + assert_ne!(client_stream, server_stream); + + let server_seen = server.recv().await.unwrap(); + let client_seen = client.recv().await.unwrap(); + match server_seen { + SessionEvent::Stream(TransportEvent::Open(open)) => { + assert_eq!(open.stream_id, client_stream); + assert_eq!(open.target_host, "@dosh-client_app"); + server.accept_stream(open.stream_id).await.unwrap(); + } + other => panic!("unexpected server event {other:?}"), + } + match client_seen { + SessionEvent::Stream(TransportEvent::Open(open)) => { + assert_eq!(open.stream_id, server_stream); + assert_eq!(open.target_host, "@dosh-server_app"); + client.accept_stream(open.stream_id).await.unwrap(); + } + other => panic!("unexpected client event {other:?}"), + } + + assert!(matches!( + client.recv().await.unwrap(), + SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == client_stream + )); + assert!(matches!( + server.recv().await.unwrap(), + SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == server_stream + )); + } + #[tokio::test] async fn session_transport_opens_service_and_moves_data_over_encrypted_udp() { let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();