From b307dad511df290268afb23d5483f4d87b4b6ea3 Mon Sep 17 00:00:00 2001 From: Vladyslav Nikonov Date: Tue, 4 Aug 2026 14:57:37 +0300 Subject: [PATCH] feat(agent): support canceling package broker operations Adds a POST /v1/package-operations/cancel endpoint to the package broker. Canceling a running operation terminates the spawned package-manager process and reports the operation as Canceled; cancelation is idempotent and terminal operations report their final status. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 20 +- Cargo.toml | 4 + devolutions-agent/Cargo.toml | 4 +- devolutions-agent/src/broker/auth.rs | 11 +- devolutions-agent/src/broker/executor/mod.rs | 15 ++ .../src/broker/executor/windows/mod.rs | 38 +++- .../src/broker/executor/windows/process.rs | 74 ++++--- .../src/broker/operation_tracker.rs | 36 ++++ .../src/broker/server/execution.rs | 16 +- devolutions-agent/src/broker/server/mod.rs | 180 ++++++++++++++++-- 10 files changed, 333 insertions(+), 65 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2b250c01f..b34a1c6f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1524,7 +1524,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 2.0.118", + "syn 1.0.109", ] [[package]] @@ -2593,8 +2593,8 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.2.1", - "windows-result 0.4.1", + "windows-link 0.1.3", + "windows-result 0.3.4", ] [[package]] @@ -3120,7 +3120,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.62.2", + "windows-core 0.61.2", ] [[package]] @@ -4715,8 +4715,6 @@ dependencies = [ [[package]] name = "now-policy" version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cc04b547209a37ecd3993eb47773ec6fa38f8cac178479079f650555bbedbbb" dependencies = [ "chrono", "schemars", @@ -4730,9 +4728,7 @@ dependencies = [ [[package]] name = "now-policy-api" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca52de7d57fca9ea49cc10f229bdc31cfbdb37f7d49a6e7407fa60248d1df7c6" +version = "0.3.0" dependencies = [ "base64 0.22.1", "chrono", @@ -4748,9 +4744,7 @@ dependencies = [ [[package]] name = "now-policy-server-template" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ce816e494a5034becfe105f30611284431275c24bdad3a8e7ef71a7b31e6e1" +version = "0.3.0" dependencies = [ "aide", "async-trait", @@ -7400,7 +7394,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index e5eb35315..444adb57a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,10 @@ lto = true [patch.crates-io] tracing-appender = { git = "https://github.com/CBenoit/tracing.git", rev = "42097daf92e683cf18da7639ddccb056721a796c" } +# TODO: remove these path patches once now-policy 0.2 and now-policy-api / now-policy-server-template 0.3 are published to crates.io. +now-policy = { path = "D:/now-proto/.worktrees/broker-cancel/policies/rust/now-policy" } +now-policy-api = { path = "D:/now-proto/.worktrees/broker-cancel/policies/rust/now-policy-api" } +now-policy-server-template = { path = "D:/now-proto/.worktrees/broker-cancel/policies/rust/now-policy-server-template" } [workspace.lints.rust] # Declare the custom cfgs. diff --git a/devolutions-agent/Cargo.toml b/devolutions-agent/Cargo.toml index c57167482..3a19c0061 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" 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..c506ed1a0 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; @@ -53,8 +54,22 @@ pub struct ExecutionContext { pub scope: Option, /// When true, capture the main command's combined stdout+stderr. pub capture_output: bool, + /// Cancelation signal for the operation; the executor terminates the running process when triggered. + pub cancel_token: CancellationToken, } +/// Marker error returned when execution is canceled by the client. +#[derive(Debug)] +pub struct ExecutionCanceled; + +impl std::fmt::Display for ExecutionCanceled { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("operation canceled") + } +} + +impl std::error::Error for ExecutionCanceled {} + pub type ProcessStartedCallback = std::sync::Arc) + Send + Sync>; /// All package managers the broker knows how to drive. diff --git a/devolutions-agent/src/broker/executor/windows/mod.rs b/devolutions-agent/src/broker/executor/windows/mod.rs index abaac8120..08c8ffcc4 100644 --- a/devolutions-agent/src/broker/executor/windows/mod.rs +++ b/devolutions-agent/src/broker/executor/windows/mod.rs @@ -10,6 +10,7 @@ use anyhow::{Context as _, bail}; use async_trait::async_trait; use devolutions_agent_shared::temp_file::{BATCH_UTF8_PREAMBLE, POWERSHELL_UTF8_ENCODING_PREAMBLE, TmpFileGuard}; use now_policy_api::{Elevation, ManagerName, Scope}; +use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; use win_api_wrappers::identity::sid::Sid; use win_api_wrappers::process::Process; @@ -18,7 +19,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, ExecutionCanceled, ExecutionContext, ExecutionOutput, + ProcessStartedCallback, +}; use crate::broker::policy_security; mod privileges; @@ -329,6 +333,10 @@ fn run_plan( process_started: Option, ) -> anyhow::Result { let requires_elevation = ctx.elevation == Elevation::Elevated || ctx.scope == Some(Scope::Machine); + if ctx.cancel_token.is_cancelled() { + return Err(anyhow::Error::new(ExecutionCanceled)).context("operation canceled by client request"); + } + if requires_elevation && command_is_bun(&ctx.command) { bail!("elevated Bun package operations are not supported by the broker"); } @@ -348,6 +356,7 @@ fn run_plan( } // 1. Kill requested processes (best-effort; a missing process is not an error). + let cleanup_cancel_token = CancellationToken::new(); for process_name in &ctx.kill_processes { let kill_cmd = vec![ trusted_system32_executable("taskkill.exe"), @@ -355,7 +364,15 @@ 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, + &cleanup_cancel_token, + None, + ) { Ok(out) => info!(%process_name, exit_code = out.exit_code, "Kill-before-operation completed"), Err(error) => warn!(%process_name, %error, "Kill-before-operation failed (ignored)"), } @@ -371,6 +388,7 @@ fn run_plan( session_id, ctx.capture_output, requires_elevation, + &ctx.cancel_token, None, ) .context("failed to run pre-operation command")?; @@ -391,6 +409,7 @@ fn run_plan( session_id, ctx.capture_output, requires_elevation, + &ctx.cancel_token, process_started, )?; @@ -399,7 +418,15 @@ fn run_plan( 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(command) => match create_process( + token, + command.args(), + session_id, + false, + requires_elevation, + &cleanup_cancel_token, + 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"), @@ -1317,7 +1344,7 @@ mod tests { use windows::Win32::Security::{NO_INHERITANCE, WinWorldSid}; use super::{ - POWERSHELL_UTF8_ENCODING_PREAMBLE, WindowsExecutor, execute_as_current_user, + CancellationToken, POWERSHELL_UTF8_ENCODING_PREAMBLE, WindowsExecutor, execute_as_current_user, prepare_chocolatey_script_in_with_default_install_root, prepare_main_command_in, prepare_shell_command_in, reject_unsupported_vcpkg_elevation, resolve_trusted_chocolatey_executable, resolve_winget_executable, }; @@ -1918,6 +1945,7 @@ mod tests { elevation: Elevation::Elevated, scope: Some(Scope::User), capture_output: false, + cancel_token: CancellationToken::new(), }; let error = reject_unsupported_vcpkg_elevation(&ctx).expect_err("elevated vcpkg should fail"); @@ -1941,6 +1969,7 @@ mod tests { elevation: Elevation::Elevated, scope: Some(Scope::User), capture_output: false, + cancel_token: CancellationToken::new(), }; let executor = WindowsExecutor { is_system: true }; @@ -1968,6 +1997,7 @@ mod tests { elevation: Elevation::Standard, scope: Some(Scope::User), capture_output: false, + cancel_token: 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..52046f6a5 100644 --- a/devolutions-agent/src/broker/executor/windows/process.rs +++ b/devolutions-agent/src/broker/executor/windows/process.rs @@ -2,9 +2,11 @@ use std::io::Read as _; use std::path::{Path, PathBuf}; +use std::time::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; @@ -16,10 +18,14 @@ use windows::Win32::System::Threading::{ }; use windows::Win32::UI::WindowsAndMessaging::SW_HIDE; -use crate::broker::executor::{ExecutionOutput, MAX_CAPTURED_OUTPUT_BYTES, ProcessStartedCallback, tail_utf8}; +use crate::broker::executor::{ + ExecutionCanceled, ExecutionOutput, MAX_CAPTURED_OUTPUT_BYTES, ProcessStartedCallback, tail_utf8, +}; use crate::broker::operation_tracker::OperationTracker; use crate::broker::policy_security; +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. @@ -36,6 +42,7 @@ pub(super) fn create_process( session_id: u32, capture: bool, requires_elevation: bool, + cancel_token: &CancellationToken, process_started: Option, ) -> anyhow::Result { let cmd_line = CommandLine::new(command.to_vec()); @@ -166,28 +173,51 @@ 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(); + loop { + 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; + } + + if cancel_token.is_cancelled() { + warn!( + session_id, + pid = process_info.process_id, + "Process canceled; terminating" + ); + process_info + .process + .terminate(1) + .context("failed to terminate canceled process")?; + let _ = process_info.process.wait(None); + if let Some(handle) = reader { + let _ = handle.join(); + } + return Err(anyhow::Error::new(ExecutionCanceled)).context("operation canceled by client request"); + } + + if Instant::now() >= deadline { + warn!( + session_id, + pid = process_info.process_id, + timeout_ms = operation_timeout_ms(), + "Process timed out; terminating" + ); + 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() + ); + } } let exit_code = process_info diff --git a/devolutions-agent/src/broker/operation_tracker.rs b/devolutions-agent/src/broker/operation_tracker.rs index 38eac2949..f506eee09 100644 --- a/devolutions-agent/src/broker/operation_tracker.rs +++ b/devolutions-agent/src/broker/operation_tracker.rs @@ -39,6 +39,8 @@ pub struct TrackedOperation { pub stdout: Option, /// Authenticated operation owner. pub owner_key: String, + /// Cancelation signal for the execution task. + pub cancel_token: CancellationToken, /// When this entry should be evicted (set upon completion/failure). pub expires_at: Option>, } @@ -122,6 +124,7 @@ impl OperationTracker { note: None, stdout: None, owner_key: owner_key.to_owned(), + cancel_token: CancellationToken::new(), expires_at: None, }, ); @@ -132,6 +135,9 @@ impl OperationTracker { pub fn mark_running(&self, request_id: &str, started_at: DateTime) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { + if op.status == OperationStatus::Canceling || op.status == OperationStatus::Canceled { + return; + } op.status = OperationStatus::Running; op.started_at = Some(started_at); } @@ -149,6 +155,9 @@ impl OperationTracker { ) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { + if op.status == OperationStatus::Canceled { + return; + } let now = Utc::now(); if op.started_at.is_none() && let Some(started_at) = started_at @@ -173,6 +182,9 @@ impl OperationTracker { pub fn mark_failed(&self, request_id: &str, note: String, stdout: Option) { let mut state = self.state.lock().expect("tracker lock poisoned"); if let Some(op) = state.operations.get_mut(request_id) { + if op.status == OperationStatus::Canceled { + return; + } let now = Utc::now(); op.status = OperationStatus::Failed; op.note = Some(note); @@ -182,6 +194,30 @@ impl OperationTracker { } } + /// Request cancelation of an operation. + pub fn request_cancel(&self, operation_id: &str) -> Option { + let mut state = self.state.lock().expect("tracker lock poisoned"); + let op = state.operations.get_mut(operation_id)?; + if !op.status.is_terminal() { + op.status = OperationStatus::Canceling; + op.cancel_token.cancel(); + } + Some(op.clone()) + } + + /// Mark an operation as Canceled. + pub fn mark_canceled(&self, operation_id: &str, note: String, stdout: Option) { + let mut state = self.state.lock().expect("tracker lock poisoned"); + if let Some(op) = state.operations.get_mut(operation_id) { + let now = Utc::now(); + op.status = OperationStatus::Canceled; + 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")); + } + } + /// 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..e5a5ed965 100644 --- a/devolutions-agent/src/broker/server/execution.rs +++ b/devolutions-agent/src/broker/server/execution.rs @@ -48,8 +48,20 @@ pub(super) fn spawn_execution( } Err(error) => { let note = format!("{error:#}"); - error!(operation_id = %operation_id_string, %error, "Background execution failed"); - tracker.mark_failed(&operation_id_string, note, None); + if error + .downcast_ref::() + .is_some() + { + info!(operation_id = %operation_id_string, "Background execution canceled"); + tracker.mark_canceled( + &operation_id_string, + "operation canceled by client request".to_owned(), + None, + ); + } else { + error!(operation_id = %operation_id_string, %error, "Background execution failed"); + tracker.mark_failed(&operation_id_string, note, None); + } } } }); diff --git a/devolutions-agent/src/broker/server/mod.rs b/devolutions-agent/src/broker/server/mod.rs index abf5b45de..9947a470f 100644 --- a/devolutions-agent/src/broker/server/mod.rs +++ b/devolutions-agent/src/broker/server/mod.rs @@ -9,10 +9,11 @@ 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, + Base64Utf8Data, 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 +149,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,29 +238,34 @@ impl BrokerState { let operation = if evaluated.would_execute { let generated_operation_id = new_operation_id()?; let submitted_at = Utc::now(); - let context = ExecutionContext { - kill_processes: request - .options - .kill_before_operation - .iter() - .map(|process| process.0.clone()) - .collect(), - pre_command: request.options.pre_operation_command.clone(), - command: evaluated.command.clone(), - post_command: request.options.post_operation_command.clone(), - effective_user: request.client.effective_user.clone(), - user_sid: user_sid.clone(), - elevation: request.client.requested_elevation, - scope: request.options.scope, - capture_output: request.capture_output, - }; - let owner_key = request.client_owner_key(); let (operation_id, is_new_operation) = self .tracker .register(&owner_key, &request, generated_operation_id) .map_err(|error| error_response(ErrorCode::Conflict, format!("{error:#}")))?; if is_new_operation { + let cancel_token = self + .tracker + .get(&operation_id) + .map(|operation| operation.cancel_token) + .unwrap_or_default(); + let context = ExecutionContext { + kill_processes: request + .options + .kill_before_operation + .iter() + .map(|process| process.0.clone()) + .collect(), + pre_command: request.options.pre_operation_command.clone(), + command: evaluated.command.clone(), + post_command: request.options.post_operation_command.clone(), + effective_user: request.client.effective_user.clone(), + user_sid: user_sid.clone(), + elevation: request.client.requested_elevation, + scope: request.options.scope, + capture_output: request.capture_output, + cancel_token, + }; execution::spawn_execution( Arc::clone(&self.executor), self.tracker.clone(), @@ -316,6 +334,42 @@ impl BrokerState { }) } + async fn cancel_for_client( + &self, + request: CancelRequest, + owner_key: String, + ) -> Result { + if !owner_key.is_empty() && self.tracker.get_for_owner(&request.operation_id, &owner_key).is_none() { + return Err(error_response(ErrorCode::NotFound, "operation not found")); + } + + let Some(operation) = self.tracker.request_cancel(&request.operation_id) else { + return Err(error_response(ErrorCode::NotFound, "operation not found")); + }; + + info!( + operation_id = %request.operation_id, + status = ?operation.status, + "Package operation cancelation requested" + ); + + let message = if operation.status == OperationStatus::Canceling { + operation.note.or_else(|| Some("cancelation requested".to_owned())) + } else { + operation.note + }; + + 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, + }) + } + fn evaluate_request(&self, request: &PackageRequest) -> Result { // SECURITY: Pre/post operation commands are raw command strings executed via // cmd.exe with the execution token, and the policy schema cannot restrict @@ -567,6 +621,15 @@ mod tests { } } + fn cancel_request(operation_id: api::ResourceId) -> CancelRequest { + CancelRequest { + request_kind: api::CancelRequestKind, + request_version: api::API_VERSION_STR.into(), + operation_id, + client: request().client, + } + } + #[test] fn elevated_pre_operation_command_is_rejected_even_under_permissive_policy() { let mut request = request(); @@ -695,4 +758,79 @@ mod tests { assert_eq!(executor.probe_count.load(Ordering::SeqCst), 2); } + + #[tokio::test] + async fn cancel_unknown_operation_returns_not_found() { + let Err(error) = state() + .cancel_for_client( + cancel_request(api::ResourceId::from("missing-operation")), + String::new(), + ) + .await + else { + panic!("expected unknown cancel target to be rejected"); + }; + + assert_eq!(error.code, ErrorCode::NotFound); + } + + #[tokio::test] + async fn cancel_running_operation_sets_canceling_and_triggers_token() { + let state = state(); + let package_request = request(); + let operation_id = api::ResourceId::from("operation-1"); + state + .tracker + .register("", &package_request, operation_id.clone()) + .unwrap(); + + let response = state + .cancel_for_client(cancel_request(operation_id.clone()), String::new()) + .await + .unwrap(); + + assert_eq!(response.status, OperationStatus::Canceling); + assert!(state.tracker.get(&operation_id).unwrap().cancel_token.is_cancelled()); + } + + #[tokio::test] + async fn cancel_terminal_operation_returns_terminal_status() { + let state = state(); + let package_request = request(); + let operation_id = api::ResourceId::from("operation-1"); + state + .tracker + .register("", &package_request, operation_id.clone()) + .unwrap(); + state + .tracker + .mark_completed(&operation_id, 0, "process exited successfully".to_owned(), None, None); + + let response = state + .cancel_for_client(cancel_request(operation_id), String::new()) + .await + .unwrap(); + + assert_eq!(response.status, OperationStatus::Completed); + } + + #[tokio::test] + async fn cancel_owner_mismatch_returns_not_found() { + let state = state(); + let package_request = request(); + let operation_id = api::ResourceId::from("operation-1"); + state + .tracker + .register("a|x", &package_request, operation_id.clone()) + .unwrap(); + + let Err(error) = state + .cancel_for_client(cancel_request(operation_id), "b|y".to_owned()) + .await + else { + panic!("expected owner mismatch to be rejected"); + }; + + assert_eq!(error.code, ErrorCode::NotFound); + } }