feat(agent): RDM messages and pipe passthrough logic

crates/devolutions-agent-shared/src/lib.rs

@@ -45,6 +45,7 @@ pub struct PackageInfoError {
 pub fn get_installed_agent_version() -> Result<Option<DateVersion>, PackageInfoError> {
     Ok(windows::registry::get_installed_product_version(
         windows::AGENT_UPDATE_CODE,
+        windows::registry::ProductVersionEncoding::Agent,
     )?)
 }
 

crates/devolutions-agent-shared/src/windows/mod.rs

@@ -17,7 +17,13 @@ pub const GATEWAY_UPDATE_CODE: Uuid = uuid!("{db3903d6-c451-4393-bd80-eb9f45b902
 ///
 /// See [`GATEWAY_UPDATE_CODE`] for more information on update codes.
 pub const AGENT_UPDATE_CODE: Uuid = uuid!("{82318d3c-811f-4d5d-9a82-b7c31b076755}");
+
 /// MSI upgrade code for the Devolutions Hub Service.
 ///
 /// See [`GATEWAY_UPDATE_CODE`] for more information on update codes.
 pub const HUB_SERVICE_UPDATE_CODE: Uuid = uuid!("{f437046e-8e13-430a-8c8f-29fcb9023b59}");
+
+/// MSI upgrade code for the Remote Desktop Manager.
+///
+/// See [`GATEWAY_UPDATE_CODE`] for more information on update codes.
+pub const RDM_UPDATE_CODE: Uuid = uuid!("{2707F3BF-4D7B-40C2-882F-14B0ED869EE8}");

crates/devolutions-agent-shared/src/windows/registry.rs

@@ -48,9 +48,17 @@ pub fn get_product_code(update_code: Uuid) -> Result<Option<Uuid>, RegistryError
     Ok(Some(reversed_hex_to_uuid(&product_code)?))
 }
 
+pub enum ProductVersionEncoding {
+    Agent,
+    Rdm,
+}
+
 /// Get the installed version of a product using Windows registry. Returns `None` if the product
 /// is not installed.
-pub fn get_installed_product_version(update_code: Uuid) -> Result<Option<DateVersion>, RegistryError> {
+pub fn get_installed_product_version(
+    update_code: Uuid,
+    version_encoding: ProductVersionEncoding,
+) -> Result<Option<DateVersion>, RegistryError> {
     let product_code_uuid = match get_product_code(update_code)? {
         Some(uuid) => uuid,
         None => return Ok(None),
@@ -79,7 +87,14 @@ pub fn get_installed_product_version(update_code: Uuid) -> Result<Option<DateVer
         })?;
 
     // Convert encoded MSI version number to human-readable date.
-    let short_year = (product_version >> 24) + 2000;
+    // The high byte encodes the year as an offset:
+    // - Agent builds use a base year of 2000 (year = high_byte + 2000).
+    // - RDM MSI packages use 0x700 (1792) as base, found empirically.
+    //   This offset must be preserved to correctly decode existing RDM installations.
+    let short_year = match version_encoding {
+        ProductVersionEncoding::Agent => (product_version >> 24) + 2000,
+        ProductVersionEncoding::Rdm => (product_version >> 24) + 0x700,
+    };
     let month = (product_version >> 16) & 0xFF;
     let day = product_version & 0xFFFF;
 
@@ -91,3 +106,36 @@ pub fn get_installed_product_version(update_code: Uuid) -> Result<Option<DateVer
         revision: 0,
     }))
 }
+
+/// Get the installation location of a product using Windows registry. Returns `None` if the product
+/// is not installed or if the InstallLocation value is not present.
+pub fn get_install_location(update_code: Uuid) -> Result<Option<String>, RegistryError> {
+    let product_code_uuid = match get_product_code(update_code)? {
+        Some(uuid) => uuid,
+        None => return Ok(None),
+    }
+    .braced();
+
+    let key_path = format!("{REG_CURRENT_VERSION}\\Uninstall\\{product_code_uuid}");
+
+    const INSTALL_LOCATION_VALUE_NAME: &str = "InstallLocation";
+
+    // Now we know the product code of installed MSI, we could read its install location.
+    let product_tree = windows_registry::LOCAL_MACHINE
+        .open(&key_path)
+        .map_err(|source| RegistryError::OpenKey {
+            key: key_path.clone(),
+            source,
+        })?;
+
+    let install_location: String = product_tree
+        .get_value(INSTALL_LOCATION_VALUE_NAME)
+        .and_then(TryInto::try_into)
+        .map_err(|source| RegistryError::ReadValue {
+            value: INSTALL_LOCATION_VALUE_NAME.to_owned(),
+            key: key_path.clone(),
+            source,
+        })?;
+
+    Ok(Some(install_location))
+}

