cli: allow exec server to listen on a port and require token authentication (#188434)

* cli: allow exec server to listen on a port and require token authentication

For remote ssh on Windows where pipe forwarding doesn't work

* fix linux build

cli/src/async_pipe.rs

@@ -4,7 +4,10 @@
  *--------------------------------------------------------------------------------------------*/
 
 use crate::{constants::APPLICATION_NAME, util::errors::CodeError};
+use async_trait::async_trait;
 use std::path::{Path, PathBuf};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpListener;
 use uuid::Uuid;
 
 // todo: we could probably abstract this into some crate, if one doesn't already exist
@@ -39,7 +42,7 @@ cfg_if::cfg_if! {
 			pipe.into_split()
 		}
 	} else {
-		use tokio::{time::sleep, io::{AsyncRead, AsyncWrite, ReadBuf}};
+		use tokio::{time::sleep, io::ReadBuf};
 		use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions, NamedPipeClient, NamedPipeServer};
 		use std::{time::Duration, pin::Pin, task::{Context, Poll}, io};
 		use pin_project::pin_project;
@@ -181,3 +184,34 @@ pub fn get_socket_name() -> PathBuf {
 		}
 	}
 }
+
+pub type AcceptedRW = (
+	Box<dyn AsyncRead + Send + Unpin>,
+	Box<dyn AsyncWrite + Send + Unpin>,
+);
+
+#[async_trait]
+pub trait AsyncRWAccepter {
+	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError>;
+}
+
+#[async_trait]
+impl AsyncRWAccepter for AsyncPipeListener {
+	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError> {
+		let pipe = self.accept().await?;
+		let (read, write) = socket_stream_split(pipe);
+		Ok((Box::new(read), Box::new(write)))
+	}
+}
+
+#[async_trait]
+impl AsyncRWAccepter for TcpListener {
+	async fn accept_rw(&mut self) -> Result<AcceptedRW, CodeError> {
+		let (stream, _) = self
+			.accept()
+			.await
+			.map_err(CodeError::AsyncPipeListenerFailed)?;
+		let (read, write) = tokio::io::split(stream);
+		Ok((Box::new(read), Box::new(write)))
+	}
+}