feat(dvc): make dvc named pipe proxy cross-platform

crates/ironrdp-dvc-pipe-proxy/Cargo.toml

@@ -22,17 +22,8 @@ ironrdp-dvc = { path = "../ironrdp-dvc", version = "0.3" }
 ironrdp-svc = { path = "../ironrdp-svc", version = "0.4" } # public (SvcMessage type)
 
 tracing = { version = "0.1", features = ["log"] }
-
-[target.'cfg(windows)'.dependencies]
-widestring = "1"
-windows = { version = "0.61", features = [
-    "Win32_Foundation",
-    "Win32_Security",
-    "Win32_System_Threading",
-    "Win32_Storage_FileSystem",
-    "Win32_System_Pipes",
-    "Win32_System_IO",
-] }
+tokio = { version = "1", features = ["net", "rt", "sync", "macros", "io-util"]}
+async-trait = "0.1"
 
 [lints]
 workspace = true

crates/ironrdp-dvc-pipe-proxy/src/error.rs

@@ -0,0 +1,23 @@
+#[derive(Debug)]
+pub(crate) enum DvcPipeProxyError {
+    Io(std::io::Error),
+    EncodeDvcMessage(ironrdp_core::EncodeError),
+}
+
+impl core::fmt::Display for DvcPipeProxyError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            DvcPipeProxyError::Io(_) => write!(f, "IO error"),
+            DvcPipeProxyError::EncodeDvcMessage(_) => write!(f, "DVC message encoding error"),
+        }
+    }
+}
+
+impl core::error::Error for DvcPipeProxyError {
+    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
+        match self {
+            DvcPipeProxyError::Io(err) => Some(err),
+            DvcPipeProxyError::EncodeDvcMessage(src) => Some(src),
+        }
+    }
+}

crates/ironrdp-dvc-pipe-proxy/src/lib.rs

@@ -4,8 +4,11 @@
 #[macro_use]
 extern crate tracing;
 
-#[cfg(target_os = "windows")]
-mod windows;
-
+mod error;
+mod message;
+mod os_pipe;
 mod platform;
-pub use self::platform::DvcNamedPipeProxy;
+mod proxy;
+mod worker;
+
+pub use self::proxy::DvcNamedPipeProxy;

crates/ironrdp-dvc-pipe-proxy/src/message.rs

@@ -0,0 +1,22 @@
+use ironrdp_core::{ensure_size, Encode, EncodeResult};
+use ironrdp_dvc::DvcEncode;
+
+pub(crate) struct RawDataDvcMessage(pub Vec<u8>);
+
+impl Encode for RawDataDvcMessage {
+    fn encode(&self, dst: &mut ironrdp_core::WriteCursor<'_>) -> EncodeResult<()> {
+        ensure_size!(in: dst, size: self.size());
+        dst.write_slice(&self.0);
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "RawDataDvcMessage"
+    }
+
+    fn size(&self) -> usize {
+        self.0.len()
+    }
+}
+
+impl DvcEncode for RawDataDvcMessage {}

crates/ironrdp-dvc-pipe-proxy/src/os_pipe.rs

@@ -0,0 +1,19 @@
+use async_trait::async_trait;
+
+use crate::error::DvcPipeProxyError;
+
+#[async_trait]
+pub(crate) trait OsPipe: Send + Sync {
+    /// Creates a new OS pipe and waits for the connection.
+    async fn connect(pipe_name: &str) -> Result<Self, DvcPipeProxyError>
+    where
+        Self: Sized;
+
+    /// Reads data from the pipe and returns the number of bytes read.
+    ///
+    /// Returned future should be stateless and can be polled multiple times.
+    async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, DvcPipeProxyError>;
+
+    /// Writes data to the pipe and returns the number of bytes written.
+    async fn write_all(&mut self, buffer: &[u8]) -> Result<(), DvcPipeProxyError>;
+}

crates/ironrdp-dvc-pipe-proxy/src/platform/mod.rs

@@ -1,9 +1,5 @@
 #[cfg(target_os = "windows")]
-mod windows;
-#[cfg(target_os = "windows")]
-pub use self::windows::DvcNamedPipeProxy;
+pub(crate) mod windows;
 
 #[cfg(not(target_os = "windows"))]
-mod unix;
-#[cfg(not(target_os = "windows"))]
-pub use self::unix::DvcNamedPipeProxy;
+pub(crate) mod unix;

crates/ironrdp-dvc-pipe-proxy/src/platform/unix.rs

