diff --git a/Cargo.lock b/Cargo.lock index 0240888a9..4f0443c24 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1530,7 +1530,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.118", ] [[package]] @@ -2296,7 +2296,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2612,7 +2612,7 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.1.3", + "windows-link 0.2.1", "windows-result 0.4.1", ] @@ -3119,7 +3119,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite 0.2.17", - "socket2 0.5.10", + "socket2 0.6.5", "system-configuration", "tokio 1.52.3", "tower-service", @@ -3658,7 +3658,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -4749,11 +4749,10 @@ dependencies = [ [[package]] name = "now-policy-api" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca52de7d57fca9ea49cc10f229bdc31cfbdb37f7d49a6e7407fa60248d1df7c6" +checksum = "11808c712bda38e8cee0ec0a6902bbfe63583f475608ee6bcf8ffd848473ab06" dependencies = [ - "base64 0.22.1", "chrono", "derive_more", "now-policy", @@ -4767,9 +4766,9 @@ dependencies = [ [[package]] name = "now-policy-server-template" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ce816e494a5034becfe105f30611284431275c24bdad3a8e7ef71a7b31e6e1" +checksum = "d4480ae83e4c1302f47f3b033804b9b2f5054b5fb8c0079a8bf7c63c8a813d7a" dependencies = [ "aide", "async-trait", @@ -4809,7 +4808,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5962,7 +5961,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.3", "rustls 0.23.42", - "socket2 0.5.10", + "socket2 0.6.5", "thiserror 2.0.18", "tokio 1.52.3", "tracing", @@ -6002,9 +6001,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.5", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6501,7 +6500,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6601,7 +6600,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7129,7 +7128,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7422,7 +7421,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8864,7 +8863,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/devolutions-agent/Cargo.toml b/devolutions-agent/Cargo.toml index e5604f523..ef72213e4 100644 --- a/devolutions-agent/Cargo.toml +++ b/devolutions-agent/Cargo.toml @@ -42,8 +42,8 @@ http-client-proxy = { path = "../crates/http-client-proxy" } ipnetwork = "0.20" notify = { version = "7", default-features = false, features = ["macos_kqueue"] } now-policy = "0.2" -now-policy-api = { version = "0.2", features = ["policy-compat"] } -now-policy-server-template = { version = "0.2", features = ["policy-compat"] } +now-policy-api = { version = "0.3", features = ["policy-compat"] } +now-policy-server-template = { version = "0.3", features = ["policy-compat"] } parking_lot = "0.12" prost = "0.13" prost-types = "0.13" @@ -123,6 +123,7 @@ features = [ "Win32_System_Pipes", "Win32_System_RemoteDesktop", "Win32_System_Com", + "Win32_System_Console", "Win32_UI_Shell", ] diff --git a/devolutions-agent/src/broker/auth.rs b/devolutions-agent/src/broker/auth.rs index ce0525cbc..009c90de3 100644 --- a/devolutions-agent/src/broker/auth.rs +++ b/devolutions-agent/src/broker/auth.rs @@ -3,7 +3,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context as _, bail}; -use now_policy_api::{ClientContext, PackageRequest, StatusRequest}; +use now_policy_api::{CancelRequest, ClientContext, PackageRequest, StatusRequest}; use tokio::net::windows::named_pipe::NamedPipeServer; use tracing::{debug, warn}; use widestring::U16CString; @@ -74,6 +74,15 @@ impl PipeClient { self.validate_signature(skip_signature_validation) } + pub(crate) fn validate_cancel_request( + &self, + request: &CancelRequest, + skip_signature_validation: bool, + ) -> anyhow::Result<()> { + self.validate_client_context(&request.client)?; + self.validate_signature(skip_signature_validation) + } + fn validate_client_context(&self, client: &ClientContext) -> anyhow::Result<()> { self.validate_effective_user(&client.effective_user)?; self.validate_executable_path(&client.client_executable_path) diff --git a/devolutions-agent/src/broker/executor/mod.rs b/devolutions-agent/src/broker/executor/mod.rs index 2cd5cd745..f7cd0c39a 100644 --- a/devolutions-agent/src/broker/executor/mod.rs +++ b/devolutions-agent/src/broker/executor/mod.rs @@ -5,6 +5,7 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; use now_policy_api::{Elevation, ManagerName, Scope}; +use tokio_util::sync::CancellationToken; use tracing::info; use win_api_wrappers::identity::sid::Sid; @@ -52,7 +53,34 @@ pub struct ExecutionContext { /// Installation scope (machine scope requires elevation). pub scope: Option, /// When true, capture the main command's combined stdout+stderr. + /// + /// Since API v0.3 this only controls whether output frames are pushed on the + /// per-operation event channel; output is no longer returned in status responses. pub capture_output: bool, + /// Fired when the client requests cancellation of the operation. + /// + /// The executor terminates the in-flight process (gracefully first, forcefully + /// after a grace period) and reports the cancellation through an error chain + /// containing [`OperationCanceled`]. + pub cancel_token: CancellationToken, +} + +/// Marker error reported by executors when an operation was terminated because its +/// cancellation was requested through [`ExecutionContext::cancel_token`]. +#[derive(Debug)] +pub struct OperationCanceled; + +impl core::fmt::Display for OperationCanceled { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + write!(f, "operation was canceled at the client's request") + } +} + +impl core::error::Error for OperationCanceled {} + +/// Whether an executor error chain denotes a client-requested cancellation. +pub fn is_canceled_error(error: &anyhow::Error) -> bool { + error.chain().any(|cause| cause.is::()) } pub type ProcessStartedCallback = std::sync::Arc) + Send + Sync>; diff --git a/devolutions-agent/src/broker/executor/windows/mod.rs b/devolutions-agent/src/broker/executor/windows/mod.rs index abaac8120..a6da675e9 100644 --- a/devolutions-agent/src/broker/executor/windows/mod.rs +++ b/devolutions-agent/src/broker/executor/windows/mod.rs @@ -18,7 +18,10 @@ use win_api_wrappers::token::Token; use win_api_wrappers::utils::WideString; use windows::Win32::Security::TOKEN_ALL_ACCESS; -use super::{BROKER_SUPPORTED_MANAGERS, CommandExecutor, ExecutionContext, ExecutionOutput, ProcessStartedCallback}; +use super::{ + BROKER_SUPPORTED_MANAGERS, CommandExecutor, ExecutionContext, ExecutionOutput, OperationCanceled, + ProcessStartedCallback, is_canceled_error, +}; use crate::broker::policy_security; mod privileges; @@ -328,6 +331,10 @@ fn run_plan( session_id: u32, process_started: Option, ) -> anyhow::Result { + if ctx.cancel_token.is_cancelled() { + return Err(anyhow::Error::new(OperationCanceled)); + } + let requires_elevation = ctx.elevation == Elevation::Elevated || ctx.scope == Some(Scope::Machine); if requires_elevation && command_is_bun(&ctx.command) { bail!("elevated Bun package operations are not supported by the broker"); @@ -347,7 +354,8 @@ fn run_plan( } } - // 1. Kill requested processes (best-effort; a missing process is not an error). + // 1. Kill requested processes (best-effort; a missing process is not an error, + // but a cancellation request must still be honored). for process_name in &ctx.kill_processes { let kill_cmd = vec![ trusted_system32_executable("taskkill.exe"), @@ -355,8 +363,17 @@ fn run_plan( "/IM".to_owned(), process_name.clone(), ]; - match create_process(token, &kill_cmd, session_id, false, requires_elevation, None) { + match create_process( + token, + &kill_cmd, + session_id, + false, + requires_elevation, + None, + Some(&ctx.cancel_token), + ) { Ok(out) => info!(%process_name, exit_code = out.exit_code, "Kill-before-operation completed"), + Err(error) if is_canceled_error(&error) => return Err(error), Err(error) => warn!(%process_name, %error, "Kill-before-operation failed (ignored)"), } } @@ -372,6 +389,7 @@ fn run_plan( ctx.capture_output, requires_elevation, None, + Some(&ctx.cancel_token), ) .context("failed to run pre-operation command")?; if out.exit_code != 0 { @@ -392,19 +410,35 @@ fn run_plan( ctx.capture_output, requires_elevation, process_started, + Some(&ctx.cancel_token), )?; // 4. Post-operation command — runs after the main command; failures are logged only // so a completed main operation is never reported as failed by its post-hook. + // Skipped when cancellation was requested, and cancellable while it runs. if let Some(post) = &ctx.post_command { - info!("Running post-operation command"); - match prepare_shell_command(token, post) { - Ok(command) => match create_process(token, command.args(), session_id, false, requires_elevation, None) { - Ok(out) if out.exit_code == 0 => {} - Ok(out) => warn!(exit_code = out.exit_code, "Post-operation command exited non-zero"), - Err(error) => warn!(%error, "Post-operation command failed"), - }, - Err(error) => warn!(%error, "Failed to prepare post-operation command"), + if ctx.cancel_token.is_cancelled() { + info!("Skipping post-operation command: operation was canceled"); + } else { + info!("Running post-operation command"); + match prepare_shell_command(token, post) { + Ok(command) => { + match create_process( + token, + command.args(), + session_id, + false, + requires_elevation, + None, + Some(&ctx.cancel_token), + ) { + Ok(out) if out.exit_code == 0 => {} + Ok(out) => warn!(exit_code = out.exit_code, "Post-operation command exited non-zero"), + Err(error) => warn!(%error, "Post-operation command failed"), + } + } + Err(error) => warn!(%error, "Failed to prepare post-operation command"), + } } } @@ -1918,6 +1952,7 @@ mod tests { elevation: Elevation::Elevated, scope: Some(Scope::User), capture_output: false, + cancel_token: tokio_util::sync::CancellationToken::new(), }; let error = reject_unsupported_vcpkg_elevation(&ctx).expect_err("elevated vcpkg should fail"); @@ -1941,6 +1976,7 @@ mod tests { elevation: Elevation::Elevated, scope: Some(Scope::User), capture_output: false, + cancel_token: tokio_util::sync::CancellationToken::new(), }; let executor = WindowsExecutor { is_system: true }; @@ -1968,6 +2004,7 @@ mod tests { elevation: Elevation::Standard, scope: Some(Scope::User), capture_output: false, + cancel_token: tokio_util::sync::CancellationToken::new(), }; let error = execute_as_current_user(&ctx, None).expect_err("mismatched client SID should fail"); diff --git a/devolutions-agent/src/broker/executor/windows/process.rs b/devolutions-agent/src/broker/executor/windows/process.rs index 330a981ab..6d928a472 100644 --- a/devolutions-agent/src/broker/executor/windows/process.rs +++ b/devolutions-agent/src/broker/executor/windows/process.rs @@ -2,9 +2,12 @@ use std::io::Read as _; use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; use anyhow::{Context as _, bail}; use chrono::Utc; +use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use win_api_wrappers::process::{self, StartupInfo}; use win_api_wrappers::security::attributes::SecurityAttributesInit; @@ -12,14 +15,23 @@ use win_api_wrappers::token::Token; use win_api_wrappers::utils::{self, CommandLine, Pipe, WideString}; use windows::Win32::Foundation::WAIT_TIMEOUT; use windows::Win32::System::Threading::{ - CREATE_NEW_CONSOLE, NORMAL_PRIORITY_CLASS, STARTF_USESHOWWINDOW, STARTF_USESTDHANDLES, + CREATE_NEW_CONSOLE, CREATE_NEW_PROCESS_GROUP, NORMAL_PRIORITY_CLASS, STARTF_USESHOWWINDOW, STARTF_USESTDHANDLES, }; use windows::Win32::UI::WindowsAndMessaging::SW_HIDE; -use crate::broker::executor::{ExecutionOutput, MAX_CAPTURED_OUTPUT_BYTES, ProcessStartedCallback, tail_utf8}; +use crate::broker::executor::{ + ExecutionOutput, MAX_CAPTURED_OUTPUT_BYTES, OperationCanceled, ProcessStartedCallback, tail_utf8, +}; use crate::broker::operation_tracker::OperationTracker; use crate::broker::policy_security; +/// How long a canceled process is given to exit after the graceful console +/// ctrl event before it is forcefully terminated. +const CANCEL_GRACE_PERIOD: Duration = Duration::from_secs(60); + +/// Granularity of the wait loop used to observe cancellation requests. +const WAIT_SLICE_MS: u32 = 500; + /// Create a process under the given token and wait for exit. /// /// This is the unified process-creation path used by both SYSTEM and current-user modes. @@ -29,6 +41,10 @@ use crate::broker::policy_security; /// [`crate::broker::executor::MAX_CAPTURED_OUTPUT_BYTES`]); otherwise no output is captured. /// /// Returns the process exit code and (when captured) its output. +/// +/// When `cancel` is provided and fires while the process is running, the process is +/// stopped (gracefully first, forcefully after [`CANCEL_GRACE_PERIOD`]) and an error +/// chain containing [`OperationCanceled`] is returned. #[allow(clippy::cast_possible_wrap)] pub(super) fn create_process( token: &Token, @@ -37,7 +53,17 @@ pub(super) fn create_process( capture: bool, requires_elevation: bool, process_started: Option, + cancel: Option<&CancellationToken>, ) -> anyhow::Result { + // Reject an already-canceled token before any setup or spawn so a cancellation + // arriving between plan stages never launches the next command. Races with a + // cancellation arriving after this check are handled by the wait loop below. + if let Some(cancel) = cancel + && cancel.is_cancelled() + { + return Err(anyhow::Error::new(OperationCanceled)); + } + let cmd_line = CommandLine::new(command.to_vec()); debug!(session_id, capture, "Building process creation parameters"); @@ -93,7 +119,9 @@ pub(super) fn create_process( (None, None, None) }; - let creation_flags = CREATE_NEW_CONSOLE | NORMAL_PRIORITY_CLASS; + // `CREATE_NEW_PROCESS_GROUP` makes the child's PID a process group ID so a + // console ctrl event can later target the whole group for graceful cancellation. + let creation_flags = CREATE_NEW_CONSOLE | CREATE_NEW_PROCESS_GROUP | NORMAL_PRIORITY_CLASS; debug!("Calling process::create_process_as_user"); @@ -166,45 +194,76 @@ pub(super) fn create_process( }) }); - let timeout_ms = operation_timeout_ms(); - if process_info - .process - .wait(Some(timeout_ms)) - .context("failed to wait for process")? - == WAIT_TIMEOUT - { - warn!( - session_id, - pid = process_info.process_id, - timeout_ms, - "Process timed out; terminating" - ); - process_info + let deadline = Instant::now() + OperationTracker::operation_timeout(); + // INVARIANT: whenever the loop breaks, the process has exited (or was terminated), + // so the capture reader below observes EOF and joining it cannot block. + let outcome = loop { + if let Some(cancel) = cancel + && cancel.is_cancelled() + { + stop_canceled_process(&process_info, session_id)?; + break WaitOutcome::Canceled; + } + + if process_info .process - .terminate(1) - .context("failed to terminate timed-out process")?; - let _ = process_info.process.wait(None); - bail!( - "operation timed out after {} seconds", - OperationTracker::operation_timeout().as_secs() - ); - } + .wait(Some(WAIT_SLICE_MS)) + .context("failed to wait for process")? + != WAIT_TIMEOUT + { + break WaitOutcome::Exited; + } - let exit_code = process_info - .process - .exit_code() - .context("failed to get process exit code")?; + if Instant::now() >= deadline { + warn!( + session_id, + pid = process_info.process_id, + "Process timed out; terminating" + ); + process_info + .process + .terminate(1) + .context("failed to terminate timed-out process")?; + let _ = process_info.process.wait(None); + break WaitOutcome::TimedOut; + } + }; + // Join the reader thread on every outcome so it is never left detached. let stdout = match reader { Some(handle) => tail_utf8(&handle.join().unwrap_or_default()), None => String::new(), }; - Ok(ExecutionOutput { - exit_code: exit_code as i32, - stdout, - started_at: Some(started_at), - }) + match outcome { + WaitOutcome::Canceled => Err(anyhow::Error::new(OperationCanceled)), + WaitOutcome::TimedOut => bail!( + "operation timed out after {} seconds", + OperationTracker::operation_timeout().as_secs() + ), + WaitOutcome::Exited => { + let exit_code = process_info + .process + .exit_code() + .context("failed to get process exit code")?; + + Ok(ExecutionOutput { + exit_code: exit_code as i32, + stdout, + started_at: Some(started_at), + }) + } + } +} + +/// How the wait loop in [`create_process`] concluded. +enum WaitOutcome { + /// The process exited on its own. + Exited, + /// The operation was canceled and the process was stopped. + Canceled, + /// The operation timeout elapsed and the process was terminated. + TimedOut, } /// Resolve an executable name to its full path using the given environment's PATH. @@ -278,8 +337,89 @@ fn resolve_executable( bail!("trusted executable '{exe_name}' not found in target user PATH"); } -fn operation_timeout_ms() -> u32 { - u32::try_from(OperationTracker::operation_timeout().as_millis()).unwrap_or(u32::MAX) +/// Stop a process whose operation was canceled. +/// +/// Attempts a graceful stop first by delivering a `CTRL_BREAK_EVENT` to the child's +/// process group, then waits up to [`CANCEL_GRACE_PERIOD`] for the process to exit. +/// If the event cannot be delivered or the grace period elapses, the root process is +/// forcefully terminated. +/// +/// Only the root process is terminated on fallback — deliberately. Package managers +/// spawn installer children (e.g. `msiexec`), and forcefully killing an MSI install +/// mid-way can corrupt the system; letting an in-flight installer finish is preferred +/// over guaranteeing the whole tree dies (job-object based tree termination was +/// considered and rejected for this reason). +fn stop_canceled_process(process_info: &process::ProcessInformation, session_id: u32) -> anyhow::Result<()> { + let pid = process_info.process_id; + info!(session_id, pid, "Cancellation requested; stopping process"); + + match send_ctrl_break(pid) { + Ok(()) => { + let grace_ms = u32::try_from(CANCEL_GRACE_PERIOD.as_millis()).expect("grace period fits into u32"); + if process_info + .process + .wait(Some(grace_ms)) + .context("failed to wait for canceled process")? + != WAIT_TIMEOUT + { + info!(session_id, pid, "Canceled process exited after ctrl event"); + return Ok(()); + } + warn!(session_id, pid, "Canceled process ignored ctrl event; terminating"); + } + Err(error) => { + warn!( + session_id, + pid, + error = format!("{error:#}"), + "Failed to deliver ctrl event to canceled process; terminating" + ); + } + } + + process_info + .process + .terminate(1) + .context("failed to terminate canceled process")?; + let _ = process_info.process.wait(None); + info!(session_id, pid, "Canceled process terminated"); + + Ok(()) +} + +/// Deliver a `CTRL_BREAK_EVENT` to the process group identified by `pid`. +/// +/// Console attachment is per-process state, so concurrent deliveries are serialized. +/// The broker temporarily attaches to the child's console; the event only targets the +/// child's process group (the broker is not part of it), so the broker is unaffected. +fn send_ctrl_break(pid: u32) -> anyhow::Result<()> { + use windows::Win32::System::Console::{AttachConsole, CTRL_BREAK_EVENT, FreeConsole, GenerateConsoleCtrlEvent}; + + static CONSOLE_LOCK: Mutex<()> = Mutex::new(()); + + let _guard = CONSOLE_LOCK.lock().expect("console lock poisoned"); + + // SAFETY: FFI call with no outstanding preconditions; detaching from a console we + // are not attached to simply fails and is ignored. + unsafe { + let _ = FreeConsole(); + } + + // SAFETY: FFI call with no outstanding preconditions; `pid` identifies the target + // process whose console we attach to, failure is reported as an error. + unsafe { AttachConsole(pid) }.context("AttachConsole failed")?; + + // SAFETY: FFI call with no outstanding preconditions; the calling process is + // attached to the target console and `pid` is a process group ID because the child + // was created with `CREATE_NEW_PROCESS_GROUP`. + let result = unsafe { GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid) }.context("GenerateConsoleCtrlEvent failed"); + + // SAFETY: FFI call with no outstanding preconditions; detach from the child's console. + unsafe { + let _ = FreeConsole(); + } + + result } fn is_trusted_winget_path(candidate: &Path, env: &std::collections::HashMap) -> bool { diff --git a/devolutions-agent/src/broker/operation_tracker.rs b/devolutions-agent/src/broker/operation_tracker.rs index 38eac2949..3dac38b5b 100644 --- a/devolutions-agent/src/broker/operation_tracker.rs +++ b/devolutions-agent/src/broker/operation_tracker.rs @@ -35,12 +35,12 @@ pub struct TrackedOperation { pub exit_code: Option, /// Human-readable note. pub note: Option, - /// Captured combined stdout+stderr (tail-truncated), when the request opted in. - pub stdout: Option, /// Authenticated operation owner. pub owner_key: String, /// When this entry should be evicted (set upon completion/failure). pub expires_at: Option>, + /// Fired when cancellation is requested, observed by the executor. + pub cancel_token: CancellationToken, } /// Thread-safe operation tracker. @@ -120,9 +120,9 @@ impl OperationTracker { completed_at: None, exit_code: None, note: None, - stdout: None, owner_key: owner_key.to_owned(), expires_at: None, + cancel_token: CancellationToken::new(), }, ); Ok((operation_id, true)) @@ -137,16 +137,9 @@ impl OperationTracker { } } - /// Mark an operation as finished with an exit code, a status note (success message or - /// short error summary), and optionally captured output. - pub fn mark_completed( - &self, - request_id: &str, - exit_code: i32, - note: String, - stdout: Option, - started_at: Option>, - ) { + /// Mark an operation as finished with an exit code and a status note (success message or + /// short error summary). + pub fn mark_completed(&self, request_id: &str, exit_code: i32, note: String, started_at: Option>) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { let now = Utc::now(); @@ -163,25 +156,56 @@ impl OperationTracker { }; op.exit_code = Some(exit_code); op.note = Some(note); - op.stdout = stdout; op.completed_at = Some(now); op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); } } /// Mark an operation as Failed without a process exit code (launch failure or timeout). - pub fn mark_failed(&self, request_id: &str, note: String, stdout: Option) { + pub fn mark_failed(&self, request_id: &str, note: String) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { let now = Utc::now(); op.status = OperationStatus::Failed; op.note = Some(note); - op.stdout = stdout; op.completed_at = Some(now); op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); } } + /// Mark an operation as Canceled (the executor terminated the process on request). + pub fn mark_canceled(&self, request_id: &str, note: String) { + let mut state = self.state.lock().expect("tracker lock poisoned"); + if let Some(op) = state.operations.get_mut(request_id) { + let now = Utc::now(); + op.status = OperationStatus::Canceled; + op.note = Some(note); + op.completed_at = Some(now); + op.expires_at = Some(now + chrono::Duration::from_std(RESULT_RETENTION).expect("valid duration")); + } + } + + /// Request cancellation of an operation owned by the authenticated client. + /// + /// Idempotent and asynchronous: a non-terminal operation is moved to `Canceling` and its + /// cancel token is fired; the executor terminates the process and the operation later + /// transitions to a terminal status. An already-terminal operation is returned unchanged. + /// Returns `None` when the operation is unknown or not owned by the client. + pub fn request_cancel(&self, request_id: &str, owner_key: &str) -> Option { + let mut state = self.state.lock().expect("tracker lock poisoned"); + let op = state + .operations + .get_mut(request_id) + .filter(|operation| operation.owner_key == owner_key)?; + + if !op.status.is_terminal() { + op.status = OperationStatus::Canceling; + op.cancel_token.cancel(); + } + + Some(op.clone()) + } + /// Query the current state of an operation. pub fn get(&self, request_id: &str) -> Option { let state = self.state.lock().expect("tracker lock poisoned"); diff --git a/devolutions-agent/src/broker/server/execution.rs b/devolutions-agent/src/broker/server/execution.rs index 4d3b8fd1f..ff141be7b 100644 --- a/devolutions-agent/src/broker/server/execution.rs +++ b/devolutions-agent/src/broker/server/execution.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use now_policy_api::ResourceId; use tracing::{error, info}; -use crate::broker::executor::{CommandExecutor, ExecutionContext, ProcessStartedCallback}; +use crate::broker::executor::{CommandExecutor, ExecutionContext, ProcessStartedCallback, is_canceled_error}; use crate::broker::operation_tracker::OperationTracker; pub(super) fn spawn_execution( @@ -25,7 +25,6 @@ pub(super) fn spawn_execution( }); match executor.execute(&context, Some(process_started)).await { Ok(output) => { - let stdout = (!output.stdout.is_empty()).then_some(output.stdout); let note = if output.exit_code == 0 { "process exited successfully".to_owned() } else { @@ -44,12 +43,19 @@ pub(super) fn spawn_execution( exit_code = output.exit_code, "Background execution completed" ); - tracker.mark_completed(&operation_id_string, output.exit_code, note, stdout, output.started_at); + tracker.mark_completed(&operation_id_string, output.exit_code, note, output.started_at); + } + Err(error) if is_canceled_error(&error) => { + info!(operation_id = %operation_id_string, "Background execution canceled"); + tracker.mark_canceled( + &operation_id_string, + "operation was canceled at the client's request".to_owned(), + ); } Err(error) => { let note = format!("{error:#}"); error!(operation_id = %operation_id_string, %error, "Background execution failed"); - tracker.mark_failed(&operation_id_string, note, None); + tracker.mark_failed(&operation_id_string, note); } } }); diff --git a/devolutions-agent/src/broker/server/mod.rs b/devolutions-agent/src/broker/server/mod.rs index abf5b45de..ee058559e 100644 --- a/devolutions-agent/src/broker/server/mod.rs +++ b/devolutions-agent/src/broker/server/mod.rs @@ -5,14 +5,14 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; use async_trait::async_trait; -use base64::Engine as _; use chrono::{DateTime, Utc}; use now_policy::PolicyDocument; use now_policy_api::{ - Base64Utf8Data, CapabilitiesResponse, CapabilitiesResponseKind, Decision, DecisionInfo, Elevation, ErrorCode, - ErrorResponse, EvaluationResponse, EvaluationResponseKind, ExecutionResponse, ExecutionResponseKind, - HealthResponse, HealthResponseKind, HealthStatus, ManagerCapability, ManagerName, OperationStatus, - OperationSubmission, PackageRequest, Scope, StatusRequest, StatusResponse, StatusResponseKind, Transport, + CancelRequest, CancelResponse, CancelResponseKind, CapabilitiesResponse, CapabilitiesResponseKind, Decision, + DecisionInfo, Elevation, ErrorCode, ErrorResponse, EvaluationResponse, EvaluationResponseKind, ExecutionResponse, + ExecutionResponseKind, HealthResponse, HealthResponseKind, HealthStatus, ManagerCapability, ManagerName, + OperationStatus, OperationSubmission, PackageRequest, Scope, StatusRequest, StatusResponse, StatusResponseKind, + Transport, }; use now_policy_server_template::{MAX_REQUEST_BODY_BYTES, PackageBrokerServer, SharedPackageBrokerServer}; use tracing::{info, trace, warn}; @@ -148,6 +148,18 @@ impl PackageBrokerServer for BrokerConnection { let owner_key = request.client.owner_key(); self.state.status_for_client(request, owner_key).await } + + async fn cancel(&self, request: CancelRequest) -> Result { + self.client + .validate_cancel_request(&request, self.state.skip_signature_validation) + .map_err(|error| { + warn!(error = format!("{error:#}"), "Rejected package broker cancel request"); + error_response(ErrorCode::Unauthorized, "pipe client authentication failed") + })?; + + let owner_key = request.client.owner_key(); + self.state.cancel_for_client(request, owner_key).await + } } impl BrokerState { @@ -225,7 +237,7 @@ impl BrokerState { let operation = if evaluated.would_execute { let generated_operation_id = new_operation_id()?; let submitted_at = Utc::now(); - let context = ExecutionContext { + let mut context = ExecutionContext { kill_processes: request .options .kill_before_operation @@ -240,6 +252,7 @@ impl BrokerState { elevation: request.client.requested_elevation, scope: request.options.scope, capture_output: request.capture_output, + cancel_token: tokio_util::sync::CancellationToken::new(), }; let owner_key = request.client_owner_key(); @@ -248,6 +261,11 @@ impl BrokerState { .register(&owner_key, &request, generated_operation_id) .map_err(|error| error_response(ErrorCode::Conflict, format!("{error:#}")))?; if is_new_operation { + // The executor observes the tracked operation's cancel token so a later + // cancel request can terminate the spawned process. + if let Some(tracked) = self.tracker.get(&operation_id) { + context.cancel_token = tracked.cancel_token; + } execution::spawn_execution( Arc::clone(&self.executor), self.tracker.clone(), @@ -255,6 +273,7 @@ impl BrokerState { context, ); } + // Query the status after spawning so fast executions are reflected in the response. let status = self .tracker .get(&operation_id) @@ -264,6 +283,8 @@ impl BrokerState { operation_id, status, submitted_at, + // The per-operation event channel is not implemented yet. + event_channel: None, }) } else { None @@ -310,9 +331,33 @@ impl BrokerState { exit_code: operation.exit_code, message: operation.note, details: None, - stdout: operation - .stdout - .map(|stdout| Base64Utf8Data(base64::engine::general_purpose::STANDARD.encode(stdout))), + }) + } + + async fn cancel_for_client( + &self, + request: CancelRequest, + owner_key: String, + ) -> Result { + let Some(operation) = self.tracker.request_cancel(&request.operation_id, &owner_key) else { + return Err(error_response(ErrorCode::NotFound, "operation not found")); + }; + + let message = if operation.status == OperationStatus::Canceling { + info!(operation_id = %request.operation_id, "Cancellation requested for operation"); + "cancellation requested; poll the status endpoint until a terminal status is reached".to_owned() + } else { + format!("operation already reached terminal status {:?}", operation.status) + }; + + Ok(CancelResponse { + response_kind: CancelResponseKind, + response_version: api_version(), + server: server_context(), + operation_id: request.operation_id, + request_id: operation.request_id, + status: operation.status, + message: Some(message), }) } @@ -451,7 +496,7 @@ mod tests { use now_policy_api as api; use super::*; - use crate::broker::executor::{ExecutionOutput, ProcessStartedCallback}; + use crate::broker::executor::{ExecutionOutput, OperationCanceled, ProcessStartedCallback}; struct NoopExecutor; @@ -695,4 +740,219 @@ mod tests { assert_eq!(executor.probe_count.load(Ordering::SeqCst), 2); } + + // ─── Cancellation ──────────────────────────────────────────────────────── + + /// Executor that blocks until the operation's cancel token fires, then reports cancellation. + struct CancelableExecutor; + + #[async_trait] + impl CommandExecutor for CancelableExecutor { + async fn execute( + &self, + ctx: &ExecutionContext, + _process_started: Option, + ) -> anyhow::Result { + ctx.cancel_token.cancelled().await; + Err(anyhow::Error::new(OperationCanceled)) + } + } + + /// Executor that never finishes, even when canceled (keeps operations in Canceling). + struct StuckExecutor; + + #[async_trait] + impl CommandExecutor for StuckExecutor { + async fn execute( + &self, + _ctx: &ExecutionContext, + _process_started: Option, + ) -> anyhow::Result { + std::future::pending().await + } + } + + /// Executor that completes instantly with exit code 0. + struct InstantExecutor; + + #[async_trait] + impl CommandExecutor for InstantExecutor { + async fn execute( + &self, + _ctx: &ExecutionContext, + _process_started: Option, + ) -> anyhow::Result { + Ok(ExecutionOutput { + exit_code: 0, + stdout: String::new(), + started_at: Some(Utc::now()), + }) + } + } + + fn state_with_executor(executor: Arc) -> BrokerState { + BrokerState { + policy: RwLock::new(Some(Arc::new(permissive_policy()))), + executor, + pipe_name: "test-pipe".to_owned(), + tracker: OperationTracker::new(), + skip_signature_validation: true, + manager_probe_cache: Default::default(), + } + } + + fn cancel_request(operation_id: &api::ResourceId, client: &api::ClientContext) -> CancelRequest { + CancelRequest { + request_kind: api::CancelRequestKind, + request_version: api::API_VERSION_STR.into(), + operation_id: operation_id.clone(), + client: client.clone(), + } + } + + fn status_request(operation_id: &api::ResourceId, client: &api::ClientContext) -> StatusRequest { + StatusRequest { + request_kind: api::StatusRequestKind, + request_version: api::API_VERSION_STR.into(), + operation_id: operation_id.clone(), + client: client.clone(), + } + } + + async fn submit_operation(state: &BrokerState, request: &PackageRequest) -> OperationSubmission { + let response = state + .execute(request.clone(), &test_sid()) + .await + .expect("execute request accepted"); + response.operation.expect("operation submitted") + } + + async fn wait_for_status( + state: &BrokerState, + request: &PackageRequest, + operation_id: &api::ResourceId, + ) -> StatusResponse { + for _ in 0..100 { + let status = state + .status_for_client( + status_request(operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("status query succeeds"); + if status.status.is_terminal() { + return status; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("operation did not reach a terminal status in time"); + } + + #[tokio::test] + async fn cancel_of_running_operation_reports_canceling_then_canceled() { + let state = state_with_executor(Arc::new(CancelableExecutor)); + let request = request(); + let operation = submit_operation(&state, &request).await; + + let response = state + .cancel_for_client( + cancel_request(&operation.operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("cancel accepted"); + + assert_eq!(response.status, OperationStatus::Canceling); + assert_eq!(response.request_id, request.request_id); + + let status = wait_for_status(&state, &request, &operation.operation_id).await; + assert_eq!(status.status, OperationStatus::Canceled); + } + + #[tokio::test] + async fn cancel_is_idempotent_while_canceling() { + let state = state_with_executor(Arc::new(StuckExecutor)); + let request = request(); + let operation = submit_operation(&state, &request).await; + + for _ in 0..2 { + let response = state + .cancel_for_client( + cancel_request(&operation.operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("cancel accepted"); + assert_eq!(response.status, OperationStatus::Canceling); + } + + let status = state + .status_for_client( + status_request(&operation.operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("status query succeeds"); + assert_eq!(status.status, OperationStatus::Canceling); + } + + #[tokio::test] + async fn cancel_of_completed_operation_returns_terminal_status() { + let state = state_with_executor(Arc::new(InstantExecutor)); + let request = request(); + let operation = submit_operation(&state, &request).await; + + let terminal = wait_for_status(&state, &request, &operation.operation_id).await; + assert_eq!(terminal.status, OperationStatus::Completed); + + let response = state + .cancel_for_client( + cancel_request(&operation.operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("cancel of terminal operation is idempotent"); + assert_eq!(response.status, OperationStatus::Completed); + + // The operation stays Completed after the cancel attempt. + let status = state + .status_for_client( + status_request(&operation.operation_id, &request.client), + request.client_owner_key(), + ) + .await + .expect("status query succeeds"); + assert_eq!(status.status, OperationStatus::Completed); + } + + #[tokio::test] + async fn cancel_of_unknown_operation_is_not_found() { + let state = state_with_executor(Arc::new(StuckExecutor)); + let request = request(); + + let error = state + .cancel_for_client( + cancel_request(&api::ResourceId::from("does-not-exist"), &request.client), + request.client_owner_key(), + ) + .await + .expect_err("unknown operation is rejected"); + assert_eq!(error.code, ErrorCode::NotFound); + } + + #[tokio::test] + async fn cancel_by_different_owner_is_not_found() { + let state = state_with_executor(Arc::new(StuckExecutor)); + let request = request(); + let operation = submit_operation(&state, &request).await; + + let error = state + .cancel_for_client( + cancel_request(&operation.operation_id, &request.client), + "someone|else".to_owned(), + ) + .await + .expect_err("foreign operation is rejected"); + assert_eq!(error.code, ErrorCode::NotFound); + } }