Implement streamlocal-forward for remote => local UDS forwarding

README.md

@@ -17,6 +17,7 @@ This is a fork of [Thrussh](https://nest.pijul.com/pijul/thrussh) by Pierre-Éti
 * `direct-tcpip` (local port forwarding)
 * `forward-tcpip` (remote port forwarding) ✨
 * `direct-streamlocal` (local UNIX socket forwarding, client only) ✨
+* `forward-streamlocal` (remote UNIX socket forwarding) ✨
 * Ciphers:
   * `[email protected]`
   * `[email protected]` ✨

russh/src/client/encrypted.rs

@@ -806,6 +806,17 @@ impl Session {
                                 )
                                 .await?
                         }
+                        ChannelType::ForwardedStreamLocal(d) => {
+                            confirm();
+                            let channel = self.accept_server_initiated_channel(id, &msg);
+                            client
+                                .server_channel_open_forwarded_streamlocal(
+                                    channel,
+                                    &d.socket_path,
+                                    self,
+                                )
+                                .await?;
+                        }
                         ChannelType::AgentForward => {
                             confirm();
                             client.server_channel_open_agent_forward(id, self).await?
@@ -849,6 +860,24 @@ impl Session {
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
                         let _ = return_channel.send(true);
                     }
+                    Some(GlobalRequestResponse::StreamLocalForward(return_channel)) => {
+                        let mut r = buf.reader(1);
+                        let socket_path: Option<String> = match r.read_string() {
+                            Ok(socket_path) => Some(
+                                std::str::from_utf8(socket_path)
+                                    .map_err(crate::Error::from)?
+                                    .into(),
+                            ),
+                            Err(e) => {
+                                error!("Error parsing socket path for StreamLocalForward request: {e:?}");
+                                None
+                            }
+                        };
+                        let _ = return_channel.send(socket_path);
+                    }
+                    Some(GlobalRequestResponse::CancelStreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(true);
+                    }
                     None => {
                         error!("Received global request failure for unknown request!")
                     }
@@ -867,6 +896,12 @@ impl Session {
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
                         let _ = return_channel.send(false);
                     }
+                    Some(GlobalRequestResponse::StreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(None);
+                    }
+                    Some(GlobalRequestResponse::CancelStreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(false);
+                    }
                     None => {
                         error!("Received global request failure for unknown request!")
                     }

russh/src/client/mod.rs

@@ -162,6 +162,16 @@ pub enum Msg {
         address: String,
         port: u32,
     },
+    StreamLocalForward {
+        /// Provide a channel for the reply result to request a reply from the server
+        reply_channel: Option<oneshot::Sender<Option<String>>>,
+        socket_path: String,
+    },
+    CancelStreamLocalForward {
+        /// Provide a channel for the reply result to request a reply from the server
+        reply_channel: Option<oneshot::Sender<bool>>,
+        socket_path: String,
+    },
     Close {
         id: ChannelId,
     },
@@ -573,6 +583,7 @@ impl<H: Handler> Handle<H> {
         }
     }
 
+    // Requests the server to close a TCP/IP forward channel
     pub async fn cancel_tcpip_forward<A: Into<String>>(
         &self,
         address: A,
@@ -598,6 +609,54 @@ impl<H: Handler> Handle<H> {
         }
     }
 
+    // Requests the server to open a UDS forward channel
+    pub async fn streamlocal_forward<A: Into<String>>(
+        &mut self,
+        socket_path: A,
+    ) -> Result<String, crate::Error> {
+        let (reply_send, reply_recv) = oneshot::channel();
+        self.sender
+            .send(Msg::StreamLocalForward {
+                reply_channel: Some(reply_send),
+                socket_path: socket_path.into(),
+            })
+            .await
+            .map_err(|_| crate::Error::SendError)?;
+
+        match reply_recv.await {
+            Ok(Some(returned_socket_path)) => Ok(returned_socket_path),
+            Ok(None) => Err(crate::Error::RequestDenied),
+            Err(e) => {
+                error!("Unable to receive StreamLocalForward result: {e:?}");
+                Err(crate::Error::Disconnect)
+            }
+        }
+    }
+
+    // Requests the server to close a UDS forward channel
+    pub async fn cancel_streamlocal_forward<A: Into<String>>(
+        &self,
+        socket_path: A,
+    ) -> Result<(), crate::Error> {
+        let (reply_send, reply_recv) = oneshot::channel();
+        self.sender
+            .send(Msg::CancelStreamLocalForward {
+                reply_channel: Some(reply_send),
+                socket_path: socket_path.into(),
+            })
+            .await
+            .map_err(|_| crate::Error::SendError)?;
+
+        match reply_recv.await {
+            Ok(true) => Ok(()),
+            Ok(false) => Err(crate::Error::RequestDenied),
+            Err(e) => {
+                error!("Unable to receive CancelStreamLocalForward result: {e:?}");
+                Err(crate::Error::Disconnect)
+            }
+        }
+    }
+
     /// Sends a disconnect message.
     pub async fn disconnect(
         &self,
@@ -1052,6 +1111,14 @@ impl Session {
                 address,
                 port,
             } => self.cancel_tcpip_forward(reply_channel, &address, port),
+            Msg::StreamLocalForward {
+                reply_channel,
+                socket_path,
+            } => self.streamlocal_forward(reply_channel, &socket_path),
+            Msg::CancelStreamLocalForward {
+                reply_channel,
+                socket_path,
+            } => self.cancel_streamlocal_forward(reply_channel, &socket_path),
             Msg::Disconnect {
                 reason,
                 description,
@@ -1553,6 +1620,17 @@ pub trait Handler: Sized + Send {
         Ok(())
     }
 
+    // Called when the server opens a channel for a new remote UDS forwarding connection
+    #[allow(unused_variables)]
+    async fn server_channel_open_forwarded_streamlocal(
+        &mut self,
+        channel: Channel<Msg>,
+        socket_path: &str,
+        session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        Ok(())
+    }
+
     /// Called when the server opens an agent forwarding channel
     #[allow(unused_variables)]
     async fn server_channel_open_agent_forward(

russh/src/client/session.rs

@@ -294,7 +294,7 @@ impl Session {
 
     /// Requests cancellation of TCP/IP forwarding from the server
     ///
-    /// If `want_reply` is `true`, returns a oneshot receiveing the server's reply:
+    /// If `want_reply` is `true`, returns a oneshot receiving the server's reply:
     /// `true` for a success message, or `false` for failure
     pub fn cancel_tcpip_forward(
         &mut self,
@@ -319,6 +319,58 @@ impl Session {
         }
     }
 
+    /// Requests a UDS forwarding from the server, `socket path` being the server side socket path.
+    ///
+    /// If `reply_channel` is not None, sets want_reply and returns the server's response via the channel,
+    /// [`Some<String>`] for a success message with the client side socket path, [`None`] for failure.
+    pub fn streamlocal_forward(
+        &mut self,
+        reply_channel: Option<oneshot::Sender<Option<String>>>,
+        socket_path: &str,
+    ) {
+        if let Some(ref mut enc) = self.common.encrypted {
+            let want_reply = reply_channel.is_some();
+            if let Some(reply_channel) = reply_channel {
+                self.open_global_requests.push_back(
+                    crate::session::GlobalRequestResponse::StreamLocalForward(reply_channel),
+                );
+            }
+            push_packet!(enc.write, {
+                enc.write.push(msg::GLOBAL_REQUEST);
+                enc.write
+                    .extend_ssh_string(b"[email protected]");
+                enc.write.push(want_reply as u8);
+                enc.write.extend_ssh_string(socket_path.as_bytes());
+            });
+        }
+    }
+
+    /// Requests cancellation of UDS forwarding from the server
+    ///
+    /// If `want_reply` is true, returns a oneshot receiving the server's reply:
+    /// `true` for a success message and `false` for failure.
+    pub fn cancel_streamlocal_forward(
+        &mut self,
+        reply_channel: Option<oneshot::Sender<bool>>,
+        socket_path: &str,
+    ) {
+        if let Some(ref mut enc) = self.common.encrypted {
+            let want_reply = reply_channel.is_some();
+            if let Some(reply_channel) = reply_channel {
+                self.open_global_requests.push_back(
+                    crate::session::GlobalRequestResponse::CancelStreamLocalForward(reply_channel),
+                );
+            }
+            push_packet!(enc.write, {
+                enc.write.push(msg::GLOBAL_REQUEST);
+                enc.write
+                    .extend_ssh_string(b"[email protected]");
+                enc.write.push(want_reply as u8);
+                enc.write.extend_ssh_string(socket_path.as_bytes());
+            });
+        }
+    }
+
     pub fn send_keepalive(&mut self, want_reply: bool) {
         self.open_global_requests
             .push_back(crate::session::GlobalRequestResponse::Keepalive);

russh/src/parsing.rs

@@ -34,6 +34,9 @@ impl OpenChannelMessage {
             }
             b"direct-tcpip" => ChannelType::DirectTcpip(TcpChannelInfo::new(r)?),
             b"forwarded-tcpip" => ChannelType::ForwardedTcpIp(TcpChannelInfo::new(r)?),
+            b"forwarded-streamlocal" => {
+                ChannelType::ForwardedStreamLocal(StreamLocalChannelInfo::new(r)?)
+            }
             b"[email protected]" => ChannelType::AgentForward,
             t => ChannelType::Unknown { typ: t.to_vec() },
         };
@@ -93,6 +96,7 @@ pub enum ChannelType {
     },
     DirectTcpip(TcpChannelInfo),
     ForwardedTcpIp(TcpChannelInfo),
+    ForwardedStreamLocal(StreamLocalChannelInfo),
     AgentForward,
     Unknown {
         typ: Vec<u8>,
@@ -107,6 +111,21 @@ pub struct TcpChannelInfo {
     pub originator_port: u32,
 }
 
+#[derive(Debug)]
+pub struct StreamLocalChannelInfo {
+    pub socket_path: String,
+}
+
+impl StreamLocalChannelInfo {
+    fn new(r: &mut Position) -> Result<Self, crate::Error> {
+        let socket_path = std::str::from_utf8(r.read_string().map_err(crate::Error::from)?)
+            .map_err(crate::Error::from)?
+            .to_owned();
+
+        Ok(Self { socket_path })
+    }
+}
+
 impl TcpChannelInfo {
     fn new(r: &mut Position) -> Result<Self, crate::Error> {
         let host_to_connect = std::str::from_utf8(r.read_string().map_err(crate::Error::from)?)

russh/src/server/encrypted.rs

@@ -1026,6 +1026,41 @@ impl Session {
                         }
                         Ok(())
                     }
+                    b"[email protected]" => {
+                        let server_socket_path =
+                            std::str::from_utf8(r.read_string().map_err(crate::Error::from)?)
+                                .map_err(crate::Error::from)?;
+                        debug!("handler.streamlocal_forward {:?}", server_socket_path);
+                        let mut client_socket_path = String::from(server_socket_path).clone();
+                        let result = handler
+                            .streamlocal_forward(server_socket_path, &mut client_socket_path, self)
+                            .await?;
+                        if let Some(ref mut enc) = self.common.encrypted {
+                            if result {
+                                push_packet!(enc.write, enc.write.push(msg::REQUEST_SUCCESS))
+                            } else {
+                                push_packet!(enc.write, enc.write.push(msg::REQUEST_FAILURE))
+                            }
+                        }
+                        Ok(())
+                    }
+                    b"[email protected]" => {
+                        let socket_path =
+                            std::str::from_utf8(r.read_string().map_err(crate::Error::from)?)
+                                .map_err(crate::Error::from)?;
+                        debug!("handler.cancel_streamlocal_forward {:?}", socket_path);
+                        let result = handler
+                            .cancel_streamlocal_forward(socket_path, self)
+                            .await?;
+                        if let Some(ref mut enc) = self.common.encrypted {
+                            if result {
+                                push_packet!(enc.write, enc.write.push(msg::REQUEST_SUCCESS))
+                            } else {
+                                push_packet!(enc.write, enc.write.push(msg::REQUEST_FAILURE))
+                            }
+                        }
+                        Ok(())
+                    }
                     _ => {
                         if let Some(ref mut enc) = self.common.encrypted {
                             push_packet!(enc.write, {
@@ -1090,6 +1125,24 @@ impl Session {
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
                         let _ = return_channel.send(true);
                     }
+                    Some(GlobalRequestResponse::StreamLocalForward(return_channel)) => {
+                        let mut r = buf.reader(1);
+                        let socket_path: Option<String> = match r.read_string() {
+                            Ok(socket_path) => Some(
+                                std::str::from_utf8(socket_path)
+                                    .map_err(crate::Error::from)?
+                                    .into(),
+                            ),
+                            Err(e) => {
+                                error!("Error parsing socket path for StreamLocalForward request: {e:?}");
+                                None
+                            }
+                        };
+                        let _ = return_channel.send(socket_path);
+                    }
+                    Some(GlobalRequestResponse::CancelStreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(true);
+                    }
                     None => {
                         error!("Received global request failure for unknown request!")
                     }
@@ -1108,6 +1161,12 @@ impl Session {
                     Some(GlobalRequestResponse::CancelTcpIpForward(return_channel)) => {
                         let _ = return_channel.send(false);
                     }
+                    Some(GlobalRequestResponse::StreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(None);
+                    }
+                    Some(GlobalRequestResponse::CancelStreamLocalForward(return_channel)) => {
+                        let _ = return_channel.send(false);
+                    }
                     None => {
                         error!("Received global request failure for unknown request!")
                     }
@@ -1214,6 +1273,16 @@ impl Session {
                 }
                 result
             }
+            ChannelType::ForwardedStreamLocal(d) => {
+                let mut result = handler
+                    .channel_open_forwarded_streamlocal(channel, &d.socket_path, self)
+                    .await;
+                if let Ok(allowed) = &mut result {
+                    self.channels.insert(sender_channel, reference);
+                    self.finalize_channel_open(&msg, channel_params, *allowed);
+                }
+                result
+            }
             ChannelType::AgentForward => {
                 if let Some(ref mut enc) = self.common.encrypted {
                     msg.fail(