Fix deadlock when handling open channel on contended handler

russh/src/client/encrypted.rs

@@ -441,7 +441,9 @@ impl Session {
                 }
 
                 if let Some(sender) = self.channels.remove(&channel_num) {
-                    let _ = sender.send(ChannelMsg::OpenFailure(reason_code.clone())).await;
+                    let _ = sender
+                        .send(ChannelMsg::OpenFailure(reason_code.clone()))
+                        .await;
                 }
 
                 let _ = self.sender.send(Reply::ChannelOpenFailure);
@@ -714,14 +716,16 @@ impl Session {
                     channel_params,
                 };
                 let reply = ChannelOpenHandle::new(
-                    self.inbound_channel_sender.clone(),
+                    self.priority_sender.clone(),
                     pending,
                     |pending, result| Msg::ServerChannelOpenReply { pending, result },
                 );
 
                 match &msg.typ {
                     ChannelType::Session => {
-                        client.server_channel_open_session(channel, reply, self).await?
+                        client
+                            .server_channel_open_session(channel, reply, self)
+                            .await?
                     }
                     ChannelType::DirectTcpip(d) => {
                         client
@@ -790,7 +794,9 @@ impl Session {
                     }
                     ChannelType::Unknown { typ } => {
                         if client.should_accept_unknown_server_channel(id, typ).await {
-                            client.server_channel_open_unknown(channel, reply, self).await?;
+                            client
+                                .server_channel_open_unknown(channel, reply, self)
+                                .await?;
                         } else {
                             debug!("unknown channel type: {typ}");
                             if let Some(ref mut enc) = self.common.encrypted {

russh/src/client/mod.rs

@@ -95,6 +95,8 @@ pub struct Session {
     target_window_size: u32,
     pending_reads: Vec<Vec<u8>>,
     pending_len: u32,
+    priority_sender: UnboundedSender<Msg>,
+    priority_receiver: UnboundedReceiver<Msg>,
     inbound_channel_sender: Sender<Msg>,
     inbound_channel_receiver: Receiver<Msg>,
     open_global_requests: VecDeque<GlobalRequestResponse>,
@@ -1100,13 +1102,16 @@ impl Session {
         receiver: Receiver<Msg>,
         sender: UnboundedSender<Reply>,
     ) -> Self {
+        let (priority_sender, priority_receiver) = unbounded_channel();
         let (inbound_channel_sender, inbound_channel_receiver) = channel(10);
         Self {
             common,
             receiver,
             sender,
             kex: SessionKexState::Idle,
             target_window_size,
+            priority_sender,
+            priority_receiver,
             inbound_channel_sender,
             inbound_channel_receiver,
             channels: HashMap::new(),
@@ -1134,6 +1139,7 @@ impl Session {
             .await;
         trace!("disconnected");
         self.receiver.close();
+        self.priority_receiver.close();
         self.inbound_channel_receiver.close();
         map_err!(stream_write.shutdown().await)?;
         match result {
@@ -1198,8 +1204,7 @@ impl Session {
             // Keep reading the network for window adjustments, but leave
             // application output in its bounded receivers while a channel is
             // window-blocked.
-            let can_receive_outbound =
-                !self.kex.active() && !self.common.has_any_pending_data();
+            let can_receive_outbound = !self.kex.active() && !self.common.has_any_pending_data();
             tokio::select! {
                 r = &mut reading => {
                     let (stream_read, mut buffer, mut opening_cipher) = match r {
@@ -1247,6 +1252,7 @@ impl Session {
                     return Err(crate::Error::InactivityTimeout.into());
                 }
                 msg = self.receiver.recv(), if can_receive_outbound => {
+                    self.drain_priority_msgs()?;
                     match msg {
                         Some(msg) => self.handle_msg(msg)?,
                         None => {
@@ -1257,20 +1263,32 @@ impl Session {
 
                     // eagerly take all outgoing messages so writes are batched
                     while !self.kex.active() && !self.common.has_any_pending_data() {
+                        self.drain_priority_msgs()?;
                         match self.receiver.try_recv() {
                             Ok(next) => self.handle_msg(next)?,
                             Err(_) => break
                         }
                     }
                 }
+                msg = self.priority_receiver.recv(), if !self.kex.active() => {
+                    match msg {
+                        Some(msg) => self.handle_msg(msg)?,
+                        None => (),
+                    }
+
+                    // eagerly take all outgoing messages so writes are batched
+                    self.drain_priority_msgs()?;
+                }
                 msg = self.inbound_channel_receiver.recv(), if can_receive_outbound => {
+                    self.drain_priority_msgs()?;
                     match msg {
                         Some(msg) => self.handle_msg(msg)?,
                         None => (),
                     }
 
                     // eagerly take all outgoing messages so writes are batched
                     while !self.kex.active() && !self.common.has_any_pending_data() {
+                        self.drain_priority_msgs()?;
                         match self.inbound_channel_receiver.try_recv() {
                             Ok(next) => self.handle_msg(next)?,
                             Err(_) => break
@@ -1340,6 +1358,21 @@ impl Session {
         })
     }
 
+    /// Channel open replies must be dispatched before any channel traffic
+    /// queued after them: the bounded receivers may hold data for a channel
+    /// whose confirmation is still sitting in the priority queue, and
+    /// dispatching that data first would silently drop it (the channel is
+    /// only registered when its open reply is processed).
+    fn drain_priority_msgs(&mut self) -> Result<(), crate::Error> {
+        while !self.kex.active() {
+            match self.priority_receiver.try_recv() {
+                Ok(msg) => self.handle_msg(msg)?,
+                Err(_) => break,
+            }
+        }
+        Ok(())
+    }
+
     fn handle_msg(&mut self, msg: Msg) -> Result<(), crate::Error> {
         match msg {
             Msg::Authenticate { user, method } => {
@@ -1728,8 +1761,7 @@ mod tests {
     use std::sync::Arc;
 
     use ssh_encoding::Encode;
-    use tokio::sync::mpsc::channel;
-    use tokio::sync::mpsc::unbounded_channel;
+    use tokio::sync::mpsc::{channel, unbounded_channel};
 
     use super::*;
     use crate::auth::{AuthRequest, Method};

russh/src/client/test.rs

@@ -8,12 +8,12 @@ mod tests {
     use tokio::net::TcpListener;
 
     // Import client types directly since we're in the client module
+    use crate::Error;
     use crate::client::{Config, Handler, connect};
     use crate::keys::PrivateKeyWithHashAlg;
-    use rand::rng;
     use crate::server::{self, Auth, Handler as ServerHandler, Server, Session};
     use crate::{ChannelId, SshId}; // Import directly from crate root
-    use crate::Error;
+    use rand::rng;
 
     #[derive(Clone)]
     struct TestServer {

russh/src/lib_inner.rs

@@ -556,14 +556,14 @@ pub struct PendingChannelOpen {
 /// [`reject`](ChannelOpenHandle::reject) automatically sends an
 /// `AdministrativelyProhibited` rejection.
 pub struct ChannelOpenHandleInner<M: Send> {
-    sender: tokio::sync::mpsc::Sender<M>,
+    sender: tokio::sync::mpsc::UnboundedSender<M>,
     inner: Option<PendingChannelOpen>,
     make_msg: fn(PendingChannelOpen, Result<(), ChannelOpenFailure>) -> M,
 }
 
 impl<M: Send> ChannelOpenHandleInner<M> {
     pub(crate) fn new(
-        sender: tokio::sync::mpsc::Sender<M>,
+        sender: tokio::sync::mpsc::UnboundedSender<M>,
         pending: PendingChannelOpen,
         make_msg: fn(PendingChannelOpen, Result<(), ChannelOpenFailure>) -> M,
     ) -> Self {
@@ -574,32 +574,28 @@ impl<M: Send> ChannelOpenHandleInner<M> {
         }
     }
 
-    fn try_send_reply(&mut self, result: Result<(), ChannelOpenFailure>) {
-        if let Some(pending) = self.inner.take() {
-            let _ = self.sender.try_send((self.make_msg)(pending, result));
-        }
-    }
-
     /// Accept the channel open request.
     pub async fn accept(mut self) {
         if let Some(pending) = self.inner.take() {
-            let _ = self.sender.send((self.make_msg)(pending, Ok(()))).await;
+            let _ = self.sender.send((self.make_msg)(pending, Ok(())));
         }
     }
 
     /// Reject the channel open request with a reason.
     pub async fn reject(mut self, reason: ChannelOpenFailure) {
         if let Some(pending) = self.inner.take() {
-            let _ = self
-                .sender
-                .send((self.make_msg)(pending, Err(reason)))
-                .await;
+            let _ = self.sender.send((self.make_msg)(pending, Err(reason)));
         }
     }
 }
 
 impl<M: Send> Drop for ChannelOpenHandleInner<M> {
     fn drop(&mut self) {
-        self.try_send_reply(Err(ChannelOpenFailure::AdministrativelyProhibited));
+        if let Some(pending) = self.inner.take() {
+            let _ = self.sender.send((self.make_msg)(
+                pending,
+                Err(ChannelOpenFailure::AdministrativelyProhibited),
+            ));
+        }
     }
 }