refactor(dvc-proxy): refactoring after review

Author: pacmancoder

Cargo.lock

@@ -25,6 +25,9 @@ tracing = { version = "0.1", features = ["log"] }
 
 
 [target.'cfg(windows)'.dependencies]
+
+widestring = "1"
+smallvec = "1"
 windows = { version = "0.61", features = [
     "Win32_Foundation",
     "Win32_Security",

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

@@ -9,4 +9,4 @@ mod windows;
 
 mod platform;
 
-pub use platform::*;
+pub use platform::DvcNamedPipeProxy;

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

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

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

@@ -0,0 +1,48 @@
+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,
+}
+
+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(),
+        }
+    }
+}
+
+impl_as_any!(DvcNamedPipeProxy);
+
+impl DvcProcessor for DvcNamedPipeProxy {
+    fn channel_name(&self) -> &str {
+        &self.channel_name
+    }
+
+    fn start(&mut self, _channel_id: u32) -> PduResult<Vec<DvcMessage>> {
+        Err(pdu_other_err!(
+            "DvcNamedPipeProxy is not implemented on Unix-like systems"
+        ))
+    }
+
+    fn process(&mut self, _channel_id: u32, _payload: &[u8]) -> PduResult<Vec<DvcMessage>> {
+        Err(pdu_other_err!(
+            "DvcNamedPipeProxy is not implemented on Unix-like systems"
+        ))
+    }
+}
+
+impl DvcClientProcessor for DvcNamedPipeProxy {}

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

@@ -3,13 +3,13 @@ mod worker;
 
 use std::sync::mpsc;
 
-use error::DvcPipeProxyError;
 use ironrdp_core::impl_as_any;
 use ironrdp_dvc::{DvcClientProcessor, DvcMessage, DvcProcessor};
 use ironrdp_pdu::{pdu_other_err, PduResult};
 use ironrdp_svc::SvcMessage;
-use worker::{worker_thread_func, OnWriteDvcMessage, WorkerCtx};
 
+use crate::platform::windows::error::DvcPipeProxyError;
+use crate::platform::windows::worker::{worker_thread_func, OnWriteDvcMessage, WorkerCtx};
 use crate::windows::{Event, MessagePipeServer, Semaphore};
 
 const IO_MPSC_CHANNEL_SIZE: usize = 100;
@@ -61,8 +61,8 @@ impl Drop for DvcNamedPipeProxy {
 
 impl DvcNamedPipeProxy {
     fn start_impl(&mut self, channel_id: u32) -> Result<(), DvcPipeProxyError> {
-        // PIPE -> DVC channel - handled via callback passed to the constructor
-        // DVC -> PIPE channel - handled via mpsc internally in the worker thread
+        // PIPE -> DVC channel - handled via callback passed to the constructor.
+        // DVC -> PIPE channel - handled via mpsc internally in the worker thread.
         let (to_pipe_tx, to_pipe_rx) = mpsc::sync_channel(IO_MPSC_CHANNEL_SIZE);
 
         let semaphore_max_count = IO_MPSC_CHANNEL_SIZE
@@ -126,7 +126,6 @@ impl DvcProcessor for DvcNamedPipeProxy {
 
     fn process(&mut self, _channel_id: u32, payload: &[u8]) -> PduResult<Vec<DvcMessage>> {
         // Send the payload to the worker thread via the mpsc channel.
-
         let ctx = match &self.worker_control_ctx {
             Some(ctx) => ctx,
             None => {

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

@@ -8,8 +8,8 @@ use ironrdp_svc::{ChannelFlags, SvcMessage};
 use crate::platform::windows::error::DvcPipeProxyError;
 use crate::windows::{wait_any, wait_any_with_timeout, Event, MessagePipeServer, Semaphore, WindowsError};
 
-const PIPE_CONNECT_TIMEOUT: u32 = 10_000; // 10 seconds
-const PIPE_WRITE_TIMEOUT: u32 = 3_000; // 3 seconds
+const PIPE_CONNECT_TIMEOUT_SECS: u32 = 10_000; // 10 seconds
+const PIPE_WRITE_TIMEOUT_SECS: u32 = 3_000; // 3 seconds
 const MESSAGE_BUFFER_SIZE: usize = 64 * 1024; // 64 KiB
 
 pub(crate) type OnWriteDvcMessage = Box<dyn Fn(u32, Vec<SvcMessage>) -> PduResult<()> + Send>;
@@ -44,9 +44,8 @@ pub(crate) fn worker_thread_func(worker_ctx: WorkerCtx) -> Result<(), DvcPipePro
 
         if !connect_ctx.overlapped_connect()? {
             const EVENT_ID_ABORT: usize = 0;
-            let events = &[abort_event.raw(), connect_ctx.event().raw()];
-
-            let wait_result = match wait_any_with_timeout(events, PIPE_CONNECT_TIMEOUT) {
+            let events = [abort_event.borrow(), connect_ctx.borrow_event()];
+            let wait_result = match wait_any_with_timeout(events, PIPE_CONNECT_TIMEOUT_SECS) {
                 Ok(idx) => idx,
                 Err(WindowsError::WaitForMultipleObjectsTimeout) => {
                     warn!(%channel_name, %pipe_name, "DVC pipe proxy connection timed out");
@@ -73,22 +72,25 @@ pub(crate) fn worker_thread_func(worker_ctx: WorkerCtx) -> Result<(), DvcPipePro
     const EVENT_ID_ABORT: usize = 0;
     const EVENT_ID_READ: usize = 1;
     const EVENT_ID_WRITE_MPSC: usize = 2;
-    let events = &[abort_event.raw(), read_ctx.event().raw(), to_pipe_semaphore.raw()];
 
     read_ctx.overlapped_read()?;
 
     info!(%channel_name, %pipe_name, "DVC pipe proxy IO loop started");
 
     loop {
+        let events = [
+            abort_event.borrow(),
+            read_ctx.borrow_event(),
+            to_pipe_semaphore.borrow(),
+        ];
         let wait_result = wait_any(events)?;
 
-        // abort event
         if wait_result == EVENT_ID_ABORT {
             info!(%channel_name, %pipe_name, "DVC pipe proxy connection has been aborted");
             return Ok(());
         }
 
-        // read from pipe
+        // Read end of pipe is ready, forward received data to DVC.
         if wait_result == EVENT_ID_READ {
             let read_result = read_ctx.get_result()?.to_vec();
 
@@ -107,12 +109,12 @@ pub(crate) fn worker_thread_func(worker_ctx: WorkerCtx) -> Result<(), DvcPipePro
                 }
             }
 
-            // Queue the read operation again
+            // Queue the read operation again.
             read_ctx.overlapped_read()?;
             continue;
         }
 
-        // read from mpsc and write to pipe
+        // DVC data received, forward it to the pipe.
         if wait_result == EVENT_ID_WRITE_MPSC {
             let payload = to_pipe_rx.recv().map_err(|_| DvcPipeProxyError::MpscIo)?;
 
@@ -125,25 +127,22 @@ pub(crate) fn worker_thread_func(worker_ctx: WorkerCtx) -> Result<(), DvcPipePro
 
             trace!(%channel_name, %pipe_name, "DVC proxy write {} bytes to pipe,", payload_len);
 
-            // write to pipe
             let mut overlapped_write = pipe.prepare_write_overlapped(payload)?;
 
-            let events = &[abort_event.raw(), overlapped_write.event().raw()];
-
             overlapped_write.overlapped_write()?;
-            let wait_result = wait_any_with_timeout(events, PIPE_WRITE_TIMEOUT)?;
 
-            // abort event
+            let events = [abort_event.borrow(), overlapped_write.borrow_event()];
+            let wait_result = wait_any_with_timeout(events, PIPE_WRITE_TIMEOUT_SECS)?;
+
             if wait_result == EVENT_ID_ABORT {
                 info!(%channel_name, %pipe_name, "DVC pipe proxy write aborted");
                 return Ok(());
             }
 
-            // write to pipe
             let bytes_written = overlapped_write.get_result()?;
 
             if bytes_written as usize != payload_len {
-                // Message-based pipe write failed
+                // Message-based pipe write failed.
                 return Err(DvcPipeProxyError::DvcIncompleteWrite);
             }
 

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

@@ -12,45 +12,27 @@ pub(crate) enum WindowsError {
     OverlappedRead(windows::core::Error),
     OverlappedWrite(windows::core::Error),
     CreateSemaphore(windows::core::Error),
-    InvalidHandle,
+    InvalidPipeName(String),
 }
 
 impl core::fmt::Display for WindowsError {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         match self {
-            WindowsError::CreateNamedPipe(_) => write!(f, "CreateNamedPipe failed"),
-            WindowsError::CreateEvent(_) => write!(f, "CreateEvent failed"),
-            WindowsError::SetEvent(_) => write!(f, "SetEvent failed"),
-            WindowsError::InvalidSemaphoreParams(cause) => {
-                write!(f, "Invalid semaphore parameters: {}", cause)
-            }
-            WindowsError::ReleaseSemaphore(_) => {
-                write!(f, "ReleaseSemaphore failed")
-            }
-            WindowsError::WaitForMultipleObjectsFailed(_) => {
-                write!(f, "WaitForMultipleObjects failed")
-            }
-            WindowsError::WaitForMultipleObjectsTimeout => {
-                write!(f, "WaitForMultipleObjects timed out")
-            }
+            WindowsError::CreateNamedPipe(_) => write!(f, "failed to create named pipe"),
+            WindowsError::CreateEvent(_) => write!(f, "failed to create event object"),
+            WindowsError::SetEvent(_) => write!(f, "failed to set event to signaled state"),
+            WindowsError::InvalidSemaphoreParams(cause) => write!(f, "invalid semaphore parameters: {}", cause),
+            WindowsError::ReleaseSemaphore(_) => write!(f, "failed to release semaphore"),
+            WindowsError::WaitForMultipleObjectsFailed(_) => write!(f, "failed to wait for multiple objects"),
+            WindowsError::WaitForMultipleObjectsTimeout => write!(f, "timed out waiting for multiple objects"),
             WindowsError::WaitForMultipleObjectsAbandoned(idx) => {
-                write!(f, "WaitForMultipleObjects handle #{idx} was abandoned")
-            }
-            WindowsError::OverlappedConnect(_) => {
-                write!(f, "Overlapped connect failed")
-            }
-            WindowsError::OverlappedRead(_) => {
-                write!(f, "Overlapped read failed")
-            }
-            WindowsError::OverlappedWrite(_) => {
-                write!(f, "Overlapped write failed")
-            }
-            WindowsError::CreateSemaphore(_) => {
-                write!(f, "CreateSemaphore failed")
-            }
-            WindowsError::InvalidHandle => {
-                write!(f, "Invalid handle")
+                write!(f, "wait for multiple objects failed, handle #{idx} was abandoned")
             }
+            WindowsError::OverlappedConnect(_) => write!(f, "overlapped connect failed"),
+            WindowsError::OverlappedRead(_) => write!(f, "overlapped read failed"),
+            WindowsError::OverlappedWrite(_) => write!(f, "overlapped write failed"),
+            WindowsError::CreateSemaphore(_) => write!(f, "failed to create semaphore object"),
+            WindowsError::InvalidPipeName(cause) => write!(f, "invalid pipe name: `{}`", cause),
         }
     }
 }
@@ -69,8 +51,8 @@ impl core::error::Error for WindowsError {
             WindowsError::CreateEvent(err) => Some(err),
             WindowsError::InvalidSemaphoreParams(_)
             | WindowsError::WaitForMultipleObjectsTimeout
-            | WindowsError::InvalidHandle
-            | WindowsError::WaitForMultipleObjectsAbandoned(_) => None,
+            | WindowsError::InvalidPipeName(_) => None,
+            WindowsError::WaitForMultipleObjectsAbandoned(_) => None,
         }
     }
 }

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

@@ -1,41 +1,53 @@
 use std::sync::Arc;
 
+use windows::core::Owned;
 use windows::Win32::Foundation::HANDLE;
 use windows::Win32::System::Threading::{CreateEventW, SetEvent};
 
-use crate::windows::{Handle, WindowsError};
+use crate::windows::{BorrowedHandle, WindowsError};
 
 /// RAII wrapper for WinAPI event handle.
 #[derive(Debug, Clone)]
 pub(crate) struct Event {
-    handle: Arc<Handle>,
+    handle: Arc<Owned<HANDLE>>,
 }
 
-/// SAFETY: It is safe to send event HANDLE between threads.
+// SAFETY: We ensure that inner handle is indeed could be sent and shared between threads via
+// Event wrapper API itself by restricting handle usage:
+// - set() method which calls SetEvent inside (which is thread-safe).
+// - borrow() method which returns a BorrowedHandle for waiting on the event.
+// - Handle lifetime is ensured by Arc, so it is always valid when used.
 unsafe impl Send for Event {}
 
 impl Event {
     pub(crate) fn new_unnamed() -> Result<Self, WindowsError> {
         // SAFETY: FFI call with no outstanding preconditions.
         let handle = unsafe { CreateEventW(None, false, false, None).map_err(WindowsError::CreateEvent)? };
+
         // SAFETY: Handle is valid and we are the owner of the handle.
-        let handle = unsafe { Handle::new_owned(handle)? };
+        let handle = unsafe { Owned::new(handle) };
 
         // CreateEventW returns a valid handle on success.
         Ok(Self {
+            // See `unsafe impl Send` comment.
+            #[allow(clippy::arc_with_non_send_sync)]
             handle: Arc::new(handle),
         })
     }
 
     pub(crate) fn set(&self) -> Result<(), WindowsError> {
         // SAFETY: The handle is valid and we are the owner of the handle.
         unsafe {
-            SetEvent(self.handle.raw()).map_err(WindowsError::SetEvent)?;
+            SetEvent(self.raw()).map_err(WindowsError::SetEvent)?;
         }
         Ok(())
     }
 
-    pub(crate) fn raw(&self) -> HANDLE {
-        self.handle.raw()
+    pub(super) fn raw(&self) -> HANDLE {
+        **self.handle
+    }
+
+    pub(crate) fn borrow(&self) -> BorrowedHandle<'_> {
+        BorrowedHandle(&self.handle)
     }
 }