From df7f6b24ef034b3974ec9fa2b60dc2aefa374186 Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Mon, 9 Nov 2026 16:05:16 -0500 Subject: [PATCH 1/8] feat(agent): support elevated NOW execution --- Cargo.lock | 57 ++--- crates/win-api-wrappers/src/process.rs | 27 ++- devolutions-session/Cargo.toml | 2 +- .../src/dvc/now_message_dissector.rs | 7 +- devolutions-session/src/dvc/process.rs | 218 +++++++++++++++++- devolutions-session/src/dvc/task.rs | 214 +++++++++++++---- 6 files changed, 423 insertions(+), 102 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f9e43bfc72..ed37f9f104 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 a2f3598740..405c9c342d 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(), diff --git a/devolutions-session/Cargo.toml b/devolutions-session/Cargo.toml index b4f86bddea..87deabe711 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 4fe8dbe80b..99549c7a74 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 de3eeebf45..10e4e0c19d 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -3,7 +3,7 @@ use std::os::windows::io::AsRawHandle as _; 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 +13,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_BROKEN_PIPE, ERROR_CANCELLED, ERROR_HANDLE_EOF, ERROR_NOT_FOUND, 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 +27,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::{ @@ -90,6 +90,7 @@ pub enum ServerChannelEvent { CloseChannel, SessionStarted { session_id: u32, + no_io_redirection: bool, }, SessionDataOut { session_id: u32, @@ -127,6 +128,9 @@ 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. @@ -178,7 +182,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. @@ -211,6 +218,10 @@ impl WinApiProcessCtx { ProcessIoInputEvent::TerminateIo => { info!(session_id, "Terminating IO loop"); + if self.elevated { + self.process.wait(None)?; + } + // Terminate IO loop return Err(ExecError::Aborted); } @@ -303,7 +314,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 { @@ -614,6 +628,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 +641,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 +672,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()); @@ -698,6 +720,22 @@ impl WinApiProcessBuilder { let io_redirection = self.enable_io_redirection; let encoding = self.encoding; + 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, + self.executable, + self.command_line, + self.current_directory, + io_redirection, + io_notification_tx, + detached, + temp_files, + ); + } let process_ctx = if io_redirection { prepare_process_with_io_redirection(session_id, command_line, current_directory, self.env, encoding, job)? @@ -775,17 +813,97 @@ impl WinApiProcessBuilder { })) } + fn run_elevated_impl( + session_id: u32, + executable: String, + command_line: String, + current_directory: String, + requested_io_redirection: bool, + io_notification_tx: Sender, + detached: bool, + temp_files: Vec, + ) -> Result, ExecError> { + 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"); + } + + if let Err(error) = process_ctx.process.wait(None) { + error!(%error, session_id, "Failed to wait for elevated detached process"); + } + }) + .map_err(anyhow::Error::from)?; + + return Ok(None); + } + + let (input_event_tx, input_event_rx) = winapi_signaled_mpsc_channel()?; + 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 { + 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(io_notification_tx), job, false)? .expect("result should be non-optional when running in non-detached mode")) } @@ -793,10 +911,20 @@ 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 sender = self.elevated.then_some(io_notification_tx); + self.run_impl(session_id, sender, None, true)?; + Ok(()) + } } /// Job object whose processes are killed when it is dropped, or when the host exits or crashes. @@ -927,6 +1055,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,12 +1132,57 @@ 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 pid = process.process_id()?; + + Ok(WinApiProcessCtx { + session_id, + stdout_read_pipe: None, + stderr_read_pipe: None, + stdin_write_pipe: None, + encoding: DataEncoding::Raw, + pid, + started_no_io_redirection: requested_io_redirection, + elevated: true, + _temp_files: temp_files, + process, + }) +} + +fn map_elevated_launch_error(error: anyhow::Error) -> ExecError { + match error.downcast::() { + Ok(error) if error.code() == ERROR_CANCELLED.to_hresult() => { + ExecError::NowStatus(NowStatusError::new_proto(NowProtoError::ElevationDenied)) + } + Ok(error) => ExecError::from(error), + Err(error) => ExecError::Other(error), + } +} + /// Represents spawned process with IO redirection. pub struct WinApiProcess { input_event_tx: WinapiSignaledSender, @@ -1072,8 +1248,26 @@ impl WinApiProcess { mod tests { use std::time::Duration; + use now_proto_pdu::NowExecResultMsg; + use super::*; + // 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 +1295,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(); @@ -1148,7 +1342,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 2a5b1618f4..e6c6f65e87 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use anyhow::{Context, bail}; @@ -419,9 +419,16 @@ async fn process_messages( match task_rx { Some(notification) => { match notification { - ServerChannelEvent::SessionStarted { session_id } => { + 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 +448,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 +457,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 +548,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 +634,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: HashSet, /// 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. @@ -653,6 +665,7 @@ impl MessageProcessor { io_notification_tx, capabilities, sessions: HashMap::new(), + pending_sessions: HashSet::new(), process_job: None, rdm_handler, pwsh_executable, @@ -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(&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,43 @@ impl MessageProcessor { Ok(()) } + async fn run_detached_exec( + &mut self, + process: WinApiProcessBuilder, + session_id: u32, + elevated: bool, + ) -> Result<(), ExecError> { + if elevated { + self.pending_sessions.insert(session_id); + } + + if let Err(error) = process.run_detached_with_notifications(session_id, self.io_notification_tx.clone()) { + self.pending_sessions.remove(&session_id); + return Err(error); + } + + if !elevated { + 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 +778,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 { @@ -919,11 +991,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 +1047,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 +1068,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 +1091,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 +1112,12 @@ 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(), + ); + 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 +1128,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 +1151,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 +1196,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 +1219,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 +1275,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 +1360,14 @@ impl MessageProcessor { } } +fn batch_command_kind(no_exit: bool, io_redirection: bool, elevated: bool) -> &'static str { + if no_exit && (!io_redirection || elevated) { + "/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()); @@ -1570,11 +1690,18 @@ mod tests { 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, + 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; + // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §NOW_EXEC_BATCH_MSG §Elevated Execution + #[test] + fn elevated_batch_no_exit_keeps_console_open_when_redirection_is_unavailable() { + assert_eq!(batch_command_kind(true, true, true), "/K"); + } + // oracle: regression #2156 detached Process must leave the channel alive for subsequent execution. #[tokio::test] async fn detached_process_keeps_channel_available_for_execution() { @@ -1993,14 +2120,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 §Elevated Execution 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"); From e47c7b2b915874a90045ebeef42b55287630112b Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Mon, 9 Nov 2026 21:38:12 -0500 Subject: [PATCH 2/8] fix(agent): resolve elevated execution CI checks --- devolutions-session/src/dvc/process.rs | 35 +++++++++++--------------- devolutions-session/src/dvc/task.rs | 4 +-- 2 files changed, 16 insertions(+), 23 deletions(-) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index 10e4e0c19d..cbfc20ad68 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -714,29 +714,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); - - let io_redirection = self.enable_io_redirection; - let encoding = self.encoding; 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, - self.executable, - self.command_line, - self.current_directory, - io_redirection, - io_notification_tx, - detached, - temp_files, - ); + 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 { @@ -814,15 +802,20 @@ impl WinApiProcessBuilder { } fn run_elevated_impl( + self, session_id: u32, - executable: String, - command_line: String, - current_directory: String, - requested_io_redirection: bool, io_notification_tx: Sender, detached: bool, - temp_files: Vec, ) -> 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")) diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index e6c6f65e87..84df84abf5 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -1696,7 +1696,7 @@ mod tests { use crate::dvc::channel::{bounded_mpsc_channel, winapi_signaled_mpsc_channel}; use crate::dvc::io::DvcInitializationProgress; - // oracle: spec Devolutions/now-libraries@4a24d1e50d1b952ff42525c9f34478abbc82746:protocols/docs/NOW-spec.md §NOW_EXEC_BATCH_MSG §Elevated Execution + // 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), "/K"); @@ -2126,7 +2126,7 @@ mod tests { .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 §Elevated Execution + // 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) From 59bda9a1527208dcfa66249bfa7df13daa10b12d Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 08:34:16 -0500 Subject: [PATCH 3/8] fix(agent): preserve elevated execution lifecycle --- crates/win-api-wrappers/src/process.rs | 14 +- devolutions-session/src/dvc/process.rs | 206 ++++++++++++++++++++++--- devolutions-session/src/dvc/task.rs | 72 +++++++-- 3 files changed, 249 insertions(+), 43 deletions(-) diff --git a/crates/win-api-wrappers/src/process.rs b/crates/win-api-wrappers/src/process.rs index 405c9c342d..3c71e6e4e3 100644 --- a/crates/win-api-wrappers/src/process.rs +++ b/crates/win-api-wrappers/src/process.rs @@ -1139,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/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index cbfc20ad68..b159aa4687 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -13,8 +13,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_CANCELLED, 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}; @@ -137,6 +137,17 @@ pub struct WinApiProcessCtx { 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 + && 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!( @@ -155,15 +166,24 @@ impl WinApiProcessCtx { "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(()) } @@ -218,10 +238,6 @@ impl WinApiProcessCtx { ProcessIoInputEvent::TerminateIo => { info!(session_id, "Terminating IO loop"); - if self.elevated { - self.process.wait(None)?; - } - // Terminate IO loop return Err(ExecError::Aborted); } @@ -845,9 +861,8 @@ impl WinApiProcessBuilder { error!(%error, session_id, "Failed to send elevated launch result"); } - if let Err(error) = process_ctx.process.wait(None) { - error!(%error, session_id, "Failed to wait for elevated detached process"); - } + // Dropping the context waits for exit before releasing temporary scripts. + drop(process_ctx); }) .map_err(anyhow::Error::from)?; @@ -1150,27 +1165,38 @@ fn prepare_elevated_process( SW_RESTORE, ) .map_err(map_elevated_launch_error)?; - let pid = process.process_id()?; - - Ok(WinApiProcessCtx { + let mut process_ctx = WinApiProcessCtx { session_id, stdout_read_pipe: None, stderr_read_pipe: None, stdin_write_pipe: None, encoding: DataEncoding::Raw, - pid, + 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.code() == ERROR_CANCELLED.to_hresult() => { + 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), } @@ -1245,6 +1271,142 @@ mod tests { use super::*; + // 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.clone()); + 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, 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.clone()); + // 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, 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") + } + + // 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() { diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index 84df84abf5..6393432248 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -274,8 +274,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 +346,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 +387,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(), ); @@ -656,11 +648,11 @@ 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, @@ -671,7 +663,8 @@ impl MessageProcessor { pwsh_executable, window_monitor_shutdown_tx: None, window_monitor_handle: None, - } + }; + (processor, task_rx) } fn process_job(&mut self) -> Result<&ProcessJob, ExecError> { @@ -1689,12 +1682,59 @@ mod tests { use windows::Win32::System::Threading::{PROCESS_SYNCHRONIZE, PROCESS_TERMINATE}; use super::{ - ChannelOutcome, DEFAULT_HEARTBEAT_INTERVAL, MAX_CONSECUTIVE_FAILED_REOPENS, REOPEN_DELAY_CAP, + 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: 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); + 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); + 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] From 4ab831cf18eca19a5432223bb7b3eaf501bc432c Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 08:42:31 -0500 Subject: [PATCH 4/8] fix(agent): discard events from aborted executions --- devolutions-session/src/dvc/process.rs | 120 ++++++++++++++++++++++--- devolutions-session/src/dvc/task.rs | 19 +++- 2 files changed, 125 insertions(+), 14 deletions(-) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index b159aa4687..197b71abe0 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -1,5 +1,7 @@ use std::collections::HashMap; use std::os::windows::io::AsRawHandle as _; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; @@ -87,6 +89,10 @@ 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, @@ -118,6 +124,40 @@ pub enum ServerChannelEvent { }, } +#[derive(Clone)] +struct ProcessNotifications { + sender: Sender, + active: Arc, +} + +impl ProcessNotifications { + fn new(sender: Sender) -> Self { + Self { + sender, + active: Arc::new(AtomicBool::new(true)), + } + } + + 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, @@ -160,7 +200,7 @@ 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" @@ -188,10 +228,10 @@ impl WinApiProcessCtx { 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; @@ -254,10 +294,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; @@ -713,6 +753,7 @@ impl WinApiProcessBuilder { job: Option<&ProcessJob>, detached: bool, ) -> Result, ExecError> { + let io_notification_tx = io_notification_tx.map(ProcessNotifications::new); let command_line = format!("\"{}\" {}", self.executable, self.command_line) .trim_end() .to_owned(); @@ -791,6 +832,7 @@ 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 join_handle = std::thread::spawn(move || { let run_result = if io_redirection { @@ -810,6 +852,7 @@ impl WinApiProcessBuilder { }); Ok(Some(WinApiProcess { + notifications_active, input_event_tx, stdin_writer, join_handle, @@ -820,7 +863,7 @@ impl WinApiProcessBuilder { fn run_elevated_impl( self, session_id: u32, - io_notification_tx: Sender, + io_notification_tx: ProcessNotifications, detached: bool, ) -> Result, ExecError> { let Self { @@ -870,6 +913,7 @@ impl WinApiProcessBuilder { } let (input_event_tx, input_event_rx) = winapi_signaled_mpsc_channel()?; + let notifications_active = Arc::clone(&io_notification_tx.active); let join_handle = std::thread::Builder::new() .name(format!("process-{session_id}-elevation")) .spawn(move || { @@ -894,6 +938,7 @@ impl WinApiProcessBuilder { })?; Ok(Some(WinApiProcess { + notifications_active, input_event_tx, stdin_writer: None, join_handle, @@ -1204,6 +1249,7 @@ fn map_elevated_launch_error(error: anyhow::Error) -> ExecError { /// Represents spawned process with IO redirection. pub struct WinApiProcess { + notifications_active: Arc, input_event_tx: WinapiSignaledSender, /// `None` without IO redirection. stdin_writer: Option, @@ -1213,6 +1259,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); @@ -1271,12 +1319,54 @@ mod tests { use super::*; + // 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), + 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.clone()); + 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; @@ -1287,7 +1377,7 @@ mod tests { drop(notification_rx); let (done_tx, done_rx) = std::sync::mpsc::channel(); let worker = std::thread::spawn(move || { - let result = context.wait(input_rx, notification_tx); + let result = context.wait(input_rx, ProcessNotifications::new(notification_tx)); done_tx.send(result).expect("report worker result"); }); @@ -1311,7 +1401,7 @@ mod tests { 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.clone()); + 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; @@ -1320,7 +1410,7 @@ mod tests { 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, notification_tx); + let result = context.wait(input_rx, ProcessNotifications::new(notification_tx)); done_tx.send(result).expect("report exit"); }); assert!(matches!( @@ -1363,6 +1453,8 @@ mod tests { .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 @@ -1459,7 +1551,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, diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index 6393432248..821331a1fb 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -410,7 +410,13 @@ 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::ProcessNotification { .. } => { + unreachable!("process notifications are unwrapped before dispatch"); + } ServerChannelEvent::SessionStarted { session_id, no_io_redirection, @@ -1109,6 +1115,7 @@ impl MessageProcessor { 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()); @@ -1353,8 +1360,8 @@ impl MessageProcessor { } } -fn batch_command_kind(no_exit: bool, io_redirection: bool, elevated: bool) -> &'static str { - if no_exit && (!io_redirection || elevated) { +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" @@ -1739,7 +1746,13 @@ mod tests { // 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), "/K"); + 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. From 67f578b6950be42aeb72f9e8cfe5e19656e29103 Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 08:49:27 -0500 Subject: [PATCH 5/8] fix(agent): bound elevated launch cleanup --- devolutions-session/src/dvc/process.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index 197b71abe0..547eaa4e1b 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -181,6 +181,7 @@ 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"); @@ -1319,6 +1320,24 @@ mod tests { use super::*; + // 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() { From a620fba5cd6b44bd617cceb6b08b4bec56dc6352 Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 08:57:50 -0500 Subject: [PATCH 6/8] fix(agent): forget aborted detached elevations --- devolutions-session/src/dvc/process.rs | 18 ++++-- devolutions-session/src/dvc/task.rs | 77 +++++++++++++++++++++----- 2 files changed, 74 insertions(+), 21 deletions(-) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index 547eaa4e1b..670aa0ff5e 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -750,11 +750,10 @@ 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> { - let io_notification_tx = io_notification_tx.map(ProcessNotifications::new); let command_line = format!("\"{}\" {}", self.executable, self.command_line) .trim_end() .to_owned(); @@ -957,7 +956,12 @@ impl WinApiProcessBuilder { job: Option<&ProcessJob>, ) -> Result { Ok(self - .run_impl(session_id, Some(io_notification_tx), 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")) } @@ -973,11 +977,13 @@ impl WinApiProcessBuilder { self, session_id: u32, io_notification_tx: Sender, - ) -> Result<(), ExecError> { + ) -> Result, ExecError> { // Result always empty and therefore ignored in detached mode. - let sender = self.elevated.then_some(io_notification_tx); + 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(()) + Ok(active) } } diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index 821331a1fb..d794d2a37f 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -1,5 +1,6 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, bail}; use async_trait::async_trait; @@ -636,7 +637,7 @@ struct MessageProcessor { /// 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: HashSet, + 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. @@ -663,7 +664,7 @@ impl MessageProcessor { io_notification_tx, capabilities, sessions: HashMap::new(), - pending_sessions: HashSet::new(), + pending_sessions: HashMap::new(), process_job: None, rdm_handler, pwsh_executable, @@ -681,7 +682,7 @@ impl MessageProcessor { } async fn ensure_session_id_free(&self, session_id: u32) -> Result<(), ExecError> { - if self.sessions.contains_key(&session_id) || self.pending_sessions.contains(&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))); } @@ -717,16 +718,12 @@ impl MessageProcessor { session_id: u32, elevated: bool, ) -> Result<(), ExecError> { - if elevated { - self.pending_sessions.insert(session_id); - } - - if let Err(error) = process.run_detached_with_notifications(session_id, self.io_notification_tx.clone()) { - self.pending_sessions.remove(&session_id); - return Err(error); - } + let notifications_active = + process.run_detached_with_notifications(session_id, self.io_notification_tx.clone())?; - if !elevated { + if elevated { + self.pending_sessions.insert(session_id, notifications_active); + } else { self.send_detached_process_success(session_id).await?; } @@ -819,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 => { @@ -1675,6 +1677,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}; @@ -1697,6 +1701,49 @@ mod tests { 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() { @@ -1704,7 +1751,7 @@ mod tests { 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); + processor.pending_sessions.insert(1, Arc::new(AtomicBool::new(true))); old_sender .send(ServerChannelEvent::SessionExited { session_id: 1, @@ -1715,7 +1762,7 @@ mod tests { assert!(!notifications.is_empty(), "old result is queued before renegotiation"); (processor, notifications) = MessageProcessor::new(NowChannelCapsetMsg::default(), write_tx, None); - processor.pending_sessions.insert(1); + processor.pending_sessions.insert(1, Arc::new(AtomicBool::new(true))); assert!( old_sender .send(ServerChannelEvent::SessionExited { From e92f905f4f4eac030bce1d03f4ece6c739a63265 Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 09:07:18 -0500 Subject: [PATCH 7/8] fix(agent): keep execution control nonblocking --- devolutions-session/src/dvc/process.rs | 34 +++++++++++++++++++++++--- devolutions-session/src/dvc/task.rs | 13 ++++++++-- 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index 670aa0ff5e..967ae21623 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -1277,12 +1277,11 @@ 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 + .try_send(ProcessIoInputEvent::AbortExecution(exit_code)) } 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. @@ -1309,7 +1308,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(()) } @@ -1326,6 +1325,33 @@ mod tests { use super::*; + // 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)), + 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_err("full queue cannot accept abort"); + 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() { diff --git a/devolutions-session/src/dvc/task.rs b/devolutions-session/src/dvc/task.rs index d794d2a37f..d0496e1878 100644 --- a/devolutions-session/src/dvc/task.rs +++ b/devolutions-session/src/dvc/task.rs @@ -829,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. @@ -854,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(); From 31316abcefb20f5c123d02c02ca3db7f7dca7b61 Mon Sep 17 00:00:00 2001 From: Richard Markiewicz Date: Tue, 10 Nov 2026 09:15:22 -0500 Subject: [PATCH 8/8] fix(agent): deliver aborts through full control queues --- devolutions-session/src/dvc/process.rs | 88 +++++++++++++++++++++++++- 1 file changed, 85 insertions(+), 3 deletions(-) diff --git a/devolutions-session/src/dvc/process.rs b/devolutions-session/src/dvc/process.rs index 967ae21623..33ff10a06a 100644 --- a/devolutions-session/src/dvc/process.rs +++ b/devolutions-session/src/dvc/process.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::os::windows::io::AsRawHandle as _; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; @@ -128,6 +128,7 @@ pub enum ServerChannelEvent { struct ProcessNotifications { sender: Sender, active: Arc, + abort_exit_code: Arc, } impl ProcessNotifications { @@ -135,6 +136,7 @@ impl ProcessNotifications { Self { sender, active: Arc::new(AtomicBool::new(true)), + abort_exit_code: Arc::new(AtomicU64::new(0)), } } @@ -259,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"); @@ -388,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"); @@ -833,6 +853,7 @@ 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 { @@ -853,6 +874,7 @@ impl WinApiProcessBuilder { Ok(Some(WinApiProcess { notifications_active, + abort_exit_code, input_event_tx, stdin_writer, join_handle, @@ -914,6 +936,7 @@ impl WinApiProcessBuilder { 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 || { @@ -939,6 +962,7 @@ impl WinApiProcessBuilder { Ok(Some(WinApiProcess { notifications_active, + abort_exit_code, input_event_tx, stdin_writer: None, join_handle, @@ -1257,6 +1281,7 @@ fn map_elevated_launch_error(error: anyhow::Error) -> ExecError { /// 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, @@ -1276,8 +1301,23 @@ impl Drop for WinApiProcess { impl WinApiProcess { pub async fn abort_execution(&self, exit_code: u32) -> anyhow::Result<()> { - self.input_event_tx + // 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<()> { @@ -1325,6 +1365,46 @@ mod tests { 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() { @@ -1332,6 +1412,7 @@ mod tests { 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(|| {}), @@ -1345,7 +1426,7 @@ mod tests { tokio::time::timeout(Duration::from_secs(1), process.abort_execution(7)) .await .expect("abort must not wait for consent") - .expect_err("full queue cannot accept abort"); + .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") @@ -1378,6 +1459,7 @@ mod tests { 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(|| {}),