crates/win-api-wrappers/src/event.rs

@@ -1,9 +1,12 @@
 use std::sync::Arc;
 
-use windows::Win32::Foundation::HANDLE;
-use windows::Win32::System::Threading::{CreateEventW, SetEvent};
+use anyhow::bail;
+use windows::Win32::Foundation::{HANDLE, WAIT_FAILED, WAIT_OBJECT_0, WAIT_TIMEOUT};
+use windows::Win32::System::Threading::{CreateEventW, INFINITE, SetEvent, WaitForSingleObject};
 
+use crate::Error;
 use crate::handle::Handle;
+use crate::utils::WideString;
 
 /// RAII wrapper for WinAPI event handle.
 #[derive(Debug, Clone)]
@@ -24,6 +27,27 @@ impl Event {
         })
     }
 
+    pub fn new_named(name: &str, manual_reset: bool, initial_state: bool) -> anyhow::Result<Self> {
+        let name_wide = WideString::from(name);
+
+        // SAFETY: name_wide is a valid null-terminated UTF-16 string
+        let raw_handle = unsafe {
+            CreateEventW(
+                None,                  // Default security
+                manual_reset,          // Manual or auto-reset
+                initial_state,         // Initially signaled or not
+                name_wide.as_pcwstr(), // Event name
+            )
+        }?;
+
+        // SAFETY: `CreateEventW` always returns a valid handle on success.
+        let handle = unsafe { Handle::new_owned(raw_handle) }?;
+
+        Ok(Self {
+            handle: Arc::new(handle),
+        })
+    }
+
     pub fn raw(&self) -> HANDLE {
         self.handle.raw()
     }
@@ -35,4 +59,16 @@ impl Event {
         }
         Ok(())
     }
+
+    pub fn wait(&self, timeout_ms: Option<u32>) -> anyhow::Result<()> {
+        // SAFETY: No preconditions.
+        let status = unsafe { WaitForSingleObject(self.handle.raw(), timeout_ms.unwrap_or(INFINITE)) };
+
+        match status {
+            WAIT_OBJECT_0 => Ok(()),
+            WAIT_TIMEOUT => bail!("Timeout waiting for event"),
+            WAIT_FAILED => bail!(Error::last_error()),
+            _ => bail!("Unexpected wait result"),
+        }
+    }
 }

crates/win-api-wrappers/src/process.rs