@@ -1,48 +1,66 @@
-use ironrdp_core::impl_as_any;
-use ironrdp_dvc::{DvcClientProcessor, DvcMessage, DvcProcessor};
-use ironrdp_pdu::{pdu_other_err, PduResult};
-use ironrdp_svc::SvcMessage;
-
-/// A proxy DVC pipe client that forwards DVC messages to/from a named pipe server.
-pub struct DvcNamedPipeProxy {
-    channel_name: String,
+use async_trait::async_trait;
+use tokio::fs;
+use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+use crate::error::DvcPipeProxyError;
+use crate::os_pipe::OsPipe;
+
+/// Unix-specific implementation of the OS pipe trait.
+pub(crate) struct UnixPipe {
+    socket: tokio::net::UnixStream,
 }
 
-impl DvcNamedPipeProxy {
-    /// Creates a new DVC named pipe proxy.
-    /// `dvc_write_callback` is called when the proxy receives a DVC message from the
-    /// named pipe server and the SVC message is ready to be sent to the DVC channel in the main
-    /// IronRDP active session loop.
-    pub fn new<F>(channel_name: &str, _named_pipe_name: &str, _dvc_write_callback: F) -> Self
-    where
-        F: Fn(u32, Vec<SvcMessage>) -> PduResult<()> + Send + 'static,
-    {
-        error!("DvcNamedPipeProxy is not implemented on Unix-like systems, using a stub implementation");
-
-        Self {
-            channel_name: channel_name.to_owned(),
+#[async_trait]
+impl OsPipe for UnixPipe {
+    async fn connect(pipe_name: &str) -> Result<Self, DvcPipeProxyError> {
+        // Domain socket file could already exist from a previous run.
+        match fs::metadata(&pipe_name).await {
+            Ok(metadata) => {
+                use std::os::unix::fs::FileTypeExt as _;
+
+                info!(
+                    %pipe_name,
+                    "DVC pipe already exists, removing stale file."
+                );
+
+                // Just to be sure, check if it's indeed a socket -
+                // throw an error if calling code accidentally passed a regular file.
+                if !metadata.file_type().is_socket() {
+                    return Err(DvcPipeProxyError::Io(std::io::Error::new(
+                        std::io::ErrorKind::InvalidInput,
+                        format!("Path {pipe_name} is not a socket"),
+                    )));
+                }
+
+                fs::remove_file(pipe_name).await.map_err(DvcPipeProxyError::Io)?;
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+                trace!(
+                    %pipe_name,
+                    "DVC pipe does not exist, creating it."
+                );
+            }
+            Err(e) => {
+                return Err(DvcPipeProxyError::Io(e));
+            }
         }
-    }
-}
 
-impl_as_any!(DvcNamedPipeProxy);
+        let listener = tokio::net::UnixListener::bind(pipe_name).map_err(DvcPipeProxyError::Io)?;
 
-impl DvcProcessor for DvcNamedPipeProxy {
-    fn channel_name(&self) -> &str {
-        &self.channel_name
+        let (socket, _) = listener.accept().await.map_err(DvcPipeProxyError::Io)?;
+
+        Ok(Self { socket })
     }
 
-    fn start(&mut self, _channel_id: u32) -> PduResult<Vec<DvcMessage>> {
-        Err(pdu_other_err!(
-            "DvcNamedPipeProxy is not implemented on Unix-like systems"
-        ))
+    async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, DvcPipeProxyError> {
+        self.socket.read(buffer).await.map_err(DvcPipeProxyError::Io)
     }
 
-    fn process(&mut self, _channel_id: u32, _payload: &[u8]) -> PduResult<Vec<DvcMessage>> {
-        Err(pdu_other_err!(
-            "DvcNamedPipeProxy is not implemented on Unix-like systems"
-        ))
+    async fn write_all(&mut self, buffer: &[u8]) -> Result<(), DvcPipeProxyError> {
+        self.socket
+            .write_all(buffer)
+            .await
+            .map_err(DvcPipeProxyError::Io)
+            .map(|_| ())
     }
 }
-
-impl DvcClientProcessor for DvcNamedPipeProxy {}

crates/ironrdp-dvc-pipe-proxy/src/platform/windows.rs

@@ -0,0 +1,47 @@
+use async_trait::async_trait;
+use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+use tokio::net::windows::named_pipe;
+
+use crate::error::DvcPipeProxyError;
+use crate::os_pipe::OsPipe;
+
+const PIPE_BUFFER_SIZE: u32 = 64 * 1024;
+
+/// Unix-specific implementation of the OS pipe trait.
+pub(crate) struct WindowsPipe {
+    pipe_server: named_pipe::NamedPipeServer,
+}
+
+#[async_trait]
+impl OsPipe for WindowsPipe {
+    async fn connect(pipe_name: &str) -> Result<Self, DvcPipeProxyError> {
+        let pipe_name = format!("\\\\.\\pipe\\{pipe_name}");
+
+        let pipe_server = named_pipe::ServerOptions::new()
+            .first_pipe_instance(true)
+            .access_inbound(true)
+            .access_outbound(true)
+            .max_instances(2)
+            .in_buffer_size(PIPE_BUFFER_SIZE)
+            .out_buffer_size(PIPE_BUFFER_SIZE)
+            .pipe_mode(named_pipe::PipeMode::Message)
+            .create(pipe_name)
+            .map_err(DvcPipeProxyError::Io)?;
+
+        pipe_server.connect().await.map_err(DvcPipeProxyError::Io)?;
+
+        Ok(Self { pipe_server })
+    }
+
+    async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, DvcPipeProxyError> {
+        self.pipe_server.read(buffer).await.map_err(DvcPipeProxyError::Io)
+    }
+
+    async fn write_all(&mut self, buffer: &[u8]) -> Result<(), DvcPipeProxyError> {
+        self.pipe_server
+            .write_all(buffer)
+            .await
+            .map_err(DvcPipeProxyError::Io)
+            .map(|_| ())
+    }
+}