diff --git a/warpgate-protocol-ssh/src/client/channel_session.rs b/warpgate-protocol-ssh/src/client/channel_session.rs index a11377c4..d93d8250 100644 --- a/warpgate-protocol-ssh/src/client/channel_session.rs +++ b/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( diff --git a/warpgate-protocol-ssh/src/client/mod.rs b/warpgate-protocol-ssh/src/client/mod.rs index d2b00ced..ac4d7380 100644 --- a/warpgate-protocol-ssh/src/client/mod.rs +++ b/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; diff --git a/warpgate-protocol-ssh/src/common.rs b/warpgate-protocol-ssh/src/common.rs index 9b463350..9207aaba 100644 --- a/warpgate-protocol-ssh/src/common.rs +++ b/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), diff --git a/warpgate-protocol-ssh/src/server/russh_handler.rs b/warpgate-protocol-ssh/src/server/russh_handler.rs index 467a2ab5..518e7f5c 100644 --- a/warpgate-protocol-ssh/src/server/russh_handler.rs +++ b/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), ChannelOpenDirectTcpIp(ServerChannelId, DirectTCPIPParams, oneshot::Sender), + ChannelOpenDirectStreamlocal(ServerChannelId, String, oneshot::Sender), EnvRequest(ServerChannelId, String, String, oneshot::Sender<()>), X11Request(ServerChannelId, X11Request, oneshot::Sender<()>), TcpIpForward(String, u32, oneshot::Sender), @@ -416,6 +417,23 @@ impl russh::server::Handler for ServerHandler { Ok(allowed) } + async fn channel_open_direct_streamlocal( + &mut self, + channel: Channel, + socket_path: &str, + _session: &mut Session, + ) -> Result { + 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, diff --git a/warpgate-protocol-ssh/src/server/session.rs b/warpgate-protocol-ssh/src/server/session.rs index 08417354..c6f574ee 100644 --- a/warpgate-protocol-ssh/src/server/session.rs +++ b/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 { + 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,