Harden SDK stream ids and duplicate opens

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<OutgoingStreamPacket>,
@@ -258,6 +262,12 @@ impl StreamMux {
     }
 
     pub fn accept_open(&mut self, open: StreamOpen) -> Result<IncomingStreamOpen> {
+        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<String> {
     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();