diff --git a/Cargo.lock b/Cargo.lock index f9e43bfc7..ed37f9f10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2064,7 +2064,7 @@ dependencies = [ "hyper-util", "ironrdp-acceptor", "ironrdp-connector", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "ironrdp-rdcleanpath", "ironrdp-tokio", @@ -3858,7 +3858,7 @@ checksum = "30885a3bdbd0fcf2b9aa27ae9298d8858b9e101c4ac795d3d47ff917efaee63f" dependencies = [ "ironrdp-async", "ironrdp-connector", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "ironrdp-svc", "tracing", @@ -3871,7 +3871,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "98ee3ba1d2c7cbe0769c5e552c9c7fbc68aa594d1f74f755c63a544a2176b974" dependencies = [ "bitflags 2.13.1", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-dvc", "num-derive", "num-traits", @@ -3885,7 +3885,7 @@ checksum = "bea12d80384007fe321eb6b36c9225be9559c655fd581458a6a1386c1774d729" dependencies = [ "bytes 1.12.1", "ironrdp-connector", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "tracing", ] @@ -3897,7 +3897,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb9050999a1e032f4313788ac5a6d06e897a4f672f1de2ed98683001fc1abf56" dependencies = [ "bitflags 2.13.1", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "ironrdp-svc", "tracing", @@ -3909,8 +3909,8 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d5898b3f1fcaca0f9b923b1463e158aeb64dfdec4ac361b7c086492466ff341c" dependencies = [ - "ironrdp-core 0.2.1", - "ironrdp-error 0.2.0", + "ironrdp-core", + "ironrdp-error", "ironrdp-pdu", "ironrdp-svc", "picky", @@ -3922,22 +3922,13 @@ dependencies = [ "url", ] -[[package]] -name = "ironrdp-core" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2db60a59716a84d09040d29c9e75e81545842510fccb0934c09b28e78b46680" -dependencies = [ - "ironrdp-error 0.1.3", -] - [[package]] name = "ironrdp-core" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef0875b98275068b88652e49ba2e0a1150a1390aad69d26c942c8c59e8ec918e" dependencies = [ - "ironrdp-error 0.2.0", + "ironrdp-error", ] [[package]] @@ -3946,7 +3937,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74a111a6fd25abbe51b6a15276fe980c6b9eee50a1421224b0c3f45848d45bcf" dependencies = [ - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-dvc", "ironrdp-pdu", "ironrdp-svc", @@ -3959,7 +3950,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a5de64988ddabf96928e2f042e1772a5f7201f0001ea99012129c73a105fc52" dependencies = [ - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "ironrdp-svc", "tracing", @@ -3971,18 +3962,12 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ed0c8de080a88b9a3c5dd7978c4c4ca46a5ddbc81315b799000938f743055c0" dependencies = [ - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-dvc", "ironrdp-pdu", "tracing", ] -[[package]] -name = "ironrdp-error" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4a9d7794e854eef2f13fdf79c8502bcc567a75a15fd0522885f37739386a4cef" - [[package]] name = "ironrdp-error" version = "0.2.0" @@ -3999,7 +3984,7 @@ dependencies = [ "bitflags 2.13.1", "bitvec", "byteorder", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "num-derive", "num-traits", @@ -4016,8 +4001,8 @@ dependencies = [ "bitflags 2.13.1", "byteorder", "der-parser 10.0.0", - "ironrdp-core 0.2.1", - "ironrdp-error 0.2.0", + "ironrdp-core", + "ironrdp-error", "md-5 0.10.6", "num-bigint 0.4.8", "num-derive", @@ -4045,7 +4030,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d1a4f7092ae8fdc0a2f61be8e100277dc6f4ad51b85455dc9e8157979993ea2" dependencies = [ "bitflags 2.13.1", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", "ironrdp-svc", "tracing", @@ -4064,7 +4049,7 @@ dependencies = [ "ironrdp-ainput", "ironrdp-async", "ironrdp-cliprdr", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-displaycontrol", "ironrdp-dvc", "ironrdp-echo", @@ -4090,7 +4075,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24c36b82ab0f7fef2668fb7004008a0f3c100a3d9a7b18bd4495a71aba55b796" dependencies = [ "bitflags 2.13.1", - "ironrdp-core 0.2.1", + "ironrdp-core", "ironrdp-pdu", ] @@ -5291,13 +5276,13 @@ dependencies = [ [[package]] name = "now-proto-pdu" -version = "0.4.4" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1fbbab671b284dfd3e1a81a84da2cd1f4d77e7da675498e59e55de69c713545" +checksum = "aa69fc8f70d8c88ca5e4d3cc30fe32a4a0477cbc529e1d9c35fb6f320e515e9e" dependencies = [ "bitflags 2.13.1", - "ironrdp-core 0.1.5", - "ironrdp-error 0.1.3", + "ironrdp-core", + "ironrdp-error", "uuid", ] diff --git a/crates/win-api-wrappers/src/process.rs b/crates/win-api-wrappers/src/process.rs index a2f359874..3c71e6e4e 100644 --- a/crates/win-api-wrappers/src/process.rs +++ b/crates/win-api-wrappers/src/process.rs @@ -29,13 +29,13 @@ use windows::Win32::System::ProcessStatus::K32GetMappedFileNameW; use windows::Win32::System::RemoteDesktop::ProcessIdToSessionId; use windows::Win32::System::Threading::{ CREATE_UNICODE_ENVIRONMENT, CreateProcessAsUserW, CreateRemoteThread, EXTENDED_STARTUPINFO_PRESENT, - GetCurrentProcess, GetCurrentProcessId, GetExitCodeProcess, GetProcessTimes, INFINITE, + GetCurrentProcess, GetCurrentProcessId, GetExitCodeProcess, GetProcessId, GetProcessTimes, 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_SYNCHRONIZE, PROCESS_TERMINATE, QueryFullProcessImageNameW, STARTUPINFOEXW, STARTUPINFOW, STARTUPINFOW_FLAGS, TerminateProcess, WaitForSingleObject, }; -use windows::Win32::UI::Shell::{SEE_MASK_NOCLOSEPROCESS, SHELLEXECUTEINFOW, ShellExecuteExW}; +use windows::Win32::UI::Shell::{SEE_MASK_NOASYNC, SEE_MASK_NOCLOSEPROCESS, SHELLEXECUTEINFOW, ShellExecuteExW}; use windows::Win32::UI::WindowsAndMessaging::{ EnumWindows, GetWindowThreadProcessId, PostThreadMessageW, SHOW_WINDOW_CMD, }; @@ -316,6 +316,17 @@ impl Process { Ok(exit_code) } + pub fn process_id(&self) -> Result { + // SAFETY: The process handle is valid for this call. + let process_id = unsafe { GetProcessId(self.handle.raw()) }; + + if process_id == 0 { + bail!(Error::last_error()); + } + + Ok(process_id) + } + pub fn query_basic_information(&self) -> Result { let mut basic_info = PROCESS_BASIC_INFORMATION::default(); @@ -431,19 +442,23 @@ impl HandleWrapper for Process { pub fn shell_execute( path: &Path, - command_line: &CommandLine, + command_line: &str, working_directory: &Path, verb: &str, show_cmd: SHOW_WINDOW_CMD, ) -> Result { let path = WideString::from(path); - let command_line = WideString::from(command_line.to_command_line()); - let working_directory = WideString::from(working_directory); + let command_line = WideString::from(command_line); + let working_directory = if working_directory.as_os_str().is_empty() { + WideString::default() + } else { + WideString::from(working_directory) + }; let verb = WideString::from(verb); let mut exec_info = SHELLEXECUTEINFOW { cbSize: u32size_of::(), - fMask: SEE_MASK_NOCLOSEPROCESS, + fMask: SEE_MASK_NOCLOSEPROCESS | SEE_MASK_NOASYNC, lpFile: path.as_pcwstr(), lpParameters: command_line.as_pcwstr(), lpDirectory: working_directory.as_pcwstr(), @@ -1124,12 +1139,16 @@ pub fn post_message_for_pid(pid: u32, msg: u32, wparam: WPARAM, lparam: LPARAM) ) }?; + if windows_enum_ctx.threads.is_empty() { + return Err(windows::core::Error::from(windows::Win32::Foundation::ERROR_NOT_FOUND).into()); + } + // Send message to all threads. - if !windows_enum_ctx.threads.is_empty() { - for thread in windows_enum_ctx.threads { - // SAFETY: No outstanding preconditions. - let _ = unsafe { PostThreadMessageW(thread, msg, wparam, lparam) }; - } + windows_enum_ctx.threads.sort_unstable(); + windows_enum_ctx.threads.dedup(); + for thread in windows_enum_ctx.threads { + // SAFETY: No outstanding preconditions. + unsafe { PostThreadMessageW(thread, msg, wparam, lparam) }?; } Ok(()) diff --git a/devolutions-session/Cargo.toml b/devolutions-session/Cargo.toml index b4f86bdde..87deabe71 100644 --- a/devolutions-session/Cargo.toml +++ b/devolutions-session/Cargo.toml @@ -42,7 +42,7 @@ win-api-wrappers = { path = "../crates/win-api-wrappers", optional = true } [dependencies.now-proto-pdu] optional = true -version = "0.4.3" +version = "0.5.0" features = ["std"] [dev-dependencies] diff --git a/devolutions-session/src/dvc/now_message_dissector.rs b/devolutions-session/src/dvc/now_message_dissector.rs index 4fe8dbe80..99549c7a7 100644 --- a/devolutions-session/src/dvc/now_message_dissector.rs +++ b/devolutions-session/src/dvc/now_message_dissector.rs @@ -1,6 +1,6 @@ use anyhow::Context; use now_proto_pdu::NowMessage; -use now_proto_pdu::ironrdp_core::{Decode, DecodeError, DecodeErrorKind, IntoOwned, ReadCursor, WriteBuf}; +use now_proto_pdu::ironrdp_core::{Decode, DecodeErrorKind, IntoOwned, ReadCursor, WriteBuf}; /// Reconstructs Now messages from a stream of bytes. #[derive(Default)] @@ -38,10 +38,7 @@ impl NowMessageDissector { // Remove the read bytes from the buffer self.start_pos += pos; } - Err(DecodeError { - kind: DecodeErrorKind::NotEnoughBytes { .. }, - .. - }) => { + Err(err) if matches!(err.kind(), DecodeErrorKind::NotEnoughBytes { .. }) => { // Need more data to read the message break; } diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index de3eeebf4..33ff10a06 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -1,9 +1,11 @@ use std::collections::HashMap; use std::os::windows::io::AsRawHandle as _; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; -use now_proto_pdu::{NowExecDataStreamKind, NowStatusError}; +use now_proto_pdu::{NowExecDataStreamKind, NowProtoError, NowStatusError}; use tokio::sync::mpsc::{Receiver, Sender}; use tracing::{debug, error, info, trace, warn}; use win_api_wrappers::event::Event; @@ -13,8 +15,8 @@ use win_api_wrappers::security::attributes::SecurityAttributesInit; use win_api_wrappers::thread::Thread; use win_api_wrappers::utils::{Pipe, WideString}; use windows::Win32::Foundation::{ - ERROR_BROKEN_PIPE, ERROR_HANDLE_EOF, ERROR_NOT_FOUND, GetLastError, HANDLE, LPARAM, WAIT_EVENT, WAIT_OBJECT_0, - WPARAM, + ERROR_ACCESS_DISABLED_BY_POLICY, ERROR_BROKEN_PIPE, ERROR_CANCELLED, ERROR_ELEVATION_REQUIRED, ERROR_HANDLE_EOF, + ERROR_NOT_FOUND, ERROR_PRIVILEGE_NOT_HELD, GetLastError, HANDLE, LPARAM, WAIT_EVENT, WAIT_OBJECT_0, WPARAM, }; use windows::Win32::Storage::FileSystem::{ReadFile, WriteFile}; use windows::Win32::System::IO::{CancelSynchronousIo, GetOverlappedResult, OVERLAPPED}; @@ -27,7 +29,7 @@ use windows::Win32::System::Threading::{ INFINITE, NORMAL_PRIORITY_CLASS, PROCESS_CREATION_FLAGS, PROCESS_INFORMATION, STARTF_USESHOWWINDOW, STARTF_USESTDHANDLES, STARTUPINFOW, WaitForMultipleObjects, }; -use windows::Win32::UI::WindowsAndMessaging::{SW_HIDE, WM_QUIT}; +use windows::Win32::UI::WindowsAndMessaging::{SW_HIDE, SW_RESTORE, WM_QUIT}; use windows::core::PCWSTR; use crate::dvc::channel::{ @@ -87,9 +89,14 @@ pub struct StdinChunk { /// Message, sent from Process IO thread to task to finalize process execution. #[derive(Debug)] pub enum ServerChannelEvent { + ProcessNotification { + active: Arc, + event: Box, + }, CloseChannel, SessionStarted { session_id: u32, + no_io_redirection: bool, }, SessionDataOut { session_id: u32, @@ -117,6 +124,42 @@ pub enum ServerChannelEvent { }, } +#[derive(Clone)] +struct ProcessNotifications { + sender: Sender, + active: Arc, + abort_exit_code: Arc, +} + +impl ProcessNotifications { + fn new(sender: Sender) -> Self { + Self { + sender, + active: Arc::new(AtomicBool::new(true)), + abort_exit_code: Arc::new(AtomicU64::new(0)), + } + } + + fn blocking_send( + &self, + event: ServerChannelEvent, + ) -> Result<(), tokio::sync::mpsc::error::SendError> { + self.sender.blocking_send(ServerChannelEvent::ProcessNotification { + active: Arc::clone(&self.active), + event: Box::new(event), + }) + } +} + +impl ServerChannelEvent { + pub fn into_active_event(self) -> Option { + match self { + Self::ProcessNotification { active, event } => active.load(Ordering::Acquire).then_some(*event), + event => Some(event), + } + } +} + pub struct WinApiProcessCtx { session_id: u32, @@ -127,12 +170,27 @@ pub struct WinApiProcessCtx { encoding: DataEncoding, pid: u32, + started_no_io_redirection: bool, + elevated: bool, + _temp_files: Vec, // NOTE: Order of fields is important, as process_handle must be dropped last in automatically // generated destructor, after all pipes were closed. process: Process, } +impl Drop for WinApiProcessCtx { + fn drop(&mut self) { + // Elevated processes can outlive their channel; keep scripts until the interpreter exits, even after errors. + if self.elevated + && !self._temp_files.is_empty() + && let Err(error) = self.process.wait(None) + { + error!(%error, session_id = self.session_id, "Failed to wait for elevated process cleanup"); + } + } +} + impl WinApiProcessCtx { pub fn process_abort(&mut self, exit_code: u32) -> Result<(), ExecError> { info!( @@ -145,29 +203,38 @@ impl WinApiProcessCtx { Ok(()) } - pub fn process_cancel(&mut self, io_notification_tx: Sender) -> Result<(), ExecError> { + fn process_cancel(&mut self, io_notification_tx: ProcessNotifications) -> Result<(), ExecError> { info!( session_id = self.session_id, "Cancelling process execution by user request" ); - post_message_for_pid(self.pid, WM_QUIT, WPARAM(0), LPARAM(0))?; - // TODO(DGW-301): Figure out how to correctly send CTRL+C to console applications. - // Acknowledge client that cancel request has been processed - // successfully. - io_notification_tx.blocking_send(ServerChannelEvent::SessionCancelSuccess { - session_id: self.session_id, - })?; + let notification = match post_message_for_pid(self.pid, WM_QUIT, WPARAM(0), LPARAM(0)) { + Ok(()) => ServerChannelEvent::SessionCancelSuccess { + session_id: self.session_id, + }, + Err(error) => { + let error = match error.downcast::() { + Ok(error) => NowStatusError::new_winapi(error.code().0.cast_unsigned()), + Err(error) => return Err(error.into()), + }; + ServerChannelEvent::SessionCancelFailed { + session_id: self.session_id, + error, + } + } + }; + io_notification_tx.blocking_send(notification)?; Ok(()) } - pub fn wait( + fn wait( mut self, mut input_event_rx: WinapiSignaledReceiver, - io_notification_tx: Sender, + io_notification_tx: ProcessNotifications, ) -> Result { let session_id = self.session_id; @@ -178,7 +245,10 @@ impl WinApiProcessCtx { const WAIT_OBJECT_INPUT_MESSAGE: WAIT_EVENT = WAIT_OBJECT_0; const WAIT_OBJECT_PROCESS_EXIT: WAIT_EVENT = WAIT_EVENT(WAIT_OBJECT_0.0 + 1); - io_notification_tx.blocking_send(ServerChannelEvent::SessionStarted { session_id })?; + io_notification_tx.blocking_send(ServerChannelEvent::SessionStarted { + session_id, + no_io_redirection: self.started_no_io_redirection, + })?; loop { // SAFETY: No preconditions. @@ -191,6 +261,15 @@ impl WinApiProcessCtx { return Ok(self.process.exit_code()?); } WAIT_OBJECT_INPUT_MESSAGE => { + if let Some(exit_code) = io_notification_tx + .abort_exit_code + .load(Ordering::Acquire) + .checked_sub(1) + { + let exit_code = u32::try_from(exit_code).expect("abort code was stored from a u32"); + self.process_abort(exit_code)?; + return Err(ExecError::Aborted); + } let process_io_message = input_event_rx.try_recv()?; trace!(session_id, ?process_io_message, "Received process IO message"); @@ -227,10 +306,10 @@ impl WinApiProcessCtx { /// Starts IO redirection loop for a launched process. /// /// Returns process exit code. - pub fn wait_with_io_redirection( + fn wait_with_io_redirection( mut self, mut input_event_rx: WinapiSignaledReceiver, - io_notification_tx: Sender, + io_notification_tx: ProcessNotifications, ) -> Result { let session_id = self.session_id; @@ -303,7 +382,10 @@ impl WinApiProcessCtx { // Signal client side about started execution - io_notification_tx.blocking_send(ServerChannelEvent::SessionStarted { session_id })?; + io_notification_tx.blocking_send(ServerChannelEvent::SessionStarted { + session_id, + no_io_redirection: false, + })?; info!(session_id, "Process IO is ready for async loop execution"); loop { @@ -317,6 +399,15 @@ impl WinApiProcessCtx { return Ok(self.process.exit_code()?); } WAIT_OBJECT_INPUT_MESSAGE => { + if let Some(exit_code) = io_notification_tx + .abort_exit_code + .load(Ordering::Acquire) + .checked_sub(1) + { + let exit_code = u32::try_from(exit_code).expect("abort code was stored from a u32"); + self.process_abort(exit_code)?; + return Err(ExecError::Aborted); + } let process_io_message = input_event_rx.try_recv()?; trace!(session_id, ?process_io_message, "Received process IO message"); @@ -614,6 +705,7 @@ pub struct WinApiProcessBuilder { command_line: String, current_directory: String, enable_io_redirection: bool, + elevated: bool, encoding: DataEncoding, env: HashMap, temp_files: Vec, @@ -626,6 +718,7 @@ impl WinApiProcessBuilder { command_line: String::new(), current_directory: String::new(), enable_io_redirection: false, + elevated: false, encoding: DataEncoding::from_oem_codepage(), env: HashMap::new(), temp_files: Vec::new(), @@ -656,6 +749,12 @@ impl WinApiProcessBuilder { self } + #[must_use] + pub fn with_elevated_execution(mut self) -> Self { + self.elevated = true; + self + } + #[must_use] pub fn with_env(mut self, key: &str, value: &str) -> Self { self.env.insert(key.to_owned(), value.to_owned()); @@ -671,7 +770,7 @@ impl WinApiProcessBuilder { fn run_impl( mut self, session_id: u32, - io_notification_tx: Option>, + io_notification_tx: Option, job: Option<&ProcessJob>, detached: bool, ) -> Result, ExecError> { @@ -692,13 +791,17 @@ impl WinApiProcessBuilder { WideString::from(current_directory) }; - // Move out temp files guard from builder to transfer over `WinApiProcess` instance - // later. - let temp_files = std::mem::take(&mut self.temp_files); + if self.elevated { + let io_notification_tx = io_notification_tx.ok_or_else(|| { + ExecError::Other(anyhow::anyhow!("elevated execution requires a notification sender")) + })?; + return self.run_elevated_impl(session_id, io_notification_tx, detached); + } + + let temp_files = std::mem::take(&mut self.temp_files); let io_redirection = self.enable_io_redirection; let encoding = self.encoding; - let process_ctx = if io_redirection { prepare_process_with_io_redirection(session_id, command_line, current_directory, self.env, encoding, job)? } else { @@ -749,6 +852,8 @@ impl WinApiProcessBuilder { let io_notification_tx = io_notification_tx.expect("BUG: io_notification_tx must be Some for non-detached mode"); + let notifications_active = Arc::clone(&io_notification_tx.active); + let abort_exit_code = Arc::clone(&io_notification_tx.abort_exit_code); let join_handle = std::thread::spawn(move || { let run_result = if io_redirection { @@ -768,6 +873,8 @@ impl WinApiProcessBuilder { }); Ok(Some(WinApiProcess { + notifications_active, + abort_exit_code, input_event_tx, stdin_writer, join_handle, @@ -775,17 +882,110 @@ impl WinApiProcessBuilder { })) } + fn run_elevated_impl( + self, + session_id: u32, + io_notification_tx: ProcessNotifications, + detached: bool, + ) -> Result, ExecError> { + let Self { + executable, + command_line, + current_directory, + enable_io_redirection: requested_io_redirection, + temp_files, + .. + } = self; + + if detached { + std::thread::Builder::new() + .name(format!("process-{session_id}-elevation")) + .spawn(move || { + let process_ctx = match prepare_elevated_process( + session_id, + &executable, + &command_line, + ¤t_directory, + requested_io_redirection, + temp_files, + ) { + Ok(process_ctx) => process_ctx, + Err(error) => { + let notification = ServerChannelEvent::SessionFailed { session_id, error }; + if let Err(error) = io_notification_tx.blocking_send(notification) { + error!(%error, session_id, "Failed to send elevated launch error"); + } + return; + } + }; + + if let Err(error) = io_notification_tx.blocking_send(ServerChannelEvent::SessionExited { + session_id, + exit_code: 0, + }) { + error!(%error, session_id, "Failed to send elevated launch result"); + } + + // Dropping the context waits for exit before releasing temporary scripts. + drop(process_ctx); + }) + .map_err(anyhow::Error::from)?; + + return Ok(None); + } + + let (input_event_tx, input_event_rx) = winapi_signaled_mpsc_channel()?; + let notifications_active = Arc::clone(&io_notification_tx.active); + let abort_exit_code = Arc::clone(&io_notification_tx.abort_exit_code); + let join_handle = std::thread::Builder::new() + .name(format!("process-{session_id}-elevation")) + .spawn(move || { + let run_result = prepare_elevated_process( + session_id, + &executable, + &command_line, + ¤t_directory, + requested_io_redirection, + temp_files, + ) + .and_then(|process_ctx| process_ctx.wait(input_event_rx, io_notification_tx.clone())); + + let notification = match run_result { + Ok(exit_code) => ServerChannelEvent::SessionExited { session_id, exit_code }, + Err(error) => ServerChannelEvent::SessionFailed { session_id, error }, + }; + + if let Err(error) = io_notification_tx.blocking_send(notification) { + error!(%error, session_id, "Failed to send elevated process result"); + } + })?; + + Ok(Some(WinApiProcess { + notifications_active, + abort_exit_code, + input_event_tx, + stdin_writer: None, + join_handle, + _temp_files: Vec::new(), + })) + } + /// Starts process execution and spawns IO thread to redirect stdio to/from dvc. /// - /// The process joins `job` before it runs, so it and every process it starts are killed when the job is dropped. + /// Standard launches join `job` when provided, so their process trees are killed when the job is dropped. pub fn run( self, session_id: u32, io_notification_tx: Sender, - job: &ProcessJob, + job: Option<&ProcessJob>, ) -> Result { Ok(self - .run_impl(session_id, Some(io_notification_tx), Some(job), false)? + .run_impl( + session_id, + Some(ProcessNotifications::new(io_notification_tx)), + job, + false, + )? .expect("result should be non-optional when running in non-detached mode")) } @@ -793,10 +993,22 @@ impl WinApiProcessBuilder { /// No IO redirection. Process exit is monitored in a background thread to manage temp file cleanup. /// Returns immediately after spawning. pub fn run_detached(self, session_id: u32) -> Result<(), ExecError> { - // Result always empty and therefore ignored in detached mode. self.run_impl(session_id, None, None, true)?; Ok(()) } + + pub fn run_detached_with_notifications( + self, + session_id: u32, + io_notification_tx: Sender, + ) -> Result, ExecError> { + // Result always empty and therefore ignored in detached mode. + let notifications = ProcessNotifications::new(io_notification_tx); + let active = Arc::clone(¬ifications.active); + let sender = self.elevated.then_some(notifications); + self.run_impl(session_id, sender, None, true)?; + Ok(active) + } } /// Job object whose processes are killed when it is dropped, or when the host exits or crashes. @@ -927,6 +1139,9 @@ fn prepare_process( stdin_write_pipe: None, encoding, pid, + started_no_io_redirection: false, + elevated: false, + _temp_files: Vec::new(), process: process_handle, }) } @@ -1001,14 +1216,72 @@ fn prepare_process_with_io_redirection( stdin_write_pipe: Some(stdin_write_pipe), encoding, pid, + started_no_io_redirection: false, + elevated: false, + _temp_files: Vec::new(), process: process_handle, }; Ok(process_ctx) } +fn prepare_elevated_process( + session_id: u32, + executable: &str, + command_line: &str, + current_directory: &str, + requested_io_redirection: bool, + temp_files: Vec, +) -> Result { + let process = win_api_wrappers::process::shell_execute( + std::path::Path::new(executable), + command_line, + std::path::Path::new(current_directory), + "runas", + SW_RESTORE, + ) + .map_err(map_elevated_launch_error)?; + let mut process_ctx = WinApiProcessCtx { + session_id, + stdout_read_pipe: None, + stderr_read_pipe: None, + stdin_write_pipe: None, + encoding: DataEncoding::Raw, + pid: 0, + started_no_io_redirection: requested_io_redirection, + elevated: true, + _temp_files: temp_files, + process, + }; + process_ctx.pid = process_ctx.process.process_id()?; + Ok(process_ctx) +} + +fn map_elevated_launch_error(error: anyhow::Error) -> ExecError { + match error.downcast::() { + Ok(error) + if [ERROR_CANCELLED, ERROR_ACCESS_DISABLED_BY_POLICY] + .iter() + .any(|code| error.code() == code.to_hresult()) => + { + ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::ElevationDenied)) + } + Ok(error) + if [ERROR_ELEVATION_REQUIRED, ERROR_PRIVILEGE_NOT_HELD] + .iter() + .any(|code| error.code() == code.to_hresult()) => + { + ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::ElevationUnavailable)) + } + Ok(error) => ExecError::from(error), + Err(error) => ExecError::Other(error), + } +} + /// Represents spawned process with IO redirection. pub struct WinApiProcess { + notifications_active: Arc, + abort_exit_code: Arc, input_event_tx: WinapiSignaledSender, /// `None` without IO redirection. stdin_writer: Option, @@ -1018,6 +1291,8 @@ pub struct WinApiProcess { impl Drop for WinApiProcess { fn drop(&mut self) { + // Invalidate queued events too, because an aborted session ID can be reused before consent completes. + self.notifications_active.store(false, Ordering::Release); // Ensure that the input event channel is closed when the process is dropped. // This will signal the IO thread to terminate if it is still running. let _ = self.input_event_tx.try_send(ProcessIoInputEvent::TerminateIo); @@ -1026,13 +1301,27 @@ impl Drop for WinApiProcess { impl WinApiProcess { pub async fn abort_execution(&self, exit_code: u32) -> anyhow::Result<()> { - self.input_event_tx - .send(ProcessIoInputEvent::AbortExecution(exit_code)) - .await + // A full queue is already signaled, so the worker will see this abort before draining another request. + self.abort_exit_code.store(u64::from(exit_code) + 1, Ordering::Release); + match self + .input_event_tx + .try_send(ProcessIoInputEvent::AbortExecution(exit_code)) + { + Ok(()) => Ok(()), + Err(error) + if matches!( + error.downcast_ref::>(), + Some(tokio::sync::mpsc::error::TrySendError::Full(_)) + ) => + { + Ok(()) + } + Err(error) => Err(error), + } } pub async fn cancel_execution(&self) -> anyhow::Result<()> { - self.input_event_tx.send(ProcessIoInputEvent::CancelExecution).await + self.input_event_tx.try_send(ProcessIoInputEvent::CancelExecution) } /// Returns a send of `data` to the process stdin that waits while the stdin queue is full. @@ -1059,7 +1348,7 @@ impl WinApiProcess { } pub async fn shutdown(&self) -> anyhow::Result<()> { - self.input_event_tx.send(ProcessIoInputEvent::TerminateIo).await?; + self.input_event_tx.try_send(ProcessIoInputEvent::TerminateIo)?; Ok(()) } @@ -1072,8 +1361,293 @@ impl WinApiProcess { mod tests { use std::time::Duration; + use now_proto_pdu::NowExecResultMsg; + use super::*; + // oracle: regression #2191 abort must terminate a standard child even when cancellation filled its control queue. + #[tokio::test(flavor = "multi_thread")] + async fn abort_with_full_control_queue_terminates_child() { + let job = ProcessJob::new().expect("create job"); + let context = running_process(&job); + let child = Process::from(context.process.handle.try_clone().expect("duplicate process handle")); + let (sender, _receiver) = tokio::sync::mpsc::channel(10); + let notifications = ProcessNotifications::new(sender); + let (input_event_tx, input_event_rx) = winapi_signaled_mpsc_channel().expect("create input channel"); + while input_event_tx.try_send(ProcessIoInputEvent::CancelExecution).is_ok() {} + let (start_tx, start_rx) = std::sync::mpsc::channel(); + let notifications_active = Arc::clone(¬ifications.active); + let abort_exit_code = Arc::clone(¬ifications.abort_exit_code); + let join_handle = std::thread::spawn(move || { + start_rx.recv().expect("start monitoring"); + let result = context.wait(input_event_rx, notifications); + assert!(matches!(result, Err(ExecError::Aborted))); + }); + let process = WinApiProcess { + notifications_active, + abort_exit_code, + input_event_tx, + stdin_writer: None, + join_handle, + _temp_files: Vec::new(), + }; + + process + .abort_execution(7) + .await + .expect("record abort despite full queue"); + start_tx.send(()).expect("start monitoring"); + assert_eq!( + child.wait(Some(10_000)).expect("wait for child"), + WAIT_OBJECT_0, + "abort must terminate the child without dropping the shared job" + ); + assert_eq!(child.exit_code().expect("child exit code"), 7); + } + + // oracle: regression #2191 pending consent must not let a full control queue block cancellation or shutdown. + #[tokio::test] + async fn full_control_queue_does_not_block_dispatcher() { + let (input_event_tx, _input_event_rx) = winapi_signaled_mpsc_channel().expect("create input channel"); + while input_event_tx.try_send(ProcessIoInputEvent::CancelExecution).is_ok() {} + let process = WinApiProcess { + notifications_active: Arc::new(AtomicBool::new(true)), + abort_exit_code: Arc::new(AtomicU64::new(0)), + input_event_tx, + stdin_writer: None, + join_handle: std::thread::spawn(|| {}), + _temp_files: Vec::new(), + }; + + tokio::time::timeout(Duration::from_secs(1), process.cancel_execution()) + .await + .expect("cancel must not wait for consent") + .expect_err("full queue cannot accept cancel"); + tokio::time::timeout(Duration::from_secs(1), process.abort_execution(7)) + .await + .expect("abort must not wait for consent") + .expect("abort does not depend on queue capacity"); + tokio::time::timeout(Duration::from_secs(1), process.shutdown()) + .await + .expect("shutdown must not wait for consent") + .expect_err("full queue cannot accept shutdown"); + } + + // oracle: regression #2191 elevated fire-and-forget launches without scripts must release their worker on launch. + #[test] + fn elevated_cleanup_without_scripts_does_not_wait_for_exit() { + let job = ProcessJob::new().expect("create job"); + let mut context = running_process(&job); + context.elevated = true; + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + drop(context); + done_tx.send(()).expect("report cleanup"); + }); + + let completed = done_rx.recv_timeout(Duration::from_secs(2)); + drop(job); + worker.join().expect("cleanup worker completes"); + completed.expect("cleanup must finish while the child is still running"); + } + + // oracle: regression #2191 aborting an execution must invalidate queued and future events before its ID is reused. + #[test] + fn dropped_execution_discards_queued_and_late_worker_failures() { + let (sender, mut receiver) = tokio::sync::mpsc::channel(10); + let notifications = ProcessNotifications::new(sender.clone()); + let (input_event_tx, _input_event_rx) = winapi_signaled_mpsc_channel().expect("create input channel"); + let process = WinApiProcess { + notifications_active: Arc::clone(¬ifications.active), + abort_exit_code: Arc::clone(¬ifications.abort_exit_code), + input_event_tx, + stdin_writer: None, + join_handle: std::thread::spawn(|| {}), + _temp_files: Vec::new(), + }; + notifications + .blocking_send(ServerChannelEvent::SessionFailed { + session_id: 1, + error: ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::ElevationDenied)), + }) + .expect("queue old failure"); + drop(process); + notifications + .blocking_send(ServerChannelEvent::SessionFailed { + session_id: 1, + error: ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::ElevationDenied)), + }) + .expect("queue late old failure"); + let replacement = ProcessNotifications::new(sender); + replacement + .blocking_send(ServerChannelEvent::SessionStarted { + session_id: 1, + no_io_redirection: true, + }) + .expect("queue replacement event"); + + assert!(receiver.try_recv().expect("old failure").into_active_event().is_none()); + assert!(receiver.try_recv().expect("late failure").into_active_event().is_none()); + assert!(matches!( + receiver.try_recv().expect("replacement event").into_active_event(), + Some(ServerChannelEvent::SessionStarted { session_id: 1, .. }) + )); + } + + // oracle: regression #2191 channel loss during consent must not delete an elevated process's script before exit. + #[test] + fn closed_notification_channel_retains_elevated_script_until_exit() { + let job = ProcessJob::new().expect("create job"); + let mut context = running_process(&job); + let process = Process::from(context.process.handle.try_clone().expect("duplicate process handle")); + let script = TmpFileGuard::new("cmd").expect("create script"); + let script_path = script.path().to_owned(); + context.elevated = true; + context._temp_files.push(script); + + let (_input_tx, input_rx) = winapi_signaled_mpsc_channel().expect("create input channel"); + let (notification_tx, notification_rx) = tokio::sync::mpsc::channel(10); + drop(notification_rx); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + let result = context.wait(input_rx, ProcessNotifications::new(notification_tx)); + done_tx.send(result).expect("report worker result"); + }); + + assert!(matches!( + done_rx.recv_timeout(Duration::from_millis(100)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + )); + assert!(script_path.exists(), "script must survive notification failure"); + + process.terminate(7).expect("stop child"); + done_rx + .recv_timeout(Duration::from_secs(10)) + .expect("worker finishes after exit") + .expect_err("the notification receiver was closed"); + worker.join().expect("worker completes"); + assert!(!script_path.exists(), "script is removed after process exit"); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[tokio::test(flavor = "multi_thread")] + async fn cancellation_failure_keeps_process_tracked_until_exit() { + let job = ProcessJob::new().expect("create job"); + let mut context = running_process(&job); + let process = Process::from(context.process.handle.try_clone().expect("duplicate process handle")); + // An absent window target deterministically exercises the same failure path as UIPI denial. + context.pid = u32::MAX; + context.elevated = true; + + let (input_tx, input_rx) = winapi_signaled_mpsc_channel().expect("create input channel"); + let (notification_tx, mut notification_rx) = tokio::sync::mpsc::channel(10); + let (done_tx, done_rx) = tokio::sync::oneshot::channel(); + let worker = std::thread::spawn(move || { + let result = context.wait(input_rx, ProcessNotifications::new(notification_tx)); + done_tx.send(result).expect("report exit"); + }); + assert!(matches!( + next_process_event(&mut notification_rx).await, + ServerChannelEvent::SessionStarted { session_id: 1, .. } + )); + input_tx + .send(ProcessIoInputEvent::CancelExecution) + .await + .expect("request cancel"); + assert!(matches!( + next_process_event(&mut notification_rx).await, + ServerChannelEvent::SessionCancelFailed { session_id: 1, .. } + )); + + process.terminate(7).expect("stop child"); + let exit_code = tokio::time::timeout(Duration::from_secs(10), done_rx) + .await + .expect("worker completes") + .expect("worker reports result") + .expect("cancel failure does not end monitoring"); + assert_eq!(exit_code, 7); + worker.join().expect("worker completes"); + } + + fn running_process(job: &ProcessJob) -> WinApiProcessCtx { + prepare_process( + 1, + WideString::from("cmd.exe /d /c ping -n 600 127.0.0.1 >nul"), + WideString::default(), + HashMap::new(), + DataEncoding::Raw, + Some(job), + ) + .expect("start child") + } + + async fn next_process_event(receiver: &mut Receiver) -> ServerChannelEvent { + tokio::time::timeout(Duration::from_secs(10), receiver.recv()) + .await + .expect("event arrives") + .expect("notification channel is open") + .into_active_event() + .expect("process is still active") + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[test] + fn elevation_policy_failure_reports_denied() { + assert_elevation_status(ERROR_ACCESS_DISABLED_BY_POLICY, NowProtoError::ElevationDenied); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[test] + fn missing_elevated_token_reports_unavailable() { + assert_elevation_status(ERROR_ELEVATION_REQUIRED, NowProtoError::ElevationUnavailable); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[test] + fn missing_elevation_privilege_reports_unavailable() { + assert_elevation_status(ERROR_PRIVILEGE_NOT_HELD, NowProtoError::ElevationUnavailable); + } + + fn assert_elevation_status(code: windows::Win32::Foundation::WIN32_ERROR, expected: NowProtoError) { + let ExecError::NowStatus(actual) = map_elevated_launch_error(windows::core::Error::from(code).into()) else { + panic!("elevation failure must use a NOW status error"); + }; + assert_eq!( + NowExecResultMsg::new_error(1, actual).expect("valid status"), + NowExecResultMsg::new_error(1, NowStatusError::new_proto(expected)).expect("valid expected status") + ); + } + + // oracle: spec ShellExecuteExW §Remarks + #[test] + fn missing_elevated_executable_preserves_platform_error() { + let error = windows::core::Error::from(windows::Win32::Foundation::ERROR_FILE_NOT_FOUND); + let expected = NowStatusError::new_winapi(error.code().0.cast_unsigned()); + let ExecError::NowStatus(actual) = map_elevated_launch_error(error.into()) else { + panic!("missing executable must use a platform status"); + }; + assert_eq!( + NowExecResultMsg::new_error(1, actual).expect("valid status"), + NowExecResultMsg::new_error(1, expected).expect("valid expected status") + ); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[test] + fn elevation_consent_cancellation_reports_elevation_denied() { + let error = windows::core::Error::from(ERROR_CANCELLED); + + let ExecError::NowStatus(actual) = map_elevated_launch_error(error.into()) else { + panic!("elevation cancellation must use a NOW status error"); + }; + let expected = NowStatusError::new_proto(NowProtoError::ElevationDenied); + + assert_eq!( + NowExecResultMsg::new_error(1, actual).expect("valid status error"), + NowExecResultMsg::new_error(1, expected).expect("valid expected status error") + ); + } + // oracle: regression #2156 detached Process without temporary files must not panic after launch. #[test] fn detached_process_without_temp_files_returns_success() { @@ -1101,7 +1675,7 @@ mod tests { )) .with_io_redirection(true) .with_encoding(DataEncoding::Raw) - .run(SESSION_ID, io_notification_tx, &job) + .run(SESSION_ID, io_notification_tx, Some(&job)) .expect("start cmd.exe"); let input: String = (0..INPUT_LINES).map(|i| format!("input-line-{i:05}\r\n")).collect(); @@ -1110,7 +1684,13 @@ mod tests { let mut stdout = Vec::new(); let exit_code = tokio::time::timeout(Duration::from_secs(60), async { loop { - match io_notification_rx.recv().await.expect("notification channel open") { + match io_notification_rx + .recv() + .await + .expect("notification channel open") + .into_active_event() + .expect("process is still active") + { ServerChannelEvent::SessionDataOut { stream: NowExecDataStreamKind::Stdout, data, @@ -1148,7 +1728,7 @@ mod tests { let process = WinApiProcessBuilder::new("cmd.exe") .with_command_line("/d /c ping -n 600 127.0.0.1 > nul") .with_io_redirection(true) - .run(SESSION_ID, io_notification_tx, &job) + .run(SESSION_ID, io_notification_tx, Some(&job)) .expect("start cmd.exe"); process.send_stdin(vec![b'x'; 1024 * 1024], false).await; diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index 2a5b1618f..d0496e187 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, bail}; use async_trait::async_trait; @@ -274,8 +275,6 @@ async fn process_messages( mut shutdown_signal: ShutdownSignal, init_progress: &DvcInitializationProgress, ) -> anyhow::Result<()> { - let (io_notification_tx, mut task_rx) = mpsc::channel(100); - // Detect pwsh (PowerShell 7+) availability once on DVC start, so that the // corresponding capability flag is only advertised when it is actually installed. let pwsh_executable = find_pwsh_executable(); @@ -348,12 +347,7 @@ async fn process_messages( // the agent applies, not the server default that omits the interval. dvc_tx.send(downgraded_caps.clone().into()).await?; - let mut processor = MessageProcessor::new( - downgraded_caps, - dvc_tx.clone(), - io_notification_tx.clone(), - pwsh_executable.clone(), - ); + let (mut processor, mut task_rx) = MessageProcessor::new(downgraded_caps, dvc_tx.clone(), pwsh_executable.clone()); init_progress.mark_negotiated(); info!("DVC negotiation completed"); @@ -394,11 +388,10 @@ async fn process_messages( _heartbeat = spawn_heartbeat(dvc_tx.clone(), downgraded_caps.heartbeat_interval()); init_progress.set_heartbeats_enabled(heartbeats_enabled(downgraded_caps.heartbeat_interval())); - // Dropping the old processor ends its exec sessions and the processes they started. - processor = MessageProcessor::new( + // Each negotiation owns its notifications so old workers cannot address reused IDs. + (processor, task_rx) = MessageProcessor::new( downgraded_caps.clone(), dvc_tx.clone(), - io_notification_tx.clone(), pwsh_executable.clone(), ); @@ -418,10 +411,23 @@ async fn process_messages( task_rx = task_rx.recv() => { match task_rx { Some(notification) => { + let Some(notification) = notification.into_active_event() else { + continue; + }; match notification { - ServerChannelEvent::SessionStarted { session_id } => { + ServerChannelEvent::ProcessNotification { .. } => { + unreachable!("process notifications are unwrapped before dispatch"); + } + ServerChannelEvent::SessionStarted { + session_id, + no_io_redirection, + } => { info!(session_id, "Session started"); - let message = NowExecStartedMsg::new(session_id); + let message = if no_io_redirection { + NowExecStartedMsg::new(session_id).with_no_io_redirection() + } else { + NowExecStartedMsg::new(session_id) + }; dvc_tx.send(message.into()).await?; } ServerChannelEvent::SessionDataOut { session_id, stream, last, data } => { @@ -441,6 +447,7 @@ async fn process_messages( ServerChannelEvent::SessionExited { session_id, exit_code } => { info!(session_id, %exit_code, "Session exited"); processor.remove_session(session_id); + processor.pending_sessions.remove(&session_id); discard_pending_stdin(&mut pending_stdin, session_id); let message = NowExecResultMsg::new_success(session_id, exit_code); @@ -449,6 +456,7 @@ async fn process_messages( ServerChannelEvent::SessionFailed { session_id, error } => { error!(session_id, %error, "Session error"); processor.remove_session(session_id); + processor.pending_sessions.remove(&session_id); discard_pending_stdin(&mut pending_stdin, session_id); handle_exec_error(&dvc_tx, session_id, error).await; @@ -539,7 +547,8 @@ fn default_server_caps(pwsh_available: bool) -> NowChannelCapsetMsg { | NowExecCapsetFlags::STYLE_BATCH | NowExecCapsetFlags::STYLE_WINPS | NowExecCapsetFlags::IO_REDIRECTION - | NowExecCapsetFlags::UNICODE_CONSOLE; + | NowExecCapsetFlags::UNICODE_CONSOLE + | NowExecCapsetFlags::ELEVATION; if pwsh_available { exec_flags |= NowExecCapsetFlags::STYLE_PWSH; @@ -624,9 +633,11 @@ struct MessageProcessor { capabilities: NowChannelCapsetMsg, /// Attached execution sessions of this channel. /// - /// Ending the channel or renegotiating capabilities drops this processor, which stops their IO and kills them with - /// `process_job`; no later channel or negotiation could reach them, because each starts with no sessions. + /// Ending the channel or renegotiating capabilities drops this processor, which stops its IO and kills + /// non-elevated sessions with `process_job`; elevated sessions may outlive their channel. sessions: HashMap, + /// Elevated launch requests that are waiting for user consent and have no process handle yet. + pending_sessions: HashMap>, /// Job of the attached executions and the processes they start; created by the first attached execution. /// /// Processes left running by an attached execution that already exited are killed with it too. @@ -644,21 +655,23 @@ impl MessageProcessor { pub(crate) fn new( capabilities: NowChannelCapsetMsg, dvc_tx: WinapiSignaledSender, - io_notification_tx: Sender, pwsh_executable: Option, - ) -> Self { + ) -> (Self, Receiver) { + let (io_notification_tx, task_rx) = mpsc::channel(100); let rdm_handler = RdmMessageProcessor::new(dvc_tx.clone()); - Self { + let processor = Self { dvc_tx, io_notification_tx, capabilities, sessions: HashMap::new(), + pending_sessions: HashMap::new(), process_job: None, rdm_handler, pwsh_executable, window_monitor_shutdown_tx: None, window_monitor_handle: None, - } + }; + (processor, task_rx) } fn process_job(&mut self) -> Result<&ProcessJob, ExecError> { @@ -669,7 +682,7 @@ impl MessageProcessor { } async fn ensure_session_id_free(&self, session_id: u32) -> Result<(), ExecError> { - if self.sessions.contains_key(&session_id) { + if self.sessions.contains_key(&session_id) || self.pending_sessions.contains_key(&session_id) { warn!(session_id, "Session ID is in use"); return Err(ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::InUse))); } @@ -679,6 +692,16 @@ impl MessageProcessor { Ok(()) } + fn ensure_elevation_supported(&self, elevated: bool) -> Result<(), ExecError> { + if elevated && !self.capabilities.exec_capset().contains(NowExecCapsetFlags::ELEVATION) { + return Err(ExecError::NowStatus(NowStatusError::new_proto( + NowProtoError::NotImplemented, + ))); + } + + Ok(()) + } + async fn send_detached_process_success(&self, session_id: u32) -> Result<(), ExecError> { self.io_notification_tx .send(ServerChannelEvent::SessionExited { @@ -689,6 +712,39 @@ impl MessageProcessor { Ok(()) } + async fn run_detached_exec( + &mut self, + process: WinApiProcessBuilder, + session_id: u32, + elevated: bool, + ) -> Result<(), ExecError> { + let notifications_active = + process.run_detached_with_notifications(session_id, self.io_notification_tx.clone())?; + + if elevated { + self.pending_sessions.insert(session_id, notifications_active); + } else { + self.send_detached_process_success(session_id).await?; + } + + Ok(()) + } + + fn run_attached_exec( + &mut self, + process: WinApiProcessBuilder, + session_id: u32, + io_redirection: bool, + elevated: bool, + ) -> Result { + let io_notification_tx = self.io_notification_tx.clone(); + let job = if elevated { None } else { Some(self.process_job()?) }; + + process + .with_io_redirection(io_redirection) + .run(session_id, io_notification_tx, job) + } + pub(crate) async fn process_message(&mut self, message: OwnedNowMessage) -> anyhow::Result { match message { NowMessage::Channel(NowChannelMessage::Capset(client_caps)) => { @@ -718,6 +774,18 @@ impl MessageProcessor { } } } + NowMessage::Exec(NowExecMessage::Shell(exec_msg)) => { + let session_id = exec_msg.session_id(); + let error = match self.ensure_elevation_supported(exec_msg.is_elevated()) { + Err(error) => error, + Ok(()) => ExecError::NowStatus(NowStatusError::new_proto(if exec_msg.is_elevated() { + NowProtoError::ElevationUnavailable + } else { + NowProtoError::NotImplemented + })), + }; + handle_exec_error(&self.dvc_tx, session_id, error).await; + } NowMessage::Exec(NowExecMessage::Batch(batch_msg)) => { let session_id = batch_msg.session_id(); match self.process_exec_batch(batch_msg).await { @@ -748,6 +816,11 @@ impl MessageProcessor { NowMessage::Exec(NowExecMessage::Abort(abort_msg)) => { let session_id = abort_msg.session_id(); + if let Some(active) = self.pending_sessions.remove(&session_id) { + active.store(false, Ordering::Release); + return Ok(ProcessMessageAction::Continue); + } + let process = match self.sessions.get_mut(&session_id) { Some(process) => process, None => { @@ -756,7 +829,9 @@ impl MessageProcessor { } }; - process.abort_execution(abort_msg.exit_code()).await?; + if let Err(error) = process.abort_execution(abort_msg.exit_code()).await { + warn!(session_id, %error, "Failed to enqueue process abort"); + } // We could drop session immediately after abort as client do not expect any further // communication. @@ -781,7 +856,14 @@ impl MessageProcessor { } }; - process.cancel_execution().await?; + if let Err(error) = process.cancel_execution().await { + warn!(session_id, %error, "Failed to enqueue process cancellation"); + let message = NowExecCancelRspMsg::new_error( + session_id, + NowStatusError::new_generic(GENERIC_ERROR_CODE_OTHER), + )?; + self.dvc_tx.send(message.into_owned().into()).await?; + } } NowMessage::Exec(NowExecMessage::Data(data_msg)) => { let session_id = data_msg.session_id(); @@ -919,11 +1001,21 @@ impl MessageProcessor { Ok(ProcessMessageAction::Continue) } - async fn process_exec_run(&self, params: NowExecRunMsg<'_>) -> Result<(), ExecError> { + async fn process_exec_run(&mut self, params: NowExecRunMsg<'_>) -> Result<(), ExecError> { + self.ensure_elevation_supported(params.is_elevated())?; self.ensure_session_id_free(params.session_id()).await?; let session_id = params.session_id(); + if params.is_elevated() { + let mut run_process = WinApiProcessBuilder::new(params.command()).with_elevated_execution(); + if let Some(directory) = params.directory() { + run_process = run_process.with_current_directory(directory); + } + + return self.run_detached_exec(run_process, session_id, true).await; + } + // Empty null-terminated string. let parameters = WideString::from(""); let operation = WideString::from("open"); @@ -965,6 +1057,7 @@ impl MessageProcessor { } async fn process_exec_process(&mut self, exec_msg: NowExecProcessMsg<'_>) -> Result<(), ExecError> { + self.ensure_elevation_supported(exec_msg.is_elevated())?; self.ensure_session_id_free(exec_msg.session_id()).await?; // Process exec: no assumptions about encoding by default (raw passthrough). @@ -985,17 +1078,21 @@ impl MessageProcessor { run_process = run_process.with_current_directory(directory); } + if exec_msg.is_elevated() { + run_process = run_process.with_elevated_execution(); + } + if exec_msg.is_detached() { - // Detached mode: fire-and-forget, no IO redirection - run_process.run_detached(exec_msg.session_id())?; - self.send_detached_process_success(exec_msg.session_id()).await?; - return Ok(()); + return self + .run_detached_exec(run_process, exec_msg.session_id(), exec_msg.is_elevated()) + .await; } - let process = run_process.with_io_redirection(exec_msg.is_with_io_redirection()).run( + let process = self.run_attached_exec( + run_process, exec_msg.session_id(), - self.io_notification_tx.clone(), - self.process_job()?, + exec_msg.is_with_io_redirection(), + exec_msg.is_elevated(), )?; self.sessions.insert(exec_msg.session_id(), process); @@ -1004,6 +1101,7 @@ impl MessageProcessor { } async fn process_exec_batch(&mut self, batch_msg: NowExecBatchMsg<'_>) -> Result<(), ExecError> { + self.ensure_elevation_supported(batch_msg.is_elevated())?; self.ensure_session_id_free(batch_msg.session_id()).await?; let mut script = batch_msg.command().to_owned(); @@ -1024,7 +1122,13 @@ impl MessageProcessor { let tmp_file = TmpFileGuard::new("bat")?; tmp_file.write_bytes(&file_encoding.encode_str(&script))?; - let parameters = format!("/Q /C \"{}\"", tmp_file.path_string()); + let command_kind = batch_command_kind( + batch_msg.is_no_exit(), + batch_msg.is_with_io_redirection(), + batch_msg.is_elevated(), + batch_msg.is_detached(), + ); + let parameters = format!("/Q {command_kind} \"{}\"", tmp_file.path_string()); let mut run_batch = WinApiProcessBuilder::new("cmd.exe") .with_temp_file(tmp_file) @@ -1035,16 +1139,21 @@ impl MessageProcessor { run_batch = run_batch.with_current_directory(directory); } + if batch_msg.is_elevated() { + run_batch = run_batch.with_elevated_execution(); + } + if batch_msg.is_detached() { - run_batch.run_detached(batch_msg.session_id())?; - self.send_detached_process_success(batch_msg.session_id()).await?; - return Ok(()); + return self + .run_detached_exec(run_batch, batch_msg.session_id(), batch_msg.is_elevated()) + .await; } - let process = run_batch.with_io_redirection(batch_msg.is_with_io_redirection()).run( + let process = self.run_attached_exec( + run_batch, batch_msg.session_id(), - self.io_notification_tx.clone(), - self.process_job()?, + batch_msg.is_with_io_redirection(), + batch_msg.is_elevated(), )?; self.sessions.insert(batch_msg.session_id(), process); @@ -1053,6 +1162,7 @@ impl MessageProcessor { } async fn process_exec_winps(&mut self, winps_msg: NowExecWinPsMsg<'_>) -> Result<(), ExecError> { + self.ensure_elevation_supported(winps_msg.is_elevated())?; self.ensure_session_id_free(winps_msg.session_id()).await?; let mut params = Vec::new(); @@ -1097,19 +1207,22 @@ impl MessageProcessor { run_process = run_process.with_current_directory(directory); } + if winps_msg.is_elevated() { + run_process = run_process.with_elevated_execution(); + } + if winps_msg.is_detached() { - run_process.run_detached(winps_msg.session_id())?; - self.send_detached_process_success(winps_msg.session_id()).await?; - return Ok(()); + return self + .run_detached_exec(run_process, winps_msg.session_id(), winps_msg.is_elevated()) + .await; } - let process = run_process - .with_io_redirection(winps_msg.is_with_io_redirection()) - .run( - winps_msg.session_id(), - self.io_notification_tx.clone(), - self.process_job()?, - )?; + let process = self.run_attached_exec( + run_process, + winps_msg.session_id(), + winps_msg.is_with_io_redirection(), + winps_msg.is_elevated(), + )?; self.sessions.insert(winps_msg.session_id(), process); @@ -1117,9 +1230,14 @@ impl MessageProcessor { } async fn process_exec_pwsh(&mut self, pwsh_msg: NowExecPwshMsg<'_>) -> Result<(), ExecError> { + self.ensure_elevation_supported(pwsh_msg.is_elevated())?; let Some(pwsh_executable) = self.pwsh_executable.clone() else { return Err(ExecError::NowStatus(NowStatusError::new_proto( - NowProtoError::NotImplemented, + if pwsh_msg.is_elevated() { + NowProtoError::ElevationUnavailable + } else { + NowProtoError::NotImplemented + }, ))); }; @@ -1168,16 +1286,21 @@ impl MessageProcessor { run_process = run_process.with_current_directory(directory); } + if pwsh_msg.is_elevated() { + run_process = run_process.with_elevated_execution(); + } + if pwsh_msg.is_detached() { - run_process.run_detached(pwsh_msg.session_id())?; - self.send_detached_process_success(pwsh_msg.session_id()).await?; - return Ok(()); + return self + .run_detached_exec(run_process, pwsh_msg.session_id(), pwsh_msg.is_elevated()) + .await; } - let process = run_process.with_io_redirection(pwsh_msg.is_with_io_redirection()).run( + let process = self.run_attached_exec( + run_process, pwsh_msg.session_id(), - self.io_notification_tx.clone(), - self.process_job()?, + pwsh_msg.is_with_io_redirection(), + pwsh_msg.is_elevated(), )?; self.sessions.insert(pwsh_msg.session_id(), process); @@ -1248,6 +1371,14 @@ impl MessageProcessor { } } +fn batch_command_kind(no_exit: bool, io_redirection: bool, elevated: bool, detached: bool) -> &'static str { + if no_exit && (!io_redirection || elevated || detached) { + "/K" + } else { + "/C" + } +} + fn append_ps_args(args: &mut Vec, msg: &NowExecWinPsMsg<'_>) { if let Some(execution_policy) = msg.execution_policy() { args.push("-ExecutionPolicy".to_owned()); @@ -1555,6 +1686,8 @@ mod tests { use core::task::Poll; use core::time::Duration; use std::collections::VecDeque; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; use anyhow::anyhow; use devolutions_gateway_task::{ShutdownHandle, ShutdownSignal}; @@ -1569,11 +1702,114 @@ mod tests { use windows::Win32::System::Threading::{PROCESS_SYNCHRONIZE, PROCESS_TERMINATE}; use super::{ - ChannelOutcome, DEFAULT_HEARTBEAT_INTERVAL, MAX_CONSECUTIVE_FAILED_REOPENS, REOPEN_DELAY_CAP, - REOPEN_DELAY_FLOOR, heartbeat_timer, process_messages, process_until_io_ends, serve_dvc, spawn_heartbeat, + ChannelOutcome, DEFAULT_HEARTBEAT_INTERVAL, MAX_CONSECUTIVE_FAILED_REOPENS, MessageProcessor, REOPEN_DELAY_CAP, + REOPEN_DELAY_FLOOR, batch_command_kind, heartbeat_timer, process_messages, process_until_io_ends, serve_dvc, + spawn_heartbeat, }; use crate::dvc::channel::{bounded_mpsc_channel, winapi_signaled_mpsc_channel}; use crate::dvc::io::DvcInitializationProgress; + use crate::dvc::process::ServerChannelEvent; + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §NOW_EXEC_ABORT_MSG + #[tokio::test] + async fn aborting_pending_detached_launch_releases_id_and_discards_late_result() { + let (write_tx, _write_rx) = winapi_signaled_mpsc_channel().expect("create outbound channel"); + let (mut processor, mut notifications) = MessageProcessor::new(NowChannelCapsetMsg::default(), write_tx, None); + let old_launch = Arc::new(AtomicBool::new(true)); + processor.pending_sessions.insert(1, Arc::clone(&old_launch)); + + processor + .process_message(now_proto_pdu::NowExecAbortMsg::new(1, 7).into()) + .await + .expect("abort pending launch"); + processor + .ensure_session_id_free(1) + .await + .expect("aborted ID is reusable"); + + processor.pending_sessions.insert(1, Arc::new(AtomicBool::new(true))); + processor + .io_notification_tx + .send(ServerChannelEvent::ProcessNotification { + active: old_launch, + event: Box::new(ServerChannelEvent::SessionExited { + session_id: 1, + exit_code: 0, + }), + }) + .await + .expect("queue late launch result"); + assert!( + notifications + .recv() + .await + .expect("late result") + .into_active_event() + .is_none() + ); + assert!( + processor.pending_sessions.contains_key(&1), + "replacement stays reserved" + ); + } + + // oracle: regression #2191 renegotiation must isolate queued and future old-worker events from reused session IDs. + #[tokio::test] + async fn renegotiation_isolates_worker_notifications() { + let (write_tx, _write_rx) = winapi_signaled_mpsc_channel().expect("create outbound channel"); + let (mut processor, mut notifications) = + MessageProcessor::new(NowChannelCapsetMsg::default(), write_tx.clone(), None); + let old_sender = processor.io_notification_tx.clone(); + processor.pending_sessions.insert(1, Arc::new(AtomicBool::new(true))); + old_sender + .send(ServerChannelEvent::SessionExited { + session_id: 1, + exit_code: 3, + }) + .await + .expect("queue old result"); + assert!(!notifications.is_empty(), "old result is queued before renegotiation"); + + (processor, notifications) = MessageProcessor::new(NowChannelCapsetMsg::default(), write_tx, None); + processor.pending_sessions.insert(1, Arc::new(AtomicBool::new(true))); + assert!( + old_sender + .send(ServerChannelEvent::SessionExited { + session_id: 1, + exit_code: 4 + }) + .await + .is_err() + ); + assert!(matches!( + notifications.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + processor + .io_notification_tx + .send(ServerChannelEvent::SessionStarted { + session_id: 1, + no_io_redirection: true, + }) + .await + .expect("send new session event"); + assert!(matches!( + notifications.recv().await, + Some(ServerChannelEvent::SessionStarted { session_id: 1, .. }) + )); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §Elevated Execution + #[test] + fn elevated_batch_no_exit_keeps_console_open_when_redirection_is_unavailable() { + assert_eq!(batch_command_kind(true, true, true, false), "/K"); + } + + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §NOW_EXEC_BATCH_MSG + #[test] + fn detached_batch_no_exit_keeps_console_open_despite_redirection_flag() { + assert_eq!(batch_command_kind(true, true, false, true), "/K"); + } // oracle: regression #2156 detached Process must leave the channel alive for subsequent execution. #[tokio::test] @@ -1993,14 +2229,17 @@ mod tests { let client_caps = NowChannelCapsetMsg::default() .with_system_capset(NowSystemCapsetFlags::SHUTDOWN) .with_session_capset(NowSessionCapsetFlags::LOCK | NowSessionCapsetFlags::MSGBOX) - .with_exec_capset(NowExecCapsetFlags::STYLE_RUN | NowExecCapsetFlags::STYLE_SHELL) + .with_exec_capset( + NowExecCapsetFlags::STYLE_RUN | NowExecCapsetFlags::STYLE_SHELL | NowExecCapsetFlags::ELEVATION, + ) .with_heartbeat_interval(Duration::from_secs(30)) .expect("30 s is a valid heartbeat interval"); + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §NOW_CHANNEL_CAPSET_MSG let expected_caps = NowChannelCapsetMsg::default() .with_system_capset(NowSystemCapsetFlags::SHUTDOWN) .with_session_capset(NowSessionCapsetFlags::LOCK | NowSessionCapsetFlags::MSGBOX) - .with_exec_capset(NowExecCapsetFlags::STYLE_RUN) + .with_exec_capset(NowExecCapsetFlags::STYLE_RUN | NowExecCapsetFlags::ELEVATION) .with_heartbeat_interval(Duration::from_secs(30)) .expect("30 s is a valid heartbeat interval");