@@ -24,10 +24,10 @@ use windows::Win32::System::LibraryLoader::{
 use windows::Win32::System::RemoteDesktop::ProcessIdToSessionId;
 use windows::Win32::System::Threading::{
     CREATE_UNICODE_ENVIRONMENT, CreateProcessAsUserW, CreateRemoteThread, EXTENDED_STARTUPINFO_PRESENT,
-    GetCurrentProcess, GetExitCodeProcess, INFINITE, LPPROC_THREAD_ATTRIBUTE_LIST, LPTHREAD_START_ROUTINE, OpenProcess,
-    OpenProcessToken, PEB, PROCESS_ACCESS_RIGHTS, PROCESS_BASIC_INFORMATION, PROCESS_CREATION_FLAGS,
-    PROCESS_INFORMATION, PROCESS_NAME_WIN32, PROCESS_TERMINATE, QueryFullProcessImageNameW, STARTUPINFOEXW,
-    STARTUPINFOW, STARTUPINFOW_FLAGS, TerminateProcess, WaitForSingleObject,
+    GetCurrentProcess, GetCurrentProcessId, GetExitCodeProcess, INFINITE, LPPROC_THREAD_ATTRIBUTE_LIST,
+    LPTHREAD_START_ROUTINE, OpenProcess, OpenProcessToken, PEB, PROCESS_ACCESS_RIGHTS, PROCESS_BASIC_INFORMATION,
+    PROCESS_CREATION_FLAGS, PROCESS_INFORMATION, PROCESS_NAME_WIN32, PROCESS_TERMINATE, QueryFullProcessImageNameW,
+    STARTUPINFOEXW, STARTUPINFOW, STARTUPINFOW_FLAGS, TerminateProcess, WaitForSingleObject,
 };
 use windows::Win32::UI::Shell::{SEE_MASK_NOCLOSEPROCESS, SHELLEXECUTEINFOW, ShellExecuteExW};
 use windows::Win32::UI::WindowsAndMessaging::{
@@ -902,13 +902,21 @@ fn terminate_process_by_name_impl(process_name: &str, session_id: Option<u32>) -
     Ok(false)
 }
 
-fn process_id_to_session(pid: u32) -> Result<u32> {
+/// Get the Windows session ID for a given process ID.
+pub fn process_id_to_session(pid: u32) -> Result<u32> {
     let mut session_id = 0;
     // SAFETY: `session_id` is always pointing to a valid memory location.
     unsafe { ProcessIdToSessionId(pid, &mut session_id as *mut _) }?;
     Ok(session_id)
 }
 
+/// Get the current Windows session ID.
+pub fn get_current_session_id() -> Result<u32> {
+    // SAFETY: FFI call with no outstanding preconditions.
+    let process_id = unsafe { GetCurrentProcessId() };
+    process_id_to_session(process_id)
+}
+
 struct EnumWindowsContext {
     expected_pid: u32,
     threads: Vec<u32>,

devolutions-agent/src/updater/detect.rs

@@ -9,10 +9,12 @@ use crate::updater::{Product, UpdaterError};
 pub(crate) fn get_installed_product_version(product: Product) -> Result<Option<DateVersion>, UpdaterError> {
     match product {
         Product::Gateway => {
-            registry::get_installed_product_version(GATEWAY_UPDATE_CODE).map_err(UpdaterError::WindowsRegistry)
+            registry::get_installed_product_version(GATEWAY_UPDATE_CODE, registry::ProductVersionEncoding::Agent)
+                .map_err(UpdaterError::WindowsRegistry)
         }
         Product::HubService => {
-            registry::get_installed_product_version(HUB_SERVICE_UPDATE_CODE).map_err(UpdaterError::WindowsRegistry)
+            registry::get_installed_product_version(HUB_SERVICE_UPDATE_CODE, registry::ProductVersionEncoding::Agent)
+                .map_err(UpdaterError::WindowsRegistry)
         }
     }
 }

devolutions-session/Cargo.toml

@@ -47,6 +47,9 @@ optional = true
 version = "0.4.2"
 features = ["std"]
 
+[target.'cfg(windows)'.dependencies]
+devolutions-agent-shared = { path = "../crates/devolutions-agent-shared" }
+
 [target.'cfg(windows)'.build-dependencies]
 embed-resource = "3.0"
 
@@ -60,4 +63,4 @@ features = [
     "Win32_UI_Shell",
     "Win32_System_Console",
     "Win32_UI_Input_KeyboardAndMouse",
-]
+]

devolutions-session/src/dvc/io.rs

@@ -9,7 +9,7 @@ use win_api_wrappers::wts::WtsVirtualChannel;
 use windows::Win32::Foundation::{ERROR_IO_PENDING, GetLastError, WAIT_EVENT, WAIT_OBJECT_0};
 use windows::Win32::Storage::FileSystem::{ReadFile, WriteFile};
 use windows::Win32::System::IO::{GetOverlappedResult, OVERLAPPED};
-use windows::Win32::System::RemoteDesktop::{CHANNEL_CHUNK_LENGTH, CHANNEL_PDU_HEADER};
+use windows::Win32::System::RemoteDesktop::CHANNEL_PDU_HEADER;
 use windows::Win32::System::Threading::{INFINITE, WaitForMultipleObjects};
 
 use crate::dvc::channel::WinapiSignaledReceiver;
@@ -31,7 +31,9 @@ pub fn run_dvc_io(
 
     trace!("DVC channel opened");
 
-    let mut pdu_chunk_buffer = [0u8; CHANNEL_CHUNK_LENGTH as usize];
+    // All DVC messages should be under CHANNEL_CHUNK_LENGTH size, but sometimes RDP stack
+    // sends a few messages together; 128Kb buffer should be enough to hold a few dozen messages.
+    let mut pdu_chunk_buffer = [0u8; 128 * 1024];
     let mut overlapped = OVERLAPPED::default();
     let mut bytes_read: u32 = 0;
 
@@ -111,7 +113,6 @@ pub fn run_dvc_io(
                         }
                     }
                 }
-
                 // Prepare async read file operation one more time.
                 // SAFETY: No preconditions.
                 let result =

devolutions-session/src/dvc/mod.rs

@@ -38,6 +38,7 @@ pub mod fs;
 pub mod io;
 pub mod now_message_dissector;
 pub mod process;
+pub mod rdm;
 pub mod task;
 
 mod env;