fixed #1220 - direct-streamlocal support

warpgate-protocol-ssh/src/client/channel_session.rs

@@ -87,6 +87,7 @@ impl SessionChannel {
                         },
                         Some(ChannelOperation::OpenShell) => unreachable!(),
                         Some(ChannelOperation::OpenDirectTCPIP { .. }) => unreachable!(),
+                        Some(ChannelOperation::OpenDirectStreamlocal { .. }) => unreachable!(),
                         Some(ChannelOperation::OpenX11 { .. }) => unreachable!(),
                         Some(ChannelOperation::RequestX11(request)) => {
                             self.client_channel.request_x11(

warpgate-protocol-ssh/src/client/mod.rs

@@ -243,6 +243,9 @@ impl RemoteClient {
             ChannelOperation::OpenDirectTCPIP(params) => {
                 self.open_direct_tcpip(channel_id, params).await?;
             }
+            ChannelOperation::OpenDirectStreamlocal(path) => {
+                self.open_direct_streamlocal(channel_id, path).await?;
+            }
             op => {
                 let mut channel_pipes = self.channel_pipes.lock().await;
                 match channel_pipes.get(&channel_id) {
@@ -759,6 +762,30 @@ impl RemoteClient {
         Ok(())
     }
 
+    async fn open_direct_streamlocal(
+        &mut self,
+        channel_id: Uuid,
+        path: String,
+    ) -> Result<(), SshClientError> {
+        if let Some(session) = &self.session {
+            let session = session.lock().await;
+            let channel = session.channel_open_direct_streamlocal(path).await?;
+
+            let (tx, rx) = unbounded_channel();
+            self.channel_pipes.lock().await.insert(channel_id, tx);
+
+            let channel =
+                DirectTCPIPChannel::new(channel, channel_id, rx, self.tx.clone(), self.id);
+            self.child_tasks.push(
+                tokio::task::Builder::new()
+                    .name(&format!("SSH {} {:?} ops", self.id, channel_id))
+                    .spawn(channel.run())
+                    .map_err(|e| SshClientError::Other(Box::new(e)))?,
+            );
+        }
+        Ok(())
+    }
+
     async fn tcpip_forward(&mut self, address: String, port: u32) -> Result<(), SshClientError> {
         if let Some(session) = &self.session {
             let mut session = session.lock().await;

warpgate-protocol-ssh/src/common.rs

@@ -55,6 +55,7 @@ pub struct X11Request {
 pub enum ChannelOperation {
     OpenShell,
     OpenDirectTCPIP(DirectTCPIPParams),
+    OpenDirectStreamlocal(String),
     OpenX11(String, u32),
     RequestPty(PtyRequest),
     ResizePty(PtyRequest),

warpgate-protocol-ssh/src/server/russh_handler.rs

@@ -43,6 +43,7 @@ pub enum ServerHandlerEvent {
     Signal(ServerChannelId, Sig, oneshot::Sender<()>),
     ExecRequest(ServerChannelId, Bytes, oneshot::Sender<bool>),
     ChannelOpenDirectTcpIp(ServerChannelId, DirectTCPIPParams, oneshot::Sender<bool>),
+    ChannelOpenDirectStreamlocal(ServerChannelId, String, oneshot::Sender<bool>),
     EnvRequest(ServerChannelId, String, String, oneshot::Sender<()>),
     X11Request(ServerChannelId, X11Request, oneshot::Sender<()>),
     TcpIpForward(String, u32, oneshot::Sender<bool>),
@@ -416,6 +417,23 @@ impl russh::server::Handler for ServerHandler {
         Ok(allowed)
     }
 
+    async fn channel_open_direct_streamlocal(
+        &mut self,
+        channel: Channel<Msg>,
+        socket_path: &str,
+        _session: &mut Session,
+    ) -> Result<bool, Self::Error> {
+        let socket_path = socket_path.to_string();
+        let (tx, rx) = oneshot::channel();
+        self.send_event(ServerHandlerEvent::ChannelOpenDirectStreamlocal(
+            ServerChannelId(channel.id()),
+            socket_path,
+            tx,
+        ))?;
+        let allowed = rx.await.unwrap_or(false);
+        Ok(allowed)
+    }
+
     async fn x11_request(
         &mut self,
         channel: ChannelId,

warpgate-protocol-ssh/src/server/session.rs

@@ -551,6 +551,10 @@ impl ServerSession {
                 let _ = reply.send(self._channel_open_direct_tcpip(channel, params).await?);
             }
 
+            ServerHandlerEvent::ChannelOpenDirectStreamlocal(channel, path, reply) => {
+                let _ = reply.send(self._channel_open_direct_streamlocal(channel, path).await?);
+            }
+
             ServerHandlerEvent::EnvRequest(channel, name, value, reply) => {
                 self._channel_env_request(channel, name, value).await?;
                 let _ = reply.send(());
@@ -1027,6 +1031,52 @@ impl ServerSession {
         }
     }
 
+    async fn _channel_open_direct_streamlocal(
+        &mut self,
+        channel: ServerChannelId,
+        path: String,
+    ) -> Result<bool> {
+        let uuid = Uuid::new_v4();
+        self.channel_map.insert(channel, uuid);
+
+        info!(%channel, "Opening direct streamlocal channel to {}", path);
+
+        let _ = self.maybe_connect_remote().await;
+
+        match self
+            .send_command_and_wait(RCCommand::Channel(
+                uuid,
+                ChannelOperation::OpenDirectStreamlocal(path.clone()),
+            ))
+            .await
+        {
+            Ok(()) => {
+                self.all_channels.push(uuid);
+
+                let recorder = self
+                    .traffic_recorder_for(
+                        TrafficRecorderKey::Socket(path.clone()),
+                        "direct-tcpip",
+                    )
+                    .await;
+                if let Some(recorder) = recorder {
+                    #[allow(clippy::unwrap_used)]
+                    let mut recorder = recorder.connection(TrafficConnectionParams::Socket {
+                        socket_path: path,
+                    });
+                    if let Err(error) = recorder.write_connection_setup().await {
+                        error!(%channel, ?error, "Failed to record connection setup");
+                    }
+                    self.traffic_connection_recorders.insert(uuid, recorder);
+                }
+
+                Ok(true)
+            }
+            Err(SshClientError::Russh(russh::Error::ChannelOpenFailure(_))) => Ok(false),
+            Err(x) => Err(x.into()),
+        }
+    }
+
     async fn _window_change_request(
         &mut self,
         server_channel_id: ServerChannelId,