From d27cb60c26b30ee033e0435d7d446f94151eec7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Mon, 21 Sep 2026 10:30:00 +0000 Subject: [PATCH 1/4] FM-603: Proxmox template, clone, and snapshot operations behind the destructive review gate --- Cargo.lock | 1 + crates/fleet-api/Cargo.toml | 1 + crates/fleet-api/src/apply.rs | 1 + crates/fleet-api/src/frogenv.rs | 1 + crates/fleet-api/src/lib.rs | 5 + crates/fleet-api/src/mise.rs | 1 + crates/fleet-api/src/onboarding.rs | 1 + crates/fleet-api/src/operations.rs | 1 + crates/fleet-api/src/projects.rs | 1 + crates/fleet-api/src/proxmox.rs | 282 +++++++++ crates/fleet-api/src/ready.rs | 1 + crates/fleet-api/src/skills.rs | 1 + crates/fleet-application/src/authz.rs | 12 +- crates/fleet-application/src/operation.rs | 39 +- crates/fleet-auth/tests/authz_adapter.rs | 2 +- crates/fleet-controller/src/apply.rs | 2 + crates/fleet-controller/src/main.rs | 7 + crates/fleet-controller/src/proxmox_exec.rs | 573 +++++++++++++++++- crates/fleet-controller/src/ready.rs | 1 + .../tests/agentless_inventory.rs | 1 + crates/fleet-controller/tests/apply.rs | 4 + crates/fleet-controller/tests/checkout.rs | 1 + crates/fleet-controller/tests/frogenv.rs | 1 + crates/fleet-controller/tests/gateway.rs | 1 + crates/fleet-controller/tests/machines.rs | 1 + crates/fleet-controller/tests/mise.rs | 1 + .../tests/operation_worker.rs | 6 + crates/fleet-controller/tests/proxmox.rs | 133 ++++ crates/fleet-controller/tests/ready.rs | 1 + crates/fleet-controller/tests/skills.rs | 1 + crates/fleet-controller/tests/ssh_exec.rs | 4 + .../tests/worker_lifecycle.rs | 1 + crates/fleetctl/src/lib.rs | 170 ++++++ crates/fleetctl/tests/cli.rs | 39 ++ crates/fleetd/tests/node_install.rs | 1 + .../fleet-provider-proxmox/src/lib.rs | 457 ++++++++++++-- packages/api-client/openapi.json | 297 +++++++++ packages/api-client/src/generated/fleet.ts | 231 +++++++ 38 files changed, 2240 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index edcfdc3..9a28387 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -567,6 +567,7 @@ dependencies = [ "http-body-util", "serde", "serde_json", + "sha2 0.10.9", "tokio", "tower", "utoipa", diff --git a/crates/fleet-api/Cargo.toml b/crates/fleet-api/Cargo.toml index 0669713..b40b32a 100644 --- a/crates/fleet-api/Cargo.toml +++ b/crates/fleet-api/Cargo.toml @@ -12,6 +12,7 @@ axum = "0.8.9" fleet-application = { version = "0.1.0", path = "../fleet-application" } fleet-core = { path = "../fleet-core" } futures-util = "0.3.31" +sha2 = "0.10" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["time", "macros", "rt", "net"] } diff --git a/crates/fleet-api/src/apply.rs b/crates/fleet-api/src/apply.rs index 5ca2ee2..f4ff108 100644 --- a/crates/fleet-api/src/apply.rs +++ b/crates/fleet-api/src/apply.rs @@ -287,6 +287,7 @@ pub async fn start_apply_workflow( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/frogenv.rs b/crates/fleet-api/src/frogenv.rs index a767485..68594d3 100644 --- a/crates/fleet-api/src/frogenv.rs +++ b/crates/fleet-api/src/frogenv.rs @@ -250,6 +250,7 @@ pub async fn start_frogenv_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/lib.rs b/crates/fleet-api/src/lib.rs index 831a80a..a12f95e 100644 --- a/crates/fleet-api/src/lib.rs +++ b/crates/fleet-api/src/lib.rs @@ -130,6 +130,9 @@ pub const API_BASE_PATH: &str = "/api/v1"; proxmox::AssociationCandidateDto, proxmox::ObserveProxmoxGuestRequest, proxmox::StartProxmoxLifecycleRequest, + proxmox::ProxmoxReviewDto, + proxmox::ReviewProxmoxOperationRequest, + proxmox::StartReviewedProxmoxOperationRequest, proxmox::ProviderAgentDto, proxmox::ProviderInterfaceDto, node::CreateEnrollmentTokenRequest, @@ -236,6 +239,8 @@ pub fn api(state: Arc) -> (Router, utoipa::openapi::OpenAp .routes(routes!(proxmox::list_proxmox_guests)) .routes(routes!(proxmox::observe_proxmox_guest)) .routes(routes!(proxmox::start_proxmox_lifecycle)) + .routes(routes!(proxmox::review_proxmox_operation)) + .routes(routes!(proxmox::start_reviewed_proxmox_operation)) .with_state(state), ) .split_for_parts(); diff --git a/crates/fleet-api/src/mise.rs b/crates/fleet-api/src/mise.rs index 810a45d..d6a26ef 100644 --- a/crates/fleet-api/src/mise.rs +++ b/crates/fleet-api/src/mise.rs @@ -286,6 +286,7 @@ pub async fn start_mise_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/onboarding.rs b/crates/fleet-api/src/onboarding.rs index 5481b60..21a3f74 100644 --- a/crates/fleet-api/src/onboarding.rs +++ b/crates/fleet-api/src/onboarding.rs @@ -702,6 +702,7 @@ async fn start_onboarding_operation( deadline_at: Some(fleet_core::SystemClock::now_unix_millis() + deadline_ms), correlation_id: Some(correlation_id.to_string()), payload_json: Some(serde_json::json!({ "draftId": draft_id }).to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/operations.rs b/crates/fleet-api/src/operations.rs index 87797b3..0a6969a 100644 --- a/crates/fleet-api/src/operations.rs +++ b/crates/fleet-api/src/operations.rs @@ -395,6 +395,7 @@ pub async fn create_operation( deadline_at: request.deadline_at, correlation_id: Some(correlation_id.to_string()), payload_json: request.payload_json.clone(), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/projects.rs b/crates/fleet-api/src/projects.rs index 694a2e7..a08b911 100644 --- a/crates/fleet-api/src/projects.rs +++ b/crates/fleet-api/src/projects.rs @@ -634,6 +634,7 @@ pub async fn start_discovery( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/proxmox.rs b/crates/fleet-api/src/proxmox.rs index 83faa37..16c0db2 100644 --- a/crates/fleet-api/src/proxmox.rs +++ b/crates/fleet-api/src/proxmox.rs @@ -1088,6 +1088,288 @@ pub async fn start_proxmox_lifecycle( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: true, + }, + ) + .await + .map_err(|error| crate::operations::map_use_case_error(&error, correlation_id))?; + Ok(( + StatusCode::ACCEPTED, + Json(Resource::new(crate::operations::OperationDto::from( + operation, + ))), + )) +} + +/// The review record: exactly what the destructive operation will run, +/// bound to a token the create call must present. +#[derive(Clone, Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ProxmoxReviewDto { + /// The token binding this review to the request it described. + pub review_token: String, + /// The action under review. + pub action: String, + /// The guest's hosting node. + pub node: String, + /// The guest's VMID. + pub vmid: u32, + /// The account the operation will run under. + pub account_id: String, + /// The action-specific parameters, as reviewed. + pub params: serde_json::Value, +} + +/// The destructive-review request. +#[derive(Debug, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ReviewProxmoxOperationRequest { + /// The guest's hosting node. + pub node: String, + /// The action-specific parameters (snapshot name/description, clone + /// target id/name, and so on). + #[serde(default)] + pub params: serde_json::Value, +} + +/// The token binding one reviewed request to its create call: the SHA-256 +/// of the canonical reviewed material. Stateless — the controller holds no +/// review store; a create presents the token computed from exactly the +/// payload it carries, so what runs is what was reviewed. +#[must_use] +fn review_token( + account_id: &str, + vmid: u32, + action: &str, + request: &ReviewProxmoxOperationRequest, +) -> String { + use sha2::Digest as _; + let canonical = serde_json::json!({ + "accountId": account_id, + "vmid": vmid, + "action": action, + "node": request.node, + "params": request.params, + }); + let mut hasher = sha2::Sha256::new(); + hasher.update(serde_json::to_string(&canonical).unwrap_or_default()); + let digest: [u8; 32] = hasher.finalize().into(); + digest.iter().fold(String::with_capacity(64), |mut out, b| { + use std::fmt::Write as _; + let _ = write!(out, "{b:02x}"); + out + }) +} + +/// Reviews one destructive-adjacent operation: renders exactly what will +/// run and returns the token the create call must present. +/// +/// # Errors +/// +/// Returns the public error envelope on refusal or a malformed request. +#[utoipa::path( + post, + path = "/proxmox/accounts/{accountId}/guests/{vmid}/{action}/review", + tag = "proxmox", + operation_id = "reviewProxmoxOperation", + params( + ("accountId" = String, Path, description = "The account's identity."), + ("vmid" = u32, Path, description = "The guest's VMID."), + ("action" = String, Path, description = "The destructive action: snapshot, snapshot-revert, snapshot-delete, clone, template, or task-cancel.") + ), + request_body = ReviewProxmoxOperationRequest, + responses( + ( + status = 200, + description = "The review record; present its token to the create call.", + body = Resource + ), + ( + status = 400, + description = "The action is unrecognized or the request is malformed.", + body = crate::error::ApiError + ), + ( + status = 403, + description = "The caller may not operate Proxmox guests destructively.", + body = crate::error::ApiError + ), + ) +)] +pub async fn review_proxmox_operation( + State(state): State>, + principal: Option>, + Extension(correlation_id): Extension, + Path((account_id, vmid, action)): Path<(String, u32, String)>, + Json(request): Json, +) -> Result>, ApiErrorResponse> { + let principal = crate::operations::principal_or_error(principal, correlation_id)?; + if !matches!( + action.as_str(), + "snapshot" | "snapshot-revert" | "snapshot-delete" | "clone" | "template" | "task-cancel" + ) { + return Err(crate::machines::invalid_request( + &format!("unrecognized destructive action {action:?}"), + correlation_id, + )); + } + if request.node.is_empty() || request.node.len() > 128 { + return Err(crate::machines::invalid_request( + "the node must be 1..=128 characters", + correlation_id, + )); + } + if let Err(decision) = fleet_application::authz::authorize( + state.authorizer.as_ref(), + fleet_application::authz::AccessRequest { + principal_id: &principal.id, + action: fleet_application::authz::Permission::ProxmoxDestructive, + resource: None, + }, + ) { + return Err(crate::machines::denied_error(decision, correlation_id)); + } + Ok(Json(Resource::new(ProxmoxReviewDto { + review_token: review_token(&account_id, vmid, &action, &request), + action, + node: request.node, + vmid, + account_id, + params: request.params, + }))) +} + +/// The create call for a reviewed destructive operation: carries the +/// review token binding it to what was reviewed. +#[derive(Debug, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct StartReviewedProxmoxOperationRequest { + /// The guest's hosting node. + pub node: String, + /// The review token from the review call. + pub review_token: String, + /// The action-specific parameters, identical to the reviewed ones. + #[serde(default)] + pub params: serde_json::Value, + /// The deadline, in seconds. Bounded by the executor. + pub timeout_seconds: u64, +} + +/// Runs a reviewed destructive-adjacent operation as a durable operation. +/// The review token must match the request exactly: what runs is what was +/// reviewed. The generic operations surface refuses these kinds outright. +/// +/// # Errors +/// +/// Returns the public error envelope on refusal, a missing/stale review +/// token, or a malformed request. +#[utoipa::path( + post, + path = "/proxmox/accounts/{accountId}/guests/{vmid}/{action}/run", + tag = "proxmox", + operation_id = "startReviewedProxmoxOperation", + params( + ("accountId" = String, Path, description = "The account's identity."), + ("vmid" = u32, Path, description = "The guest's VMID."), + ("action" = String, Path, description = "The reviewed destructive action.") + ), + request_body = StartReviewedProxmoxOperationRequest, + responses( + ( + status = 202, + description = "The operation was accepted and is durable.", + body = Resource + ), + ( + status = 400, + description = "The action is unrecognized or the review token does not match the request.", + body = crate::error::ApiError + ), + ( + status = 403, + description = "The caller may not operate Proxmox guests destructively.", + body = crate::error::ApiError + ), + ) +)] +pub async fn start_reviewed_proxmox_operation( + State(state): State>, + principal: Option>, + Extension(correlation_id): Extension, + headers: axum::http::HeaderMap, + Path((account_id, vmid, action)): Path<(String, u32, String)>, + Json(request): Json, +) -> Result<(StatusCode, Json>), ApiErrorResponse> { + let principal = crate::operations::principal_or_error(principal, correlation_id)?; + if !matches!( + action.as_str(), + "snapshot" | "snapshot-revert" | "snapshot-delete" | "clone" | "template" | "task-cancel" + ) { + return Err(crate::machines::invalid_request( + &format!("unrecognized destructive action {action:?}"), + correlation_id, + )); + } + if let Err(decision) = fleet_application::authz::authorize( + state.authorizer.as_ref(), + fleet_application::authz::AccessRequest { + principal_id: &principal.id, + action: fleet_application::authz::Permission::ProxmoxDestructive, + resource: None, + }, + ) { + return Err(crate::machines::denied_error(decision, correlation_id)); + } + // The review gate: the token is recomputed from THIS request, so the + // create runs only what was reviewed — any parameter change invalidates + // the token. + let presented = ReviewProxmoxOperationRequest { + node: request.node.clone(), + params: request.params.clone(), + }; + let expected = review_token(&account_id, vmid, &action, &presented); + if expected != request.review_token { + return Err(crate::machines::invalid_request( + "the review token does not match this request; review again", + correlation_id, + )); + } + if request.node.is_empty() || request.node.len() > 128 { + return Err(crate::machines::invalid_request( + "the node must be 1..=128 characters", + correlation_id, + )); + } + let mut payload = serde_json::json!({ + "accountId": account_id, + "node": request.node, + "vmid": vmid, + "timeoutSeconds": request.timeout_seconds, + "params": request.params, + }); + let _ = &mut payload; + let kind = format!("proxmox.guest.{action}"); + let kind = if action == "task-cancel" { + "proxmox.task-cancel".to_owned() + } else { + kind + }; + let idempotency_key = headers + .get(crate::IDEMPOTENCY_KEY_HEADER) + .and_then(|value| value.to_str().ok()) + .map(|key| format!("{}:{key}", principal.id)); + let operation = state + .operations + .create( + state.authorizer.as_ref(), + &principal.id, + &fleet_application::operation::NewOperation { + kind, + idempotency_key, + deadline_at: None, + correlation_id: Some(correlation_id.to_string()), + payload_json: Some(payload.to_string()), + reviewed: true, }, ) .await diff --git a/crates/fleet-api/src/ready.rs b/crates/fleet-api/src/ready.rs index 50c83a5..525a087 100644 --- a/crates/fleet-api/src/ready.rs +++ b/crates/fleet-api/src/ready.rs @@ -269,6 +269,7 @@ pub async fn start_ready_workflow( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-api/src/skills.rs b/crates/fleet-api/src/skills.rs index 8189777..53f9a2b 100644 --- a/crates/fleet-api/src/skills.rs +++ b/crates/fleet-api/src/skills.rs @@ -205,6 +205,7 @@ pub async fn start_skills_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-application/src/authz.rs b/crates/fleet-application/src/authz.rs index 8651b8e..dcc41ba 100644 --- a/crates/fleet-application/src/authz.rs +++ b/crates/fleet-application/src/authz.rs @@ -139,6 +139,10 @@ pub enum Permission { /// Run a lifecycle action (start/stop/shutdown/reboot) on a Proxmox /// guest. A mutation: it changes the guest's power state. ProxmoxOperate, + /// Run a destructive-adjacent Proxmox operation: snapshot, revert, + /// snapshot delete, clone, template conversion, or remote task + /// cancellation. A mutation: it changes or destroys guest state. + ProxmoxDestructive, } impl Permission { @@ -185,6 +189,7 @@ impl Permission { Permission::ProxmoxRead, Permission::ProxmoxConfig, Permission::ProxmoxOperate, + Permission::ProxmoxDestructive, ]; /// The stable action id, as recorded in decisions and audit events. @@ -230,6 +235,7 @@ impl Permission { Permission::ProxmoxRead => "proxmox.read", Permission::ProxmoxConfig => "proxmox.config", Permission::ProxmoxOperate => "proxmox.operate", + Permission::ProxmoxDestructive => "proxmox.destructive", } } @@ -277,7 +283,8 @@ impl Permission { | Permission::SourceActivate | Permission::ProxmoxRead | Permission::ProxmoxConfig - | Permission::ProxmoxOperate => true, + | Permission::ProxmoxOperate + | Permission::ProxmoxDestructive => true, } } @@ -303,7 +310,8 @@ impl Permission { | Permission::SourceActivate | Permission::ProxmoxRead | Permission::ProxmoxConfig - | Permission::ProxmoxOperate => false, + | Permission::ProxmoxOperate + | Permission::ProxmoxDestructive => false, Permission::MachineReadSensitive | Permission::OperationCancel | Permission::SecretRead diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index edf105a..f1cae66 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -43,7 +43,7 @@ use crate::authz::{AccessRequest, Authorizer, Decision, Permission, ReasonId, au /// machine-scoped shape plus the plan and its approval identities /// (FM-402); the source kinds carry the remote/commit payloads and are /// catalog-level (FM-403). -pub const CREATABLE_KINDS: [&str; 35] = [ +pub const CREATABLE_KINDS: [&str; 41] = [ "noop", "ssh.exec", "agentless.inventory", @@ -79,6 +79,12 @@ pub const CREATABLE_KINDS: [&str; 35] = [ "proxmox.guest.stop", "proxmox.guest.shutdown", "proxmox.guest.reboot", + "proxmox.guest.snapshot", + "proxmox.guest.snapshot-revert", + "proxmox.guest.snapshot-delete", + "proxmox.guest.clone", + "proxmox.guest.template", + "proxmox.task-cancel", ]; /// The machine-scoped permission a kind's creation requires, when any. @@ -103,6 +109,12 @@ fn catalog_scoped_kind_permission(kind: &str) -> Option { | "proxmox.guest.stop" | "proxmox.guest.shutdown" | "proxmox.guest.reboot" => Some(Permission::ProxmoxOperate), + "proxmox.guest.snapshot" + | "proxmox.guest.snapshot-revert" + | "proxmox.guest.snapshot-delete" + | "proxmox.guest.clone" + | "proxmox.guest.template" + | "proxmox.task-cancel" => Some(Permission::ProxmoxDestructive), _ => None, } } @@ -438,6 +450,11 @@ pub struct NewOperation { pub correlation_id: Option, /// The bounded provider input, for kinds that need one. pub payload_json: Option, + /// Whether the request arrived through a dedicated endpoint that + /// already enforced the kind's extra gate (e.g. the destructive + /// review). Only the dedicated surface sets it; the generic surface + /// leaves it off, which refuses destructive kinds outright. + pub reviewed: bool, } /// The authorized operation use cases. @@ -460,6 +477,7 @@ impl Operations { /// # Errors /// /// Fails on denial, unknown kind, or a backend failure. + #[allow(clippy::too_many_lines)] pub async fn create( &self, authorizer: &dyn Authorizer, @@ -525,6 +543,25 @@ impl Operations { ) .map_err(OperationUseCaseError::Denied)?; } else if let Some(permission) = catalog_scoped_kind_permission(&new.kind) { + // The destructive-adjacent Proxmox kinds never route through + // the generic surface: their creation goes through the + // dedicated reviewed endpoint, which binds the operation to a + // confirmed review token. A generic-surface create is a + // route-around attempt, refused as malformed. + if !new.reviewed + && (new.kind.starts_with("proxmox.guest.snapshot") + || matches!( + new.kind.as_str(), + "proxmox.guest.clone" | "proxmox.guest.template" | "proxmox.task-cancel" + )) + { + return Err(OperationUseCaseError::Invalid { + detail: format!( + "the {kind} kind is destructive-adjacent and runs only through its reviewed dedicated endpoint", + kind = new.kind + ), + }); + } authorize( authorizer, AccessRequest { diff --git a/crates/fleet-auth/tests/authz_adapter.rs b/crates/fleet-auth/tests/authz_adapter.rs index 0b6bedb..4d972c5 100644 --- a/crates/fleet-auth/tests/authz_adapter.rs +++ b/crates/fleet-auth/tests/authz_adapter.rs @@ -103,7 +103,7 @@ fn every_catalog_action_has_a_unique_stable_id_and_a_risk_ruling() { assert!(Permission::MachineReadSensitive.is_risky()); assert!(!Permission::SystemRead.is_risky()); // The catalog is the complete vocabulary the adapter permits. - assert_eq!(Permission::ALL.len(), 39); + assert_eq!(Permission::ALL.len(), 40); } #[test] diff --git a/crates/fleet-controller/src/apply.rs b/crates/fleet-controller/src/apply.rs index 1e9bfdb..29b765f 100644 --- a/crates/fleet-controller/src/apply.rs +++ b/crates/fleet-controller/src/apply.rs @@ -551,6 +551,7 @@ impl ApplyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_owned()), + reviewed: false, }, ) .await @@ -639,6 +640,7 @@ impl ApplyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/src/main.rs b/crates/fleet-controller/src/main.rs index 7f16db4..89f4279 100644 --- a/crates/fleet-controller/src/main.rs +++ b/crates/fleet-controller/src/main.rs @@ -313,6 +313,13 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { with_source.clone(), std::sync::Arc::new( fleet_controller::proxmox_exec::ProxmoxLifecycleExecutor::new( + accounts.clone(), + credentials.clone(), + proxmox_client.clone(), + ), + ), + std::sync::Arc::new( + fleet_controller::proxmox_exec::ProxmoxDestructiveExecutor::new( accounts, credentials, proxmox_client, diff --git a/crates/fleet-controller/src/proxmox_exec.rs b/crates/fleet-controller/src/proxmox_exec.rs index f3199ee..c2a1da0 100644 --- a/crates/fleet-controller/src/proxmox_exec.rs +++ b/crates/fleet-controller/src/proxmox_exec.rs @@ -21,7 +21,7 @@ use std::time::Duration; use fleet_application::operation::{Operation, Operations}; use fleet_application::proxmox::ProxmoxCredentialStore; use fleet_application::worker::OperationExecutor; -use fleet_provider_proxmox::{LifecycleAction, ProxmoxSource as _, TaskStatus}; +use fleet_provider_proxmox::{LifecycleAction, ProxmoxSource as _, TaskStatus, Upid}; use serde::Deserialize; /// The interval between task-status polls. Fixed, not clock-derived. @@ -317,6 +317,24 @@ struct RunParams { sleep: Arc futures_util::future::BoxFuture<'static, ()> + Send + Sync>, } +/// The payload the destructive kinds carry: the lifecycle fields plus the +/// reviewed action parameters. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DestructivePayload { + /// The Proxmox account running the action. + pub account_id: String, + /// The guest's hosting node. + pub node: String, + /// The guest's VMID. + pub vmid: u32, + /// The deadline, in seconds. Bounded by the executor. + pub timeout_seconds: u64, + /// The reviewed action parameters. + #[serde(default)] + pub params: serde_json::Value, +} + /// Decodes and validates an operation's payload. fn payload(operation: &Operation) -> Result { serde_json::from_str( @@ -328,13 +346,14 @@ fn payload(operation: &Operation) -> Result, lifecycle: Arc, + destructive: Arc, } impl ProxmoxDispatch { @@ -343,10 +362,12 @@ impl ProxmoxDispatch { pub fn new( fallback: Arc, lifecycle: Arc, + destructive: Arc, ) -> Self { Self { fallback, lifecycle, + destructive, } } } @@ -354,10 +375,552 @@ impl ProxmoxDispatch { #[async_trait::async_trait] impl OperationExecutor for ProxmoxDispatch { async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { - if operation.kind.starts_with("proxmox.guest.") { + if DESTRUCTIVE_KINDS.contains(&operation.kind.as_str()) { + self.destructive.execute(operations, operation).await + } else if operation.kind.starts_with("proxmox.guest.") { self.lifecycle.execute(operations, operation).await } else { self.fallback.execute(operations, operation).await } } } + +/// The destructive-adjacent kinds; they route to their own executor. +pub const DESTRUCTIVE_KINDS: [&str; 6] = [ + "proxmox.guest.snapshot", + "proxmox.guest.snapshot-revert", + "proxmox.guest.snapshot-delete", + "proxmox.guest.clone", + "proxmox.guest.template", + "proxmox.task-cancel", +]; + +/// The destructive-adjacent executor: snapshot, revert, snapshot-delete, +/// clone, template conversion, and remote task cancellation. Every kind +/// arrived through the reviewed dedicated endpoint (the generic surface +/// refuses them); the executor re-applies the trust gate and classifies +/// idempotency before touching anything. +#[derive(Debug)] +pub struct ProxmoxDestructiveExecutor { + accounts: Arc, + credentials: Arc, + client: fleet_provider_proxmox::ProxmoxClient, +} + +impl ProxmoxDestructiveExecutor { + /// Composes the executor from its parts. + #[must_use] + pub fn new( + accounts: Arc, + credentials: Arc, + client: fleet_provider_proxmox::ProxmoxClient, + ) -> Self { + Self { + accounts, + credentials, + client, + } + } +} + +#[async_trait::async_trait] +impl OperationExecutor for ProxmoxDestructiveExecutor { + #[allow(clippy::too_many_lines)] + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + let payload: DestructivePayload = payload(operation)?; + let (account, secret) = self.bound(&payload.account_id).await?; + let request = self.request(&account, &secret); + let deadline = Duration::from_secs(payload.timeout_seconds.min(MAX_LIFECYCLE_TIMEOUT)); + let kind = operation.kind.as_str(); + let outcome: Result<(), String> = match kind { + "proxmox.guest.snapshot" => { + let name = payload + .params + .get("snapshot") + .and_then(serde_json::Value::as_str) + .ok_or("the reviewed parameters carry no snapshot name")?; + validate_snapshot_name(name)?; + let description = payload + .params + .get("description") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + let include_ram = payload + .params + .get("includeRam") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + // Idempotency classification: an existing snapshot with the + // same name succeeds without an operation; a differing + // description is a conflict, not a silent overwrite. + let existing = self + .client + .guest_snapshots(request.clone(), &payload.node, payload.vmid) + .await + .map_err(|error| format!("the snapshot listing failed: {error}"))?; + if let Some(snapshot) = existing.iter().find(|snapshot| snapshot.name == name) { + if snapshot.description == description { + return self + .finish_noop( + operations, + &operation.id, + format!( + "snapshot {name} already exists with the reviewed description" + ), + ) + .await; + } + return complete_failure( + operations, + &operation.id, + "conflict", + &format!( + "a snapshot named {name} already exists with a different description; pick another name" + ), + ) + .await; + } + self.run_to_terminal( + operations, + &operation.id, + &request, + &payload.node, + payload.vmid, + deadline, + { + let node = payload.node.clone(); + let vmid = payload.vmid; + let name = name.to_owned(); + let description = description.to_owned(); + move |client, request| { + Box::pin(async move { + client + .guest_snapshot( + request, + &node, + vmid, + &name, + &description, + include_ram, + ) + .await + }) + as futures_util::future::BoxFuture<'static, _> + } + }, + ) + .await + } + "proxmox.guest.snapshot-revert" => { + let name = payload + .params + .get("snapshot") + .and_then(serde_json::Value::as_str) + .ok_or("the reviewed parameters carry no snapshot name")?; + validate_snapshot_name(name)?; + self.run_to_terminal( + operations, + &operation.id, + &request, + &payload.node, + payload.vmid, + deadline, + { + let node = payload.node.clone(); + let vmid = payload.vmid; + let name = name.to_owned(); + move |client, request| { + Box::pin(async move { + client + .guest_snapshot_rollback(request, &node, vmid, &name) + .await + }) + as futures_util::future::BoxFuture<'static, _> + } + }, + ) + .await + } + "proxmox.guest.snapshot-delete" => { + let name = payload + .params + .get("snapshot") + .and_then(serde_json::Value::as_str) + .ok_or("the reviewed parameters carry no snapshot name")?; + validate_snapshot_name(name)?; + self.run_to_terminal( + operations, + &operation.id, + &request, + &payload.node, + payload.vmid, + deadline, + { + let node = payload.node.clone(); + let vmid = payload.vmid; + let name = name.to_owned(); + move |client, request| { + Box::pin(async move { + client + .guest_snapshot_delete(request, &node, vmid, &name) + .await + .map(|()| None) + }) + as futures_util::future::BoxFuture<'static, _> + } + }, + ) + .await + } + "proxmox.guest.clone" => { + let new_id = payload + .params + .get("newId") + .and_then(serde_json::Value::as_u64) + .ok_or("the reviewed parameters carry no new VMID")?; + let new_id = u32::try_from(new_id) + .map_err(|_| "the new VMID exceeds the u32 bound".to_owned())?; + let name = payload + .params + .get("name") + .and_then(serde_json::Value::as_str) + .ok_or("the reviewed parameters carry no target name")?; + let full_copy = payload + .params + .get("fullCopy") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + // Idempotency classification: a target VMID that already + // exists is a conflict (never a duplicate); the cluster + // resources are the truth. + let resources = self + .client + .list_qemu_resources(request.clone()) + .await + .map_err(|error| format!("the resource listing failed: {error}"))?; + if resources + .iter() + .any(|resource| resource.vmid == Some(new_id)) + { + return complete_failure( + operations, + &operation.id, + "conflict", + &format!("a guest with VMID {new_id} already exists; pick another target"), + ) + .await; + } + self.run_to_terminal( + operations, + &operation.id, + &request, + &payload.node, + payload.vmid, + deadline, + { + let node = payload.node.clone(); + let vmid = payload.vmid; + let name = name.to_owned(); + move |client, request| { + Box::pin(async move { + client + .guest_clone(request, &node, vmid, new_id, &name, full_copy) + .await + .map(Some) + }) + as futures_util::future::BoxFuture<'static, _> + } + }, + ) + .await + } + "proxmox.guest.template" => { + // Idempotency classification: converting a template + // succeeds without an operation. + let resources = self + .client + .list_qemu_resources(request.clone()) + .await + .map_err(|error| format!("the resource listing failed: {error}"))?; + let resource = resources + .iter() + .find(|resource| resource.vmid == Some(payload.vmid)) + .ok_or_else(|| format!("guest qemu/{} not found", payload.vmid))?; + if resource.kind == "qemu-template" { + return self + .finish_noop(operations, &operation.id, "already a template".to_owned()) + .await; + } + self.run_to_terminal( + operations, + &operation.id, + &request, + &payload.node, + payload.vmid, + deadline, + { + let node = payload.node.clone(); + let vmid = payload.vmid; + move |client, request| { + Box::pin(async move { + client.guest_convert_template(request, &node, vmid).await + }) + as futures_util::future::BoxFuture<'static, _> + } + }, + ) + .await + } + "proxmox.task-cancel" => { + let raw = payload + .params + .get("upid") + .and_then(serde_json::Value::as_str) + .ok_or("the reviewed parameters carry no UPID")?; + let upid = Upid::parse(raw)?; + self.client + .stop_task(request.clone(), &upid) + .await + .map_err(|error| format!("the task cancellation failed: {error}"))?; + // The outcome is read back honestly: the task may have + // finished between the stop and the status read. + let status = self + .client + .task_status(request.clone(), &upid) + .await + .unwrap_or(TaskStatus::Unknown); + self.finish(operations, &operation.id, status).await + } + other => return Err(format!("not a Proxmox destructive kind: {other}")), + }; + outcome?; + Ok(()) + } +} + +impl ProxmoxDestructiveExecutor { + /// The trusted account and its resolved secret: the same explicit-trust + /// gate the other surfaces apply. + async fn bound( + &self, + account_id: &str, + ) -> Result<(fleet_application::proxmox::ProxmoxAccount, String), String> { + let account = self + .accounts + .get(account_id) + .await + .map_err(|detail| format!("the account is unreadable: {detail}"))?; + let Some(pinned) = account.fingerprint.clone() else { + return Err(format!( + "the account {} has no confirmed fingerprint; confirm the host's trust first", + account.name + )); + }; + let secret = self + .credentials + .load(account_id) + .await + .map_err(|error| format!("the credential store failed: {error}"))? + .ok_or_else(|| { + format!( + "the API token for account {} is not in the secret store", + account.name + ) + })?; + Ok(( + fleet_application::proxmox::ProxmoxAccount { + fingerprint: Some(pinned), + ..account + }, + secret, + )) + } + + /// The request every provider call in this executor carries. + fn request( + &self, + account: &fleet_application::proxmox::ProxmoxAccount, + secret: &str, + ) -> fleet_provider_proxmox::PveHttpRequest { + fleet_provider_proxmox::PveHttpRequest { + host: account.host.clone(), + port: account.port, + path: "/api2/json/version".to_owned(), + pinned_fingerprint: account.fingerprint.clone(), + credentials: Arc::new(fleet_provider_proxmox::PveCredentials { + token_id: account.token_id.clone(), + token: fleet_core::SensitiveString::new(secret.to_owned()), + }), + method: fleet_provider_proxmox::PveHttpMethod::Get, + } + } + + /// Completes the operation as a no-op success with the reason as the + /// result: an idempotent hit is a success, not an error. + async fn finish_noop( + &self, + operations: &Operations, + operation_id: &str, + note: String, + ) -> Result<(), String> { + operations + .complete( + operation_id, + "succeeded", + Some(&serde_json::json!({ "noop": note }).to_string()), + None, + ) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + + /// Completes the operation from the terminal task status. + async fn finish( + &self, + operations: &Operations, + operation_id: &str, + status: TaskStatus, + ) -> Result<(), String> { + match status { + TaskStatus::Ok => operations + .complete( + operation_id, + "succeeded", + Some(&serde_json::json!({ "taskState": "ok" }).to_string()), + None, + ) + .await + .map(|_| ()) + .map_err(|error| error.to_string()), + TaskStatus::Running => { + Err("the executor returned while the task still runs".to_owned()) + } + TaskStatus::Error { detail } => { + complete_failure(operations, operation_id, "task_failed", &detail).await + } + TaskStatus::Unknown => { + complete_failure( + operations, + operation_id, + "task_unknown", + "the task's status could not be read; its outcome is unknown, not assumed", + ) + .await + } + } + } + + /// Runs one mutating call and polls its task to a terminal state; a + /// synchronous outcome completes immediately. Compensation is by + /// record: the failure detail names what ran so the operator can + /// reconcile, and nothing is deleted on failure. + #[allow(clippy::too_many_arguments)] + async fn run_to_terminal( + &self, + operations: &Operations, + operation_id: &str, + request: &fleet_provider_proxmox::PveHttpRequest, + node: &str, + vmid: u32, + deadline: Duration, + call: impl FnOnce( + fleet_provider_proxmox::ProxmoxClient, + fleet_provider_proxmox::PveHttpRequest, + ) -> futures_util::future::BoxFuture< + 'static, + Result, fleet_provider_proxmox::PveApiError>, + >, + ) -> Result<(), String> { + operations + .record_progress( + operation_id, + Some(0), + Some(1), + Some(&format!("{node}/qemu/{vmid}")), + ) + .await + .map_err(|error| error.to_string())?; + let upid = call(self.client.clone(), request.clone()) + .await + .map_err(|error| format!("the operation failed: {error}"))?; + let Some(upid) = upid else { + // Synchronous outcome: read the guest state as the + // verification, not an assumption. + return self.finish(operations, operation_id, TaskStatus::Ok).await; + }; + let started = std::time::Instant::now(); + loop { + let cancelled = operations + .cancel_requested(operation_id) + .await + .unwrap_or(false); + if cancelled { + return complete_failure( + operations, + operation_id, + "cancelled", + &format!( + "cancelled while waiting; the remote task on node {} keeps running and its outcome is unknown", + upid.node + ), + ) + .await; + } + let status = self + .client + .task_status(request.clone(), &upid) + .await + .map_err(|error| { + format!("the task status failed: {error}; the task's outcome is unknown") + })?; + match status { + TaskStatus::Running => {} + terminal => return self.finish(operations, operation_id, terminal).await, + } + if started.elapsed() >= deadline { + return complete_failure( + operations, + operation_id, + "deadline_expired", + &format!( + "the deadline expired while the task still runs; its final state is unknown (task on node {})", + upid.node + ), + ) + .await; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + } +} + +/// Snapshot names are PVE identifiers: bounded, no path material. +fn validate_snapshot_name(name: &str) -> Result<(), String> { + if name.is_empty() + || name.len() > 64 + || !name + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') + { + return Err(format!( + "the snapshot name must be 1..=64 characters of [a-zA-Z0-9_-], not {name:?}" + )); + } + Ok(()) +} + +/// Completes an operation as a failure with a redacted detail. +async fn complete_failure( + operations: &Operations, + operation_id: &str, + reason: &str, + detail: &str, +) -> Result<(), String> { + let error_json = serde_json::json!({ "reason": reason, "detail": detail }).to_string(); + operations + .complete(operation_id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) +} diff --git a/crates/fleet-controller/src/ready.rs b/crates/fleet-controller/src/ready.rs index 3170b27..67a32d3 100644 --- a/crates/fleet-controller/src/ready.rs +++ b/crates/fleet-controller/src/ready.rs @@ -526,6 +526,7 @@ impl ReadyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_owned()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/agentless_inventory.rs b/crates/fleet-controller/tests/agentless_inventory.rs index 2a1cea8..c800911 100644 --- a/crates/fleet-controller/tests/agentless_inventory.rs +++ b/crates/fleet-controller/tests/agentless_inventory.rs @@ -194,6 +194,7 @@ async fn an_inventory_operation_probes_and_ingests() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/apply.rs b/crates/fleet-controller/tests/apply.rs index 45b3083..6aa7395 100644 --- a/crates/fleet-controller/tests/apply.rs +++ b/crates/fleet-controller/tests/apply.rs @@ -166,6 +166,7 @@ async fn an_unapproved_plan_completes_blocked_naming_the_steps() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await @@ -237,6 +238,7 @@ async fn a_kind_state_mismatched_payload_fails_honestly() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await @@ -309,6 +311,7 @@ async fn an_approved_plan_executes_every_action_and_succeeds() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await @@ -387,6 +390,7 @@ async fn a_failing_step_stops_with_compensations_and_remainder() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/checkout.rs b/crates/fleet-controller/tests/checkout.rs index 85b31ba..a7c0492 100644 --- a/crates/fleet-controller/tests/checkout.rs +++ b/crates/fleet-controller/tests/checkout.rs @@ -98,6 +98,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/frogenv.rs b/crates/fleet-controller/tests/frogenv.rs index d00b1ef..f2dc8fb 100644 --- a/crates/fleet-controller/tests/frogenv.rs +++ b/crates/fleet-controller/tests/frogenv.rs @@ -101,6 +101,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/gateway.rs b/crates/fleet-controller/tests/gateway.rs index 855dd11..2c7efd4 100644 --- a/crates/fleet-controller/tests/gateway.rs +++ b/crates/fleet-controller/tests/gateway.rs @@ -162,6 +162,7 @@ impl Harness { deadline_at, correlation_id: None, payload_json: Some(serde_json::json!({ "machineId": machine_id }).to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/machines.rs b/crates/fleet-controller/tests/machines.rs index ab790fa..7c58116 100644 --- a/crates/fleet-controller/tests/machines.rs +++ b/crates/fleet-controller/tests/machines.rs @@ -162,6 +162,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: Some(serde_json::json!({ "machineId": machine_id }).to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/mise.rs b/crates/fleet-controller/tests/mise.rs index 5304ebe..901d60f 100644 --- a/crates/fleet-controller/tests/mise.rs +++ b/crates/fleet-controller/tests/mise.rs @@ -100,6 +100,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/operation_worker.rs b/crates/fleet-controller/tests/operation_worker.rs index 396ac60..58092ab 100644 --- a/crates/fleet-controller/tests/operation_worker.rs +++ b/crates/fleet-controller/tests/operation_worker.rs @@ -30,6 +30,7 @@ async fn the_noop_operation_runs_end_to_end() { deadline_at: None, correlation_id: Some("corr-worker-1".to_owned()), payload_json: None, + reviewed: false, }, ) .await @@ -72,6 +73,7 @@ async fn two_workers_cannot_claim_the_same_operation() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await @@ -110,6 +112,7 @@ async fn a_crashed_workers_lease_is_recovered_as_failed_not_retried() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await @@ -160,6 +163,7 @@ async fn a_cancelled_operation_stops_without_running() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await @@ -212,6 +216,7 @@ async fn a_deadline_expires_even_when_no_worker_claims_it() { deadline_at: Some(now - 1_000), correlation_id: None, payload_json: None, + reviewed: false, }, ) .await @@ -254,6 +259,7 @@ async fn restart_preserves_terminal_truth() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/proxmox.rs b/crates/fleet-controller/tests/proxmox.rs index 592829b..d1b148e 100644 --- a/crates/fleet-controller/tests/proxmox.rs +++ b/crates/fleet-controller/tests/proxmox.rs @@ -576,6 +576,13 @@ const TASK_OK_BODY: &str = r#"{"data":{"status":"stopped","exitstatus":"OK"}}"#; const TASK_ERROR_BODY: &str = r#"{"data":{"status":"stopped","exitstatus":"ERROR: start failed: KVM is not available"}}"#; +const SNAPSHOT_CREATE_BODY: &str = + r#"{"data":"UPID:pve:0015523F:0C6DF532:6AAFE1EC:qmsnapshot:101:root@pam!GLM-AGENT:"}"#; + +const SNAPSHOT_LIST_BODY: &str = r#"{"data":[ + {"name":"current","description":"","vmstate":0} +]}"#; + /// What the lifecycle fixture answers per poll. #[derive(Debug, Clone, Copy)] enum TaskOutcome { @@ -604,6 +611,25 @@ impl PveTransport for LifecycleTransport { observed: FP.to_owned(), }); } + if request.pinned_fingerprint.is_none() { + return Err(PveTransportError::ObserveRefused { + observed: FP.to_owned(), + }); + } + if request.path.ends_with("/snapshot") + && request.method == fleet_provider_proxmox::PveHttpMethod::Post + { + return Ok(PveHttpResponse { + status: 200, + body: SNAPSHOT_CREATE_BODY.as_bytes().to_vec(), + }); + } + if request.path.ends_with("/snapshot") { + return Ok(PveHttpResponse { + status: 200, + body: SNAPSHOT_LIST_BODY.as_bytes().to_vec(), + }); + } if request.path.contains("/status/start") { return Ok(PveHttpResponse { status: 200, @@ -680,6 +706,13 @@ async fn lifecycle_harness(outcome: TaskOutcome) -> Harness { Arc::new(fleet_application::worker::NoopExecutor), Arc::new( fleet_controller::proxmox_exec::ProxmoxLifecycleExecutor::new( + accounts.clone(), + credentials.clone(), + fleet_provider_proxmox::ProxmoxClient::new(transport.clone()), + ), + ), + Arc::new( + fleet_controller::proxmox_exec::ProxmoxDestructiveExecutor::new( accounts, credentials, fleet_provider_proxmox::ProxmoxClient::new(transport), @@ -899,3 +932,103 @@ async fn an_unrecognized_action_refuses_with_invalid_request() { assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); assert_eq!(body["code"], "invalid_request"); } + +#[tokio::test] +async fn a_destructive_operation_requires_the_review_token_and_runs() { + let harness = lifecycle_harness(TaskOutcome::OkAfterOne).await; + let account_id = trusted_account(&harness).await; + + // The create without a review token is refused. + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/snapshot/run"), + json!({ + "node": "pve", + "reviewToken": "not-the-token", + "params": {"snapshot": "demo-snap", "description": "demo"}, + "timeoutSeconds": 30 + }), + ) + .await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["code"], "invalid_request"); + + // The review renders exactly what will run and returns the token. + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/snapshot/review"), + json!({ + "node": "pve", + "params": {"snapshot": "demo-snap", "description": "demo"} + }), + ) + .await; + assert_eq!(status, axum::http::StatusCode::OK, "{body}"); + assert_eq!(body["data"]["action"], "snapshot"); + assert_eq!(body["data"]["vmid"], 101); + let token = body["data"]["reviewToken"].as_str().unwrap().to_owned(); + + // The create with the token is accepted and runs to task OK. + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/snapshot/run"), + json!({ + "node": "pve", + "reviewToken": token, + "params": {"snapshot": "demo-snap", "description": "demo"}, + "timeoutSeconds": 30 + }), + ) + .await; + assert_eq!(status, axum::http::StatusCode::ACCEPTED, "{body}"); + let operation_id = body["data"]["id"].as_str().unwrap().to_owned(); + let mut terminal = String::new(); + for _ in 0..50 { + let (_, body) = harness + .get(&format!("/api/v1/operations/{operation_id}")) + .await; + terminal = body["data"]["state"] + .as_str() + .unwrap_or_default() + .to_owned(); + if terminal == "succeeded" || terminal == "failed" { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert_eq!(terminal, "succeeded", "{body}"); +} + +#[tokio::test] +async fn a_tampered_review_payload_is_refused() { + let harness = lifecycle_harness(TaskOutcome::OkAfterOne).await; + let account_id = trusted_account(&harness).await; + + // Review one payload... + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/snapshot/review"), + json!({ + "node": "pve", + "params": {"snapshot": "demo-snap", "description": "demo"} + }), + ) + .await; + assert_eq!(status, axum::http::StatusCode::OK, "{body}"); + let token = body["data"]["reviewToken"].as_str().unwrap().to_owned(); + + // ...then try to run a DIFFERENT one with the same token: refused. + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/snapshot/run"), + json!({ + "node": "pve", + "reviewToken": token, + "params": {"snapshot": "OTHER-snap", "description": "demo"}, + "timeoutSeconds": 30 + }), + ) + .await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["code"], "invalid_request"); +} diff --git a/crates/fleet-controller/tests/ready.rs b/crates/fleet-controller/tests/ready.rs index a4b2858..ae4f9e3 100644 --- a/crates/fleet-controller/tests/ready.rs +++ b/crates/fleet-controller/tests/ready.rs @@ -134,6 +134,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/skills.rs b/crates/fleet-controller/tests/skills.rs index 1db8e6c..2d9a06e 100644 --- a/crates/fleet-controller/tests/skills.rs +++ b/crates/fleet-controller/tests/skills.rs @@ -102,6 +102,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/ssh_exec.rs b/crates/fleet-controller/tests/ssh_exec.rs index 3f5174f..118d829 100644 --- a/crates/fleet-controller/tests/ssh_exec.rs +++ b/crates/fleet-controller/tests/ssh_exec.rs @@ -76,6 +76,7 @@ async fn an_unverified_endpoint_refuses_to_execute() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json_string), + reviewed: false, }, ) .await @@ -154,6 +155,7 @@ async fn a_verified_endpoint_runs_the_script_and_reports_output() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json_string), + reviewed: false, }, ) .await @@ -195,6 +197,7 @@ async fn an_unknown_kind_fails_honestly() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await @@ -216,6 +219,7 @@ async fn the_noop_kind_still_runs_through_the_composed_executor() { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await diff --git a/crates/fleet-controller/tests/worker_lifecycle.rs b/crates/fleet-controller/tests/worker_lifecycle.rs index c6635a7..7ef080c 100644 --- a/crates/fleet-controller/tests/worker_lifecycle.rs +++ b/crates/fleet-controller/tests/worker_lifecycle.rs @@ -99,6 +99,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: None, + reviewed: false, }, ) .await diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index 2531a94..c9c29a3 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -542,6 +542,24 @@ pub enum Command { /// The machine the guest is confirmed to be. machine_id: String, }, + /// Review then run a destructive-adjacent operation on a guest. + ProxmoxDestructive { + /// The action: snapshot, snapshot-revert, snapshot-delete, clone, + /// template, or task-cancel. + action: String, + /// The account's identity. + account_id: String, + /// The guest's hosting node. + node: String, + /// The guest's VMID. + vmid: u32, + /// The action-specific parameters, as JSON on standard input. + params: Option, + /// Wait for the operation to finish. + wait: bool, + /// How long to wait, in seconds. + timeout: Option, + }, /// Run a lifecycle action on a guest as a durable operation. ProxmoxLifecycle { /// The action: start, stop, shutdown, or reboot. @@ -1223,6 +1241,81 @@ fn parse_proxmox_command(verb: &str, rest: &[&str]) -> Result } _ => Err(CliError { message: usage() }), }, + "snapshot" | "snapshot-revert" | "snapshot-delete" | "clone" | "template" + | "task-cancel" => { + let mut account_id = None; + let mut node = None; + let mut vmid = None; + let mut wait = false; + let mut timeout = None; + let mut flags = rest.iter().copied(); + while let Some(flag) = flags.next() { + match flag { + "--account" => { + account_id = Some( + flags + .next() + .ok_or_else(|| CliError { + message: "--account requires a value".to_owned(), + })? + .to_owned(), + ); + } + "--node" => { + node = Some( + flags + .next() + .ok_or_else(|| CliError { + message: "--node requires a value".to_owned(), + })? + .to_owned(), + ); + } + "--vmid" => { + let value = flags.next().ok_or_else(|| CliError { + message: "--vmid requires a value".to_owned(), + })?; + vmid = Some(value.parse().map_err(|_| CliError { + message: format!("--vmid must be a number, not {value:?}"), + })?); + } + "--wait" => wait = true, + "--timeout" => { + let value = flags.next().ok_or_else(|| CliError { + message: "--timeout requires a value".to_owned(), + })?; + timeout = Some(value.parse().map_err(|_| CliError { + message: format!("--timeout must be a number, not {value:?}"), + })?); + } + other => { + return Err(CliError { + message: format!( + "unknown flag {other:?}; see the usage below\n\n{}", + usage() + ), + }); + } + } + } + // The action's parameters arrive as JSON on standard input at + // request time; the CLI never puts them in argv. + Ok(Command::ProxmoxDestructive { + action: verb.to_owned(), + account_id: account_id.ok_or_else(|| CliError { + message: "--account is required".to_owned(), + })?, + node: node.ok_or_else(|| CliError { + message: "--node is required".to_owned(), + })?, + vmid: vmid.ok_or_else(|| CliError { + message: "--vmid is required".to_owned(), + })?, + params: None, + wait, + timeout, + }) + } "start" | "stop" | "shutdown" | "reboot" => { let mut account_id = None; let mut node = None; @@ -1636,6 +1729,7 @@ pub fn run(invocation: &Invocation) -> Result { correlation_id, )?; let body = follow_wait_stage(&client, invocation, body)?; + let body = follow_review(&client, invocation, body)?; let body = follow_install_wait(&client, invocation, body)?; let body = follow_checkout_wait(&client, invocation, body)?; let payload = if body.get("items").is_some() { @@ -1671,6 +1765,52 @@ fn follow_install_wait( wait_for_operation(client, invocation, &operation_id, *poll_timeout) } +/// The destructive command's two-step flow: the review's answer carries +/// the token; the run request presents it with the identical payload, so +/// what runs is what was reviewed. +fn follow_review( + client: &reqwest::blocking::Client, + invocation: &Invocation, + body: Value, +) -> Result { + let Command::ProxmoxDestructive { + action, + account_id, + node, + vmid, + params, + .. + } = &invocation.command + else { + return Ok(body); + }; + let Some(token) = body["data"]["reviewToken"].as_str() else { + return Err(CliError { + message: "the review did not answer with a token".to_owned(), + }); + }; + let params_value: Value = params + .as_deref() + .and_then(|text| serde_json::from_str(text).ok()) + .unwrap_or(Value::Null); + let run_body = serde_json::json!({ + "node": node, + "reviewToken": token, + "params": params_value, + "timeoutSeconds": 300, + }); + let correlation_id = uuid::Uuid::now_v7().to_string(); + send( + client, + invocation, + reqwest::Method::POST, + &format!("/api/v1/proxmox/accounts/{account_id}/guests/{vmid}/{action}/run"), + &[], + Some(&run_body), + correlation_id, + ) +} + /// The `--wait` stages chase their own operation to a terminal state and /// then answer with the refreshed draft: the review surface, not the /// operation. @@ -2297,6 +2437,36 @@ fn request_for(command: &Command) -> Result { Vec::new(), Some(serde_json::json!({ "machineId": machine_id })), ), + Command::ProxmoxDestructive { + action, + account_id, + node, + vmid, + params, + .. + } => { + // Step one: the review. The run request is composed after the + // review's token arrives (see follow_review). The parameters + // arrive as JSON on standard input; the CLI never puts them + // in argv. + let text = match params.as_deref() { + Some(text) => text.to_owned(), + None => read_stdin_line("the action parameters as JSON")?, + }; + let params_value: serde_json::Value = + serde_json::from_str(&text).map_err(|error| CliError { + message: format!("the action parameters are not valid JSON: {error}"), + })?; + ( + reqwest::Method::POST, + format!("/api/v1/proxmox/accounts/{account_id}/guests/{vmid}/{action}/review"), + Vec::new(), + Some(serde_json::json!({ + "node": node, + "params": params_value, + })), + ) + } Command::ProxmoxLifecycle { action, account_id, diff --git a/crates/fleetctl/tests/cli.rs b/crates/fleetctl/tests/cli.rs index 02a73ee..b3c983d 100644 --- a/crates/fleetctl/tests/cli.rs +++ b/crates/fleetctl/tests/cli.rs @@ -2167,3 +2167,42 @@ fn parsing_walks_the_proxmox_lifecycle_forms() { let error = fleetctl::parse(&args).unwrap_err(); assert!(error.message.contains("must be a number"), "{error}"); } + +#[test] +fn parsing_walks_the_proxmox_destructive_forms() { + for verb in [ + "snapshot", + "snapshot-revert", + "snapshot-delete", + "clone", + "template", + "task-cancel", + ] { + let args: Vec = [ + "proxmox", + verb, + "--account", + "acc-1", + "--node", + "pve", + "--vmid", + "101", + ] + .iter() + .map(ToString::to_string) + .collect(); + let invocation = fleetctl::parse(&args).unwrap(); + match invocation.command { + fleetctl::Command::ProxmoxDestructive { action, .. } => { + assert_eq!(action, verb); + } + other => panic!("unexpected command {other:?}"), + } + } + let args: Vec = ["proxmox", "clone", "--account", "acc-1"] + .iter() + .map(ToString::to_string) + .collect(); + let error = fleetctl::parse(&args).unwrap_err(); + assert!(error.message.contains("is required"), "{error}"); +} diff --git a/crates/fleetd/tests/node_install.rs b/crates/fleetd/tests/node_install.rs index 960c887..0006a93 100644 --- a/crates/fleetd/tests/node_install.rs +++ b/crates/fleetd/tests/node_install.rs @@ -306,6 +306,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), + reviewed: false, }, ) .await diff --git a/crates/providers/fleet-provider-proxmox/src/lib.rs b/crates/providers/fleet-provider-proxmox/src/lib.rs index 3b8812b..a5c3324 100644 --- a/crates/providers/fleet-provider-proxmox/src/lib.rs +++ b/crates/providers/fleet-provider-proxmox/src/lib.rs @@ -83,6 +83,8 @@ pub enum PveHttpMethod { Get, /// A mutation. Post, + /// A removal. + Delete, } impl PveHttpRequest { @@ -179,6 +181,21 @@ pub trait PveTransport: fmt::Debug + Send + Sync { /// Fails with [`PveTransportError`]; HTTP statuses travel inside the /// response. async fn execute(&self, request: PveHttpRequest) -> Result; + + /// Executes one request with a JSON body. + /// + /// # Errors + /// + /// Fails with [`PveTransportError`]; HTTP statuses travel inside the + /// response. + async fn execute_with_body( + &self, + request: PveHttpRequest, + body: Vec, + ) -> Result { + let _ = body; + self.execute(request).await + } } /// The reqwest-backed transport: rustls with the pinned-fingerprint @@ -318,7 +335,25 @@ pub fn normalize_fingerprint(value: &str) -> String { #[async_trait] impl PveTransport for ReqwestPveTransport { + async fn execute_with_body( + &self, + request: PveHttpRequest, + body: Vec, + ) -> Result { + self.execute_inner(request, Some(body)).await + } + async fn execute(&self, request: PveHttpRequest) -> Result { + self.execute_inner(request, None).await + } +} + +impl ReqwestPveTransport { + async fn execute_inner( + &self, + request: PveHttpRequest, + body: Option>, + ) -> Result { let policy = match &request.pinned_fingerprint { Some(pinned) => TlsPolicy::Pin(normalize_fingerprint(pinned)), None => TlsPolicy::Observe, @@ -357,44 +392,48 @@ impl PveTransport for ReqwestPveTransport { let request_builder = match request.method { PveHttpMethod::Get => client.get(&url), PveHttpMethod::Post => client.post(&url), + PveHttpMethod::Delete => client.delete(&url), }; - let response = request_builder - .header( - "Authorization", - format!( - "PVEAPIToken={}={}", - request.credentials.token_id, - request.credentials.token.expose() - ), - ) - .send() - .await - .map_err(|error| { - // The verifier's refusal surfaces as an opaque connect - // error; the trust facts live in the capture the verifier - // wrote before refusing. - let observed = captured - .lock() - .expect("the capture lock is not poisoned") - .clone(); - match (observed, request.pinned_fingerprint.as_deref()) { - // A mismatch is only a mismatch when the fingerprints - // differ: a later TLS failure with a matching pin is a - // connection failure, not an instruction to re-confirm. - (Some(observed), Some(pinned)) - if normalize_fingerprint(&observed) != normalize_fingerprint(pinned) => - { - PveTransportError::FingerprintMismatch { - observed, - pinned: Some(pinned.to_owned()), - } + let request_builder = request_builder.header( + "Authorization", + format!( + "PVEAPIToken={}={}", + request.credentials.token_id, + request.credentials.token.expose() + ), + ); + let request_builder = match body { + Some(bytes) => request_builder + .header("Content-Type", "application/json") + .body(bytes), + None => request_builder, + }; + let response = request_builder.send().await.map_err(|error| { + // The verifier's refusal surfaces as an opaque connect + // error; the trust facts live in the capture the verifier + // wrote before refusing. + let observed = captured + .lock() + .expect("the capture lock is not poisoned") + .clone(); + match (observed, request.pinned_fingerprint.as_deref()) { + // A mismatch is only a mismatch when the fingerprints + // differ: a later TLS failure with a matching pin is a + // connection failure, not an instruction to re-confirm. + (Some(observed), Some(pinned)) + if normalize_fingerprint(&observed) != normalize_fingerprint(pinned) => + { + PveTransportError::FingerprintMismatch { + observed, + pinned: Some(pinned.to_owned()), } - (Some(_), Some(_)) | (None, _) => PveTransportError::Connect { - detail: error.to_string(), - }, - (Some(observed), None) => PveTransportError::ObserveRefused { observed }, } - })?; + (Some(_), Some(_)) | (None, _) => PveTransportError::Connect { + detail: error.to_string(), + }, + (Some(observed), None) => PveTransportError::ObserveRefused { observed }, + } + })?; let status = u16::from(response.status()); // The body bound is enforced while streaming: a hostile or broken // host cannot make Fleet materialize an unbounded response. @@ -763,6 +802,114 @@ pub trait ProxmoxSource: fmt::Debug + Send + Sync { request: PveHttpRequest, upid: &Upid, ) -> Result; + + /// Creates a snapshot of one guest. Idempotent on name: the caller + /// checks existence first (or classifies the existing snapshot), and + /// the provider refuses only transport/API-level failures. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_snapshot( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + description: &str, + include_ram: bool, + ) -> Result, PveApiError>; + + /// Rolls one guest back to a snapshot. The task is synchronous for + /// LXC and a UPID for QEMU; the provider normalizes both to an + /// optional UPID. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_snapshot_rollback( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + ) -> Result, PveApiError>; + + /// Deletes one snapshot. Synchronous on both guest kinds: the + /// outcome is immediate. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_snapshot_delete( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + ) -> Result<(), PveApiError>; + + /// Clones one guest to a new VMID with the requested name. The caller + /// classifies idempotency; the provider refuses only transport/API + /// failures. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_clone( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + new_id: u32, + name: &str, + full_copy: bool, + ) -> Result; + + /// Converts one guest into a template. Idempotent: converting an + /// existing template succeeds without an operation. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_convert_template( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + ) -> Result, PveApiError>; + + /// Stops one running task. Destructive-adjacent: the caller owns the + /// authorization; the outcome may be unknown if the task exits + /// between the stop and the status read. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn stop_task(&self, request: PveHttpRequest, upid: &Upid) -> Result<(), PveApiError>; + + /// Lists one guest's snapshots, normalized. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + async fn guest_snapshots( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + ) -> Result, PveApiError>; +} + +/// One guest snapshot, normalized. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PveSnapshot { + /// The snapshot's name. + pub name: String, + /// The snapshot's description, when carried. + pub description: String, + /// Whether the snapshot holds the guest's RAM. + pub includes_ram: bool, } /// The provider client: transport plus normalization. Stateless — every @@ -787,6 +934,37 @@ impl ProxmoxClient { Self { transport } } + /// One POST with a JSON body, unwrapping the envelope. + async fn call_with_body( + &self, + request: PveHttpRequest, + path: &str, + body: &serde_json::Value, + ) -> Result { + let mut request = request; + request.path = path.to_owned(); + request.method = PveHttpMethod::Post; + let response = self + .transport + .execute_with_body(request, body.to_string().into_bytes()) + .await + .map_err(PveApiError::Transport)?; + Self::status_to_result(&response) + } + + /// One DELETE, unwrapping the envelope. + async fn call_delete(&self, request: PveHttpRequest, path: &str) -> Result<(), PveApiError> { + let mut request = request; + request.path = path.to_owned(); + request.method = PveHttpMethod::Delete; + let response = self + .transport + .execute(request) + .await + .map_err(PveApiError::Transport)?; + Self::status_to_result(&response).map(|_| ()) + } + /// The version string and the raw cluster-resources entries: the /// prologue both discovery paths share. async fn version_and_resources( @@ -838,6 +1016,12 @@ impl ProxmoxClient { .execute(request) .await .map_err(PveApiError::Transport)?; + Self::status_to_result(&response) + } + + /// Maps one response onto the envelope: statuses become caller-safe + /// errors, a good body unwraps `{"data": ...}`. + fn status_to_result(response: &PveHttpResponse) -> Result { match response.status { 401 => Err(PveApiError::Auth), 403 => Err(PveApiError::Forbidden { @@ -877,6 +1061,12 @@ fn push_bounded(body: &mut Vec, chunk: &[u8]) -> Result<(), PveTransportErro Ok(()) } +/// The optional UPID string a mutating endpoint answered; a synchronous +/// outcome carries no UPID. +fn upid_from_data(data: &serde_json::Value) -> Option { + data.as_str().and_then(|raw| Upid::parse(raw).ok()) +} + /// A bounded, credential-free body excerpt for error details. fn bounded_body(body: &[u8]) -> String { let text = String::from_utf8_lossy(body); @@ -1031,6 +1221,179 @@ impl ProxmoxSource for ProxmoxClient { ) -> Result { self.task_status_impl(&request, upid).await } + async fn guest_snapshot( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + description: &str, + include_ram: bool, + ) -> Result, PveApiError> { + // PVE's qemu snapshot schema names the RAM flag `vmstate`; the + // caller's `include_ram` intent maps onto it. + let body = if include_ram { + serde_json::json!({ + "snapname": snapshot, + "description": description, + "vmstate": 1, + }) + } else { + serde_json::json!({ + "snapname": snapshot, + "description": description, + }) + }; + let data = self + .call_with_body( + request, + &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot"), + &body, + ) + .await?; + Ok(upid_from_data(&data)) + } + + async fn guest_snapshot_rollback( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + ) -> Result, PveApiError> { + let data = self + .call_with_body( + request, + &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot/{snapshot}/rollback"), + &serde_json::json!({}), + ) + .await?; + Ok(upid_from_data(&data)) + } + + async fn guest_snapshot_delete( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + snapshot: &str, + ) -> Result<(), PveApiError> { + self.call_delete( + request, + &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot/{snapshot}"), + ) + .await + } + + async fn guest_clone( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + new_id: u32, + name: &str, + full_copy: bool, + ) -> Result { + let body = serde_json::json!({ + "newid": new_id, + "name": name, + "full": full_copy, + }); + let data = self + .call_with_body( + request, + &format!("/api2/json/nodes/{node}/qemu/{vmid}/clone"), + &body, + ) + .await?; + let Some(upid_raw) = data.as_str() else { + return Err(PveApiError::InvalidPayload { + detail: "the clone answer carries no UPID string".to_owned(), + }); + }; + Upid::parse(upid_raw).map_err(|detail| PveApiError::InvalidPayload { detail }) + } + + async fn guest_convert_template( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + ) -> Result, PveApiError> { + let data = self + .call_with_body( + request, + &format!("/api2/json/nodes/{node}/qemu/{vmid}/template"), + &serde_json::json!({}), + ) + .await?; + Ok(upid_from_data(&data)) + } + + async fn stop_task(&self, request: PveHttpRequest, upid: &Upid) -> Result<(), PveApiError> { + self.call_delete( + request, + &format!( + "/api2/json/nodes/{}/tasks/{}/status", + urlencode(&upid.node), + urlencode(&upid.raw) + ), + ) + .await + } + + async fn guest_snapshots( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + ) -> Result, PveApiError> { + let data = self + .call(PveHttpRequest { + path: format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot"), + ..request.clone() + }) + .await?; + let entries = match data { + serde_json::Value::Array(entries) => entries, + serde_json::Value::Null => Vec::new(), + other => { + return Err(PveApiError::InvalidPayload { + detail: format!( + "the snapshots payload is not a list (it is a {})", + type_name_of(&other) + ), + }); + } + }; + let mut snapshots = Vec::new(); + for entry in entries { + // `current` is the live state marker, not a snapshot. + let name = entry + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + if name == "current" || name.is_empty() { + continue; + } + snapshots.push(PveSnapshot { + name: name.chars().take(64).collect(), + description: entry + .get("description") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .chars() + .take(512) + .collect(), + includes_ram: entry + .get("vmstate") + .and_then(serde_json::Value::as_i64) + .unwrap_or(0) + == 1, + }); + } + Ok(snapshots) + } } /// The config's `netN` entries, parsed for MAC addresses. The value shape @@ -1177,6 +1540,30 @@ impl ProxmoxClient { agent } + /// Lists the cluster's QEMU resources by the caller's purpose: the + /// idempotency classifications read the cluster's truth, not a cache. + /// + /// # Errors + /// + /// Fails with [`PveApiError`]. + pub async fn list_qemu_resources( + &self, + request: PveHttpRequest, + ) -> Result, PveApiError> { + let (version, entries) = self.version_and_resources(&request).await?; + let mut resources = Vec::new(); + for entry in entries { + if let Ok(Some(resource)) = normalize_resource(&entry) { + let _ = &version; + resources.push(resource); + } + } + Ok(resources + .into_iter() + .filter(|resource| resource.kind == "qemu" || resource.kind == "qemu-template") + .collect()) + } + /// Reads one task's status. async fn task_status_impl( &self, diff --git a/packages/api-client/openapi.json b/packages/api-client/openapi.json index ed0c67a..87f7fff 100644 --- a/packages/api-client/openapi.json +++ b/packages/api-client/openapi.json @@ -2500,6 +2500,172 @@ } } }, + "/api/v1/proxmox/accounts/{accountId}/guests/{vmid}/{action}/review": { + "post": { + "tags": [ + "proxmox" + ], + "summary": "Reviews one destructive-adjacent operation: renders exactly what will\nrun and returns the token the create call must present.", + "description": "# Errors\n\nReturns the public error envelope on refusal or a malformed request.", + "operationId": "reviewProxmoxOperation", + "parameters": [ + { + "name": "accountId", + "in": "path", + "description": "The account's identity.", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "vmid", + "in": "path", + "description": "The guest's VMID.", + "required": true, + "schema": { + "type": "integer", + "format": "int32", + "minimum": 0 + } + }, + { + "name": "action", + "in": "path", + "description": "The destructive action: snapshot, snapshot-revert, snapshot-delete, clone, template, or task-cancel.", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReviewProxmoxOperationRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "The review record; present its token to the create call.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Resource_ProxmoxReviewDto" + } + } + } + }, + "400": { + "description": "The action is unrecognized or the request is malformed.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + }, + "403": { + "description": "The caller may not operate Proxmox guests destructively.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + } + } + } + }, + "/api/v1/proxmox/accounts/{accountId}/guests/{vmid}/{action}/run": { + "post": { + "tags": [ + "proxmox" + ], + "summary": "Runs a reviewed destructive-adjacent operation as a durable operation.\nThe review token must match the request exactly: what runs is what was\nreviewed. The generic operations surface refuses these kinds outright.", + "description": "# Errors\n\nReturns the public error envelope on refusal, a missing/stale review\ntoken, or a malformed request.", + "operationId": "startReviewedProxmoxOperation", + "parameters": [ + { + "name": "accountId", + "in": "path", + "description": "The account's identity.", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "vmid", + "in": "path", + "description": "The guest's VMID.", + "required": true, + "schema": { + "type": "integer", + "format": "int32", + "minimum": 0 + } + }, + { + "name": "action", + "in": "path", + "description": "The reviewed destructive action.", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StartReviewedProxmoxOperationRequest" + } + } + }, + "required": true + }, + "responses": { + "202": { + "description": "The operation was accepted and is durable.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Resource_OperationDto" + } + } + } + }, + "400": { + "description": "The action is unrecognized or the review token does not match the request.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + }, + "403": { + "description": "The caller may not operate Proxmox guests destructively.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + } + } + } + }, "/api/v1/proxmox/accounts/{accountId}/observe": { "post": { "tags": [ @@ -5713,6 +5879,45 @@ } } }, + "ProxmoxReviewDto": { + "type": "object", + "description": "The review record: exactly what the destructive operation will run,\nbound to a token the create call must present.", + "required": [ + "reviewToken", + "action", + "node", + "vmid", + "accountId", + "params" + ], + "properties": { + "accountId": { + "type": "string", + "description": "The account the operation will run under." + }, + "action": { + "type": "string", + "description": "The action under review." + }, + "node": { + "type": "string", + "description": "The guest's hosting node." + }, + "params": { + "description": "The action-specific parameters, as reviewed." + }, + "reviewToken": { + "type": "string", + "description": "The token binding this review to the request it described." + }, + "vmid": { + "type": "integer", + "format": "int32", + "description": "The guest's VMID.", + "minimum": 0 + } + } + }, "ReadyAuthDto": { "oneOf": [ { @@ -6652,6 +6857,54 @@ } } }, + "Resource_ProxmoxReviewDto": { + "type": "object", + "description": "A single resource.\n\nThe payload is nested under `data` so that later top-level fields are an\nadditive change rather than a breaking one.", + "required": [ + "data" + ], + "properties": { + "data": { + "type": "object", + "description": "The review record: exactly what the destructive operation will run,\nbound to a token the create call must present.", + "required": [ + "reviewToken", + "action", + "node", + "vmid", + "accountId", + "params" + ], + "properties": { + "accountId": { + "type": "string", + "description": "The account the operation will run under." + }, + "action": { + "type": "string", + "description": "The action under review." + }, + "node": { + "type": "string", + "description": "The guest's hosting node." + }, + "params": { + "description": "The action-specific parameters, as reviewed." + }, + "reviewToken": { + "type": "string", + "description": "The token binding this review to the request it described." + }, + "vmid": { + "type": "integer", + "format": "int32", + "description": "The guest's VMID.", + "minimum": 0 + } + } + } + } + }, "Resource_ReadyPlanDto": { "type": "object", "description": "A single resource.\n\nThe payload is nested under `data` so that later top-level fields are an\nadditive change rather than a breaking one.", @@ -6740,6 +6993,22 @@ "backoff" ] }, + "ReviewProxmoxOperationRequest": { + "type": "object", + "description": "The destructive-review request.", + "required": [ + "node" + ], + "properties": { + "node": { + "type": "string", + "description": "The guest's hosting node." + }, + "params": { + "description": "The action-specific parameters (snapshot name/description, clone\ntarget id/name, and so on)." + } + } + }, "SkillsAuthDto": { "oneOf": [ { @@ -7065,6 +7334,34 @@ } } }, + "StartReviewedProxmoxOperationRequest": { + "type": "object", + "description": "The create call for a reviewed destructive operation: carries the\nreview token binding it to what was reviewed.", + "required": [ + "node", + "reviewToken", + "timeoutSeconds" + ], + "properties": { + "node": { + "type": "string", + "description": "The guest's hosting node." + }, + "params": { + "description": "The action-specific parameters, identical to the reviewed ones." + }, + "reviewToken": { + "type": "string", + "description": "The review token from the review call." + }, + "timeoutSeconds": { + "type": "integer", + "format": "int64", + "description": "The deadline, in seconds. Bounded by the executor.", + "minimum": 0 + } + } + }, "StartSkillsOperationRequest": { "type": "object", "description": "The body of the start-skills-operation request.", diff --git a/packages/api-client/src/generated/fleet.ts b/packages/api-client/src/generated/fleet.ts index d2da351..aa85a0c 100644 --- a/packages/api-client/src/generated/fleet.ts +++ b/packages/api-client/src/generated/fleet.ts @@ -1550,6 +1550,28 @@ export interface ProxmoxFingerprintDto { fingerprint: string; } +/** + * The review record: exactly what the destructive operation will run, + * bound to a token the create call must present. + */ +export interface ProxmoxReviewDto { + /** The account the operation will run under. */ + accountId: string; + /** The action under review. */ + action: string; + /** The guest's hosting node. */ + node: string; + /** The action-specific parameters, as reviewed. */ + params: unknown; + /** The token binding this review to the request it described. */ + reviewToken: string; + /** + * The guest's VMID. + * @minimum 0 + */ + vmid: number; +} + /** * How the workflow's endpoint authenticates. */ @@ -2096,6 +2118,42 @@ export interface ResourceProxmoxFingerprintDto { data: ResourceProxmoxFingerprintDtoData; } +/** + * The review record: exactly what the destructive operation will run, + * bound to a token the create call must present. + */ +export type ResourceProxmoxReviewDtoData = { + /** The account the operation will run under. */ + accountId: string; + /** The action under review. */ + action: string; + /** The guest's hosting node. */ + node: string; + /** The action-specific parameters, as reviewed. */ + params: unknown; + /** The token binding this review to the request it described. */ + reviewToken: string; + /** + * The guest's VMID. + * @minimum 0 + */ + vmid: number; +}; + +/** + * A single resource. + * + * The payload is nested under `data` so that later top-level fields are an + * additive change rather than a breaking one. + */ +export interface ResourceProxmoxReviewDto { + /** + * The review record: exactly what the destructive operation will run, + * bound to a token the create call must present. + */ + data: ResourceProxmoxReviewDtoData; +} + /** * The dry run's plan response: the step vocabulary and the conditions * under which each step runs. @@ -2157,6 +2215,19 @@ export interface ResourceTailnetStatusDto { data: ResourceTailnetStatusDtoData; } +/** + * The destructive-review request. + */ +export interface ReviewProxmoxOperationRequest { + /** The guest's hosting node. */ + node: string; + /** + * The action-specific parameters (snapshot name/description, clone + * target id/name, and so on). + */ + params?: unknown; +} + /** * How a skills operation's endpoint authenticates. */ @@ -2337,6 +2408,24 @@ export interface StartReadyRequest { tools?: ReadyToolDto[]; } +/** + * The create call for a reviewed destructive operation: carries the + * review token binding it to what was reviewed. + */ +export interface StartReviewedProxmoxOperationRequest { + /** The guest's hosting node. */ + node: string; + /** The action-specific parameters, identical to the reviewed ones. */ + params?: unknown; + /** The review token from the review call. */ + reviewToken: string; + /** + * The deadline, in seconds. Bounded by the executor. + * @minimum 0 + */ + timeoutSeconds: number; +} + /** * The body of the start-skills-operation request. */ @@ -5122,6 +5211,148 @@ const res = await fetch(getStartProxmoxLifecycleUrl(accountId,vmid,action), +export type reviewProxmoxOperationResponse200 = { + data: ResourceProxmoxReviewDto + status: 200 +} + +export type reviewProxmoxOperationResponse400 = { + data: ApiError + status: 400 +} + +export type reviewProxmoxOperationResponse403 = { + data: ApiError + status: 403 +} + +export type reviewProxmoxOperationResponseSuccess = (reviewProxmoxOperationResponse200) & { + headers: Headers; +}; +export type reviewProxmoxOperationResponseError = (reviewProxmoxOperationResponse400 | reviewProxmoxOperationResponse403) & { + headers: Headers; +}; + +export type reviewProxmoxOperationResponse = (reviewProxmoxOperationResponseSuccess | reviewProxmoxOperationResponseError) + +export const getReviewProxmoxOperationUrl = (accountId: string, + vmid: number, + action: string,) => { + + + + + return `/api/v1/proxmox/accounts/${accountId}/guests/${vmid}/${action}/review` +} + +/** + * # Errors + * + * Returns the public error envelope on refusal or a malformed request. + * @summary Reviews one destructive-adjacent operation: renders exactly what will +run and returns the token the create call must present. + */ +export const reviewProxmoxOperation = async (accountId: string, + vmid: number, + action: string, + reviewProxmoxOperationRequest: ReviewProxmoxOperationRequest, options?: RequestInit): Promise => { + + const getHeaders = (h?: NonNullable): Record => { + if (!h) return {}; + if (h instanceof Headers) return Object.fromEntries(h.entries()); + if (Array.isArray(h)) return Object.fromEntries(h); + return h; + }; +const res = await fetch(getReviewProxmoxOperationUrl(accountId,vmid,action), + { + ...options, + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getHeaders(options?.headers) }, + body: JSON.stringify(reviewProxmoxOperationRequest) + } +) + + + const body = [204, 205, 304].includes(res.status) ? null : await res.text(); + + const data: reviewProxmoxOperationResponse['data'] = body ? JSON.parse(body) : {} + return { data, status: res.status, headers: res.headers } as reviewProxmoxOperationResponse +} + + + +export type startReviewedProxmoxOperationResponse202 = { + data: ResourceOperationDto + status: 202 +} + +export type startReviewedProxmoxOperationResponse400 = { + data: ApiError + status: 400 +} + +export type startReviewedProxmoxOperationResponse403 = { + data: ApiError + status: 403 +} + +export type startReviewedProxmoxOperationResponseSuccess = (startReviewedProxmoxOperationResponse202) & { + headers: Headers; +}; +export type startReviewedProxmoxOperationResponseError = (startReviewedProxmoxOperationResponse400 | startReviewedProxmoxOperationResponse403) & { + headers: Headers; +}; + +export type startReviewedProxmoxOperationResponse = (startReviewedProxmoxOperationResponseSuccess | startReviewedProxmoxOperationResponseError) + +export const getStartReviewedProxmoxOperationUrl = (accountId: string, + vmid: number, + action: string,) => { + + + + + return `/api/v1/proxmox/accounts/${accountId}/guests/${vmid}/${action}/run` +} + +/** + * # Errors + * + * Returns the public error envelope on refusal, a missing/stale review + * token, or a malformed request. + * @summary Runs a reviewed destructive-adjacent operation as a durable operation. +The review token must match the request exactly: what runs is what was +reviewed. The generic operations surface refuses these kinds outright. + */ +export const startReviewedProxmoxOperation = async (accountId: string, + vmid: number, + action: string, + startReviewedProxmoxOperationRequest: StartReviewedProxmoxOperationRequest, options?: RequestInit): Promise => { + + const getHeaders = (h?: NonNullable): Record => { + if (!h) return {}; + if (h instanceof Headers) return Object.fromEntries(h.entries()); + if (Array.isArray(h)) return Object.fromEntries(h); + return h; + }; +const res = await fetch(getStartReviewedProxmoxOperationUrl(accountId,vmid,action), + { + ...options, + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getHeaders(options?.headers) }, + body: JSON.stringify(startReviewedProxmoxOperationRequest) + } +) + + + const body = [204, 205, 304].includes(res.status) ? null : await res.text(); + + const data: startReviewedProxmoxOperationResponse['data'] = body ? JSON.parse(body) : {} + return { data, status: res.status, headers: res.headers } as startReviewedProxmoxOperationResponse +} + + + export type observeProxmoxFingerprintResponse200 = { data: ResourceProxmoxFingerprintDto status: 200 From 56ac3001d67838bcd7849e2c30ca5ff15c1a0470 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Mon, 21 Sep 2026 11:26:03 +0000 Subject: [PATCH 2/4] FM-603: address the cubic review findings --- Cargo.lock | 2 +- crates/fleet-api/Cargo.toml | 1 - crates/fleet-api/src/apply.rs | 2 +- crates/fleet-api/src/frogenv.rs | 2 +- crates/fleet-api/src/mise.rs | 2 +- crates/fleet-api/src/onboarding.rs | 2 +- crates/fleet-api/src/operations.rs | 2 +- crates/fleet-api/src/projects.rs | 2 +- crates/fleet-api/src/proxmox.rs | 149 ++++++++++++------ crates/fleet-api/src/ready.rs | 2 +- crates/fleet-api/src/skills.rs | 2 +- crates/fleet-application/Cargo.toml | 1 + crates/fleet-application/src/operation.rs | 86 +++++++--- crates/fleet-controller/src/apply.rs | 4 +- crates/fleet-controller/src/proxmox_exec.rs | 49 +++++- crates/fleet-controller/src/ready.rs | 2 +- .../tests/agentless_inventory.rs | 2 +- crates/fleet-controller/tests/apply.rs | 8 +- crates/fleet-controller/tests/checkout.rs | 2 +- crates/fleet-controller/tests/frogenv.rs | 2 +- crates/fleet-controller/tests/gateway.rs | 2 +- crates/fleet-controller/tests/machines.rs | 2 +- crates/fleet-controller/tests/mise.rs | 2 +- .../tests/operation_worker.rs | 12 +- crates/fleet-controller/tests/proxmox.rs | 23 ++- crates/fleet-controller/tests/ready.rs | 2 +- crates/fleet-controller/tests/skills.rs | 2 +- crates/fleet-controller/tests/ssh_exec.rs | 8 +- .../tests/worker_lifecycle.rs | 2 +- crates/fleetctl/src/lib.rs | 27 ++-- crates/fleetctl/tests/cli.rs | 62 ++++++++ crates/fleetd/tests/node_install.rs | 2 +- .../fleet-provider-proxmox/src/lib.rs | 53 +++++-- 33 files changed, 375 insertions(+), 148 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9a28387..9bdb6f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -567,7 +567,6 @@ dependencies = [ "http-body-util", "serde", "serde_json", - "sha2 0.10.9", "tokio", "tower", "utoipa", @@ -582,6 +581,7 @@ dependencies = [ "fleet-core", "serde", "serde_json", + "sha2 0.10.9", "tokio", ] diff --git a/crates/fleet-api/Cargo.toml b/crates/fleet-api/Cargo.toml index b40b32a..0669713 100644 --- a/crates/fleet-api/Cargo.toml +++ b/crates/fleet-api/Cargo.toml @@ -12,7 +12,6 @@ axum = "0.8.9" fleet-application = { version = "0.1.0", path = "../fleet-application" } fleet-core = { path = "../fleet-core" } futures-util = "0.3.31" -sha2 = "0.10" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["time", "macros", "rt", "net"] } diff --git a/crates/fleet-api/src/apply.rs b/crates/fleet-api/src/apply.rs index f4ff108..8b328fd 100644 --- a/crates/fleet-api/src/apply.rs +++ b/crates/fleet-api/src/apply.rs @@ -287,7 +287,7 @@ pub async fn start_apply_workflow( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/frogenv.rs b/crates/fleet-api/src/frogenv.rs index 68594d3..b331e90 100644 --- a/crates/fleet-api/src/frogenv.rs +++ b/crates/fleet-api/src/frogenv.rs @@ -250,7 +250,7 @@ pub async fn start_frogenv_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/mise.rs b/crates/fleet-api/src/mise.rs index d6a26ef..8b563ed 100644 --- a/crates/fleet-api/src/mise.rs +++ b/crates/fleet-api/src/mise.rs @@ -286,7 +286,7 @@ pub async fn start_mise_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/onboarding.rs b/crates/fleet-api/src/onboarding.rs index 21a3f74..1fa2dc2 100644 --- a/crates/fleet-api/src/onboarding.rs +++ b/crates/fleet-api/src/onboarding.rs @@ -702,7 +702,7 @@ async fn start_onboarding_operation( deadline_at: Some(fleet_core::SystemClock::now_unix_millis() + deadline_ms), correlation_id: Some(correlation_id.to_string()), payload_json: Some(serde_json::json!({ "draftId": draft_id }).to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/operations.rs b/crates/fleet-api/src/operations.rs index 0a6969a..a344e47 100644 --- a/crates/fleet-api/src/operations.rs +++ b/crates/fleet-api/src/operations.rs @@ -395,7 +395,7 @@ pub async fn create_operation( deadline_at: request.deadline_at, correlation_id: Some(correlation_id.to_string()), payload_json: request.payload_json.clone(), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/projects.rs b/crates/fleet-api/src/projects.rs index a08b911..49690d3 100644 --- a/crates/fleet-api/src/projects.rs +++ b/crates/fleet-api/src/projects.rs @@ -634,7 +634,7 @@ pub async fn start_discovery( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/proxmox.rs b/crates/fleet-api/src/proxmox.rs index 16c0db2..87666dd 100644 --- a/crates/fleet-api/src/proxmox.rs +++ b/crates/fleet-api/src/proxmox.rs @@ -1088,7 +1088,7 @@ pub async fn start_proxmox_lifecycle( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: true, + review_token: None, }, ) .await @@ -1132,33 +1132,70 @@ pub struct ReviewProxmoxOperationRequest { pub params: serde_json::Value, } -/// The token binding one reviewed request to its create call: the SHA-256 -/// of the canonical reviewed material. Stateless — the controller holds no -/// review store; a create presents the token computed from exactly the -/// payload it carries, so what runs is what was reviewed. +/// The token binding one reviewed request to its create call: the +/// application's payload-bound review token, computed over the exact +/// operation bytes the run will carry. #[must_use] -fn review_token( - account_id: &str, - vmid: u32, - action: &str, - request: &ReviewProxmoxOperationRequest, -) -> String { - use sha2::Digest as _; - let canonical = serde_json::json!({ - "accountId": account_id, - "vmid": vmid, - "action": action, - "node": request.node, - "params": request.params, - }); - let mut hasher = sha2::Sha256::new(); - hasher.update(serde_json::to_string(&canonical).unwrap_or_default()); - let digest: [u8; 32] = hasher.finalize().into(); - digest.iter().fold(String::with_capacity(64), |mut out, b| { - use std::fmt::Write as _; - let _ = write!(out, "{b:02x}"); - out - }) +fn review_token(kind: &str, payload_json: &str) -> String { + fleet_application::operation::review_token_for(kind, payload_json) +} + +/// Validates one action's required parameters: a malformed review is a +/// client error, not a durable failure. +fn validate_destructive_params(action: &str, params: &serde_json::Value) -> Result<(), String> { + match action { + "snapshot" => { + let name = params + .get("snapshot") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + if name.is_empty() { + return Err("the snapshot action requires a snapshot name".to_owned()); + } + Ok(()) + } + "snapshot-revert" | "snapshot-delete" => { + let name = params + .get("snapshot") + .and_then(serde_json::Value::as_str) + .unwrap_or_default(); + if name.is_empty() { + return Err( + "the {action} action requires a snapshot name".replace("{action}", action) + ); + } + Ok(()) + } + "clone" => { + if params + .get("newId") + .and_then(serde_json::Value::as_u64) + .is_none() + { + return Err("the clone action requires a newId".to_owned()); + } + if params + .get("name") + .and_then(serde_json::Value::as_str) + .is_none() + { + return Err("the clone action requires a name".to_owned()); + } + Ok(()) + } + "task-cancel" => { + if params + .get("upid") + .and_then(serde_json::Value::as_str) + .is_none() + { + return Err("the task-cancel action requires a upid".to_owned()); + } + Ok(()) + } + // `template` carries no parameters. + _ => Ok(()), + } } /// Reviews one destructive-adjacent operation: renders exactly what will @@ -1219,6 +1256,12 @@ pub async fn review_proxmox_operation( correlation_id, )); } + // Each action's required parameters are validated here, before a + // token exists: a malformed request is a 400, never a durable + // operation the worker immediately fails. + if let Err(detail) = validate_destructive_params(&action, &request.params) { + return Err(crate::machines::invalid_request(&detail, correlation_id)); + } if let Err(decision) = fleet_application::authz::authorize( state.authorizer.as_ref(), fleet_application::authz::AccessRequest { @@ -1229,8 +1272,23 @@ pub async fn review_proxmox_operation( ) { return Err(crate::machines::denied_error(decision, correlation_id)); } + // The token binds the exact payload the run will carry, so the + // reviewed bytes and the executed bytes are the same by construction. + let kind = if action == "task-cancel" { + "proxmox.task-cancel".to_owned() + } else { + format!("proxmox.guest.{action}") + }; + let payload = serde_json::json!({ + "accountId": account_id, + "node": request.node, + "vmid": vmid, + "timeoutSeconds": 300, + "params": request.params, + }); + let token = review_token(&kind, &payload.to_string()); Ok(Json(Resource::new(ProxmoxReviewDto { - review_token: review_token(&account_id, vmid, &action, &request), + review_token: token, action, node: request.node, vmid, @@ -1320,40 +1378,27 @@ pub async fn start_reviewed_proxmox_operation( ) { return Err(crate::machines::denied_error(decision, correlation_id)); } - // The review gate: the token is recomputed from THIS request, so the - // create runs only what was reviewed — any parameter change invalidates - // the token. - let presented = ReviewProxmoxOperationRequest { - node: request.node.clone(), - params: request.params.clone(), - }; - let expected = review_token(&account_id, vmid, &action, &presented); - if expected != request.review_token { - return Err(crate::machines::invalid_request( - "the review token does not match this request; review again", - correlation_id, - )); - } if request.node.is_empty() || request.node.len() > 128 { return Err(crate::machines::invalid_request( "the node must be 1..=128 characters", correlation_id, )); } - let mut payload = serde_json::json!({ + // The payload is byte-identical to the reviewed one (the timeout is + // fixed at review time), so the application's token comparison binds + // what runs to what was reviewed. + let kind = if action == "task-cancel" { + "proxmox.task-cancel".to_owned() + } else { + format!("proxmox.guest.{action}") + }; + let payload = serde_json::json!({ "accountId": account_id, "node": request.node, "vmid": vmid, - "timeoutSeconds": request.timeout_seconds, + "timeoutSeconds": 300, "params": request.params, }); - let _ = &mut payload; - let kind = format!("proxmox.guest.{action}"); - let kind = if action == "task-cancel" { - "proxmox.task-cancel".to_owned() - } else { - kind - }; let idempotency_key = headers .get(crate::IDEMPOTENCY_KEY_HEADER) .and_then(|value| value.to_str().ok()) @@ -1369,7 +1414,7 @@ pub async fn start_reviewed_proxmox_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: true, + review_token: Some(request.review_token), }, ) .await diff --git a/crates/fleet-api/src/ready.rs b/crates/fleet-api/src/ready.rs index 525a087..6d4e782 100644 --- a/crates/fleet-api/src/ready.rs +++ b/crates/fleet-api/src/ready.rs @@ -269,7 +269,7 @@ pub async fn start_ready_workflow( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-api/src/skills.rs b/crates/fleet-api/src/skills.rs index 53f9a2b..21eca9c 100644 --- a/crates/fleet-api/src/skills.rs +++ b/crates/fleet-api/src/skills.rs @@ -205,7 +205,7 @@ pub async fn start_skills_operation( deadline_at: None, correlation_id: Some(correlation_id.to_string()), payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-application/Cargo.toml b/crates/fleet-application/Cargo.toml index 78cb964..8e7560f 100644 --- a/crates/fleet-application/Cargo.toml +++ b/crates/fleet-application/Cargo.toml @@ -10,6 +10,7 @@ publish.workspace = true async-trait = "0.1.89" fleet-core = { path = "../fleet-core" } tokio = { version = "1", default-features = false, features = ["rt", "time"] } +sha2 = "0.10" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index f1cae66..3e49be7 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -157,6 +157,48 @@ fn machine_scoped_kind_permission_inner(kind: &str, payload: Option<&str>) -> Op /// The payload bound for provider inputs. pub const MAX_PAYLOAD_JSON: usize = 128 * 1024; +/// The destructive-adjacent kinds: their creation requires a review token +/// computed over exactly the payload being created. +pub const DESTRUCTIVE_KINDS: [&str; 6] = [ + "proxmox.guest.snapshot", + "proxmox.guest.snapshot-revert", + "proxmox.guest.snapshot-delete", + "proxmox.guest.clone", + "proxmox.guest.template", + "proxmox.task-cancel", +]; + +/// The review token for one destructive operation: the SHA-256 of the +/// kind and the exact payload bytes. Deterministic, payload-bound, and +/// computable only over material the caller actually holds. +#[must_use] +pub fn review_token_for(kind: &str, payload_json: &str) -> String { + use sha2::Digest as _; + let mut hasher = sha2::Sha256::new(); + hasher.update(kind.as_bytes()); + hasher.update(b"\n"); + hasher.update(payload_json.as_bytes()); + let digest: [u8; 32] = hasher.finalize().into(); + digest.iter().fold(String::with_capacity(64), |mut out, b| { + use std::fmt::Write as _; + let _ = write!(out, "{b:02x}"); + out + }) +} + +/// Compares two token strings in constant time over their bytes. +fn constant_time_eq(a: &str, b: &str) -> bool { + let (a, b) = (a.as_bytes(), b.as_bytes()); + if a.len() != b.len() { + return false; + } + let mut diff = 0u8; + for (x, y) in a.iter().zip(b.iter()) { + diff |= x ^ y; + } + diff == 0 +} + /// The public view of a durable operation. #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -450,11 +492,13 @@ pub struct NewOperation { pub correlation_id: Option, /// The bounded provider input, for kinds that need one. pub payload_json: Option, - /// Whether the request arrived through a dedicated endpoint that - /// already enforced the kind's extra gate (e.g. the destructive - /// review). Only the dedicated surface sets it; the generic surface - /// leaves it off, which refuses destructive kinds outright. - pub reviewed: bool, + /// The verified review material for destructive-adjacent kinds: the + /// review token recomputed over exactly this payload, which only a + /// caller that ran the review over the same bytes can present. The + /// generic surface leaves it `None`, which refuses destructive kinds + /// outright; a `Some` value that does not match the payload is refused + /// as well, so the field cannot be forged by setting it. + pub review_token: Option, } /// The authorized operation use cases. @@ -548,19 +592,25 @@ impl Operations { // dedicated reviewed endpoint, which binds the operation to a // confirmed review token. A generic-surface create is a // route-around attempt, refused as malformed. - if !new.reviewed - && (new.kind.starts_with("proxmox.guest.snapshot") - || matches!( - new.kind.as_str(), - "proxmox.guest.clone" | "proxmox.guest.template" | "proxmox.task-cancel" - )) - { - return Err(OperationUseCaseError::Invalid { - detail: format!( - "the {kind} kind is destructive-adjacent and runs only through its reviewed dedicated endpoint", - kind = new.kind - ), - }); + if DESTRUCTIVE_KINDS.contains(&new.kind.as_str()) { + // The review token is the SHA-256 of the canonical + // operation material, recomputed here: a caller that never + // reviewed these exact bytes cannot present a matching + // token, and setting the field arbitrarily fails the + // comparison. + let expected = + review_token_for(&new.kind, new.payload_json.as_deref().unwrap_or_default()); + match new.review_token.as_deref() { + Some(presented) if constant_time_eq(presented, &expected) => {} + _ => { + return Err(OperationUseCaseError::Invalid { + detail: format!( + "the {kind} kind is destructive-adjacent and requires a valid review token; review the operation first", + kind = new.kind + ), + }); + } + } } authorize( authorizer, diff --git a/crates/fleet-controller/src/apply.rs b/crates/fleet-controller/src/apply.rs index 29b765f..c961ba4 100644 --- a/crates/fleet-controller/src/apply.rs +++ b/crates/fleet-controller/src/apply.rs @@ -551,7 +551,7 @@ impl ApplyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_owned()), - reviewed: false, + review_token: None, }, ) .await @@ -640,7 +640,7 @@ impl ApplyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/src/proxmox_exec.rs b/crates/fleet-controller/src/proxmox_exec.rs index c2a1da0..92d9905 100644 --- a/crates/fleet-controller/src/proxmox_exec.rs +++ b/crates/fleet-controller/src/proxmox_exec.rs @@ -168,7 +168,10 @@ impl ProxmoxLifecycleExecutor { .task_status(request.clone(), &upid) .await .map_err(|error| { - format!("the task status failed: {error}; the task's outcome is unknown") + format!( + "the task status failed: {error}; the task {} on node {} keeps its outcome unknown", + upid.raw, upid.node + ) })?; match status { TaskStatus::Running => {} @@ -595,7 +598,7 @@ impl OperationExecutor for ProxmoxDestructiveExecutor { // resources are the truth. let resources = self .client - .list_qemu_resources(request.clone()) + .list_guest_resources(request.clone()) .await .map_err(|error| format!("the resource listing failed: {error}"))?; if resources @@ -639,7 +642,7 @@ impl OperationExecutor for ProxmoxDestructiveExecutor { // succeeds without an operation. let resources = self .client - .list_qemu_resources(request.clone()) + .list_guest_resources(request.clone()) .await .map_err(|error| format!("the resource listing failed: {error}"))?; let resource = resources @@ -678,6 +681,33 @@ impl OperationExecutor for ProxmoxDestructiveExecutor { .and_then(serde_json::Value::as_str) .ok_or("the reviewed parameters carry no UPID")?; let upid = Upid::parse(raw)?; + // The reviewed UPID must belong to the reviewed guest: a + // task on another node or for another VMID is refused, so + // the review scope is the cancellation scope. + if upid.node != payload.node { + return complete_failure( + operations, + &operation.id, + "conflict", + &format!( + "the reviewed task runs on node {}, not the reviewed node {}", + upid.node, payload.node + ), + ) + .await; + } + if upid.target != payload.vmid.to_string() { + return complete_failure( + operations, + &operation.id, + "conflict", + &format!( + "the reviewed task targets {}, not the reviewed guest qemu/{}", + upid.target, payload.vmid + ), + ) + .await; + } self.client .stop_task(request.clone(), &upid) .await @@ -861,8 +891,8 @@ impl ProxmoxDestructiveExecutor { operation_id, "cancelled", &format!( - "cancelled while waiting; the remote task on node {} keeps running and its outcome is unknown", - upid.node + "cancelled while waiting; the remote task {} on node {} keeps running and its outcome is unknown", + upid.raw, upid.node ), ) .await; @@ -872,7 +902,10 @@ impl ProxmoxDestructiveExecutor { .task_status(request.clone(), &upid) .await .map_err(|error| { - format!("the task status failed: {error}; the task's outcome is unknown") + format!( + "the task status failed: {error}; the task {} on node {} keeps its outcome unknown", + upid.raw, upid.node + ) })?; match status { TaskStatus::Running => {} @@ -884,8 +917,8 @@ impl ProxmoxDestructiveExecutor { operation_id, "deadline_expired", &format!( - "the deadline expired while the task still runs; its final state is unknown (task on node {})", - upid.node + "the deadline expired while the task still runs; its final state is unknown (task {} on node {})", + upid.raw, upid.node ), ) .await; diff --git a/crates/fleet-controller/src/ready.rs b/crates/fleet-controller/src/ready.rs index 67a32d3..1c1135d 100644 --- a/crates/fleet-controller/src/ready.rs +++ b/crates/fleet-controller/src/ready.rs @@ -526,7 +526,7 @@ impl ReadyExecutor { deadline_at: None, correlation_id: None, payload_json: Some(payload_json.to_owned()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/agentless_inventory.rs b/crates/fleet-controller/tests/agentless_inventory.rs index c800911..7d99394 100644 --- a/crates/fleet-controller/tests/agentless_inventory.rs +++ b/crates/fleet-controller/tests/agentless_inventory.rs @@ -194,7 +194,7 @@ async fn an_inventory_operation_probes_and_ingests() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/apply.rs b/crates/fleet-controller/tests/apply.rs index 6aa7395..42552e5 100644 --- a/crates/fleet-controller/tests/apply.rs +++ b/crates/fleet-controller/tests/apply.rs @@ -166,7 +166,7 @@ async fn an_unapproved_plan_completes_blocked_naming_the_steps() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await @@ -238,7 +238,7 @@ async fn a_kind_state_mismatched_payload_fails_honestly() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await @@ -311,7 +311,7 @@ async fn an_approved_plan_executes_every_action_and_succeeds() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await @@ -390,7 +390,7 @@ async fn a_failing_step_stops_with_compensations_and_remainder() { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/checkout.rs b/crates/fleet-controller/tests/checkout.rs index a7c0492..ceda0ee 100644 --- a/crates/fleet-controller/tests/checkout.rs +++ b/crates/fleet-controller/tests/checkout.rs @@ -98,7 +98,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/frogenv.rs b/crates/fleet-controller/tests/frogenv.rs index f2dc8fb..fe804f4 100644 --- a/crates/fleet-controller/tests/frogenv.rs +++ b/crates/fleet-controller/tests/frogenv.rs @@ -101,7 +101,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/gateway.rs b/crates/fleet-controller/tests/gateway.rs index 2c7efd4..86d76ef 100644 --- a/crates/fleet-controller/tests/gateway.rs +++ b/crates/fleet-controller/tests/gateway.rs @@ -162,7 +162,7 @@ impl Harness { deadline_at, correlation_id: None, payload_json: Some(serde_json::json!({ "machineId": machine_id }).to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/machines.rs b/crates/fleet-controller/tests/machines.rs index 7c58116..a19f4cf 100644 --- a/crates/fleet-controller/tests/machines.rs +++ b/crates/fleet-controller/tests/machines.rs @@ -162,7 +162,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: Some(serde_json::json!({ "machineId": machine_id }).to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/mise.rs b/crates/fleet-controller/tests/mise.rs index 901d60f..685df09 100644 --- a/crates/fleet-controller/tests/mise.rs +++ b/crates/fleet-controller/tests/mise.rs @@ -100,7 +100,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/operation_worker.rs b/crates/fleet-controller/tests/operation_worker.rs index 58092ab..f9923e7 100644 --- a/crates/fleet-controller/tests/operation_worker.rs +++ b/crates/fleet-controller/tests/operation_worker.rs @@ -30,7 +30,7 @@ async fn the_noop_operation_runs_end_to_end() { deadline_at: None, correlation_id: Some("corr-worker-1".to_owned()), payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -73,7 +73,7 @@ async fn two_workers_cannot_claim_the_same_operation() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -112,7 +112,7 @@ async fn a_crashed_workers_lease_is_recovered_as_failed_not_retried() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -163,7 +163,7 @@ async fn a_cancelled_operation_stops_without_running() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -216,7 +216,7 @@ async fn a_deadline_expires_even_when_no_worker_claims_it() { deadline_at: Some(now - 1_000), correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -259,7 +259,7 @@ async fn restart_preserves_terminal_truth() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/proxmox.rs b/crates/fleet-controller/tests/proxmox.rs index d1b148e..f4e1b17 100644 --- a/crates/fleet-controller/tests/proxmox.rs +++ b/crates/fleet-controller/tests/proxmox.rs @@ -81,6 +81,16 @@ impl FixedTransport { #[async_trait] impl PveTransport for FixedTransport { + async fn execute_with_body( + &self, + request: PveHttpRequest, + _body: Vec, + ) -> Result { + // The canned fixtures answer by path regardless of the body: the + // recorded responses are keyed on the endpoint, not the payload. + self.execute(request).await + } + async fn execute(&self, request: PveHttpRequest) -> Result { let behavior = *self.behavior.lock().unwrap(); match (&request.pinned_fingerprint, behavior) { @@ -604,6 +614,14 @@ impl LifecycleTransport { #[async_trait] impl PveTransport for LifecycleTransport { + async fn execute_with_body( + &self, + request: PveHttpRequest, + _body: Vec, + ) -> Result { + self.execute(request).await + } + async fn execute(&self, request: PveHttpRequest) -> Result { // The observe-only trust probe: capture and refuse. if request.pinned_fingerprint.is_none() { @@ -611,11 +629,6 @@ impl PveTransport for LifecycleTransport { observed: FP.to_owned(), }); } - if request.pinned_fingerprint.is_none() { - return Err(PveTransportError::ObserveRefused { - observed: FP.to_owned(), - }); - } if request.path.ends_with("/snapshot") && request.method == fleet_provider_proxmox::PveHttpMethod::Post { diff --git a/crates/fleet-controller/tests/ready.rs b/crates/fleet-controller/tests/ready.rs index ae4f9e3..08792e1 100644 --- a/crates/fleet-controller/tests/ready.rs +++ b/crates/fleet-controller/tests/ready.rs @@ -134,7 +134,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/skills.rs b/crates/fleet-controller/tests/skills.rs index 2d9a06e..49024e9 100644 --- a/crates/fleet-controller/tests/skills.rs +++ b/crates/fleet-controller/tests/skills.rs @@ -102,7 +102,7 @@ impl Fixture { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/ssh_exec.rs b/crates/fleet-controller/tests/ssh_exec.rs index 118d829..454da71 100644 --- a/crates/fleet-controller/tests/ssh_exec.rs +++ b/crates/fleet-controller/tests/ssh_exec.rs @@ -76,7 +76,7 @@ async fn an_unverified_endpoint_refuses_to_execute() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json_string), - reviewed: false, + review_token: None, }, ) .await @@ -155,7 +155,7 @@ async fn a_verified_endpoint_runs_the_script_and_reports_output() { deadline_at: None, correlation_id: None, payload_json: Some(payload_json_string), - reviewed: false, + review_token: None, }, ) .await @@ -197,7 +197,7 @@ async fn an_unknown_kind_fails_honestly() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await @@ -219,7 +219,7 @@ async fn the_noop_kind_still_runs_through_the_composed_executor() { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleet-controller/tests/worker_lifecycle.rs b/crates/fleet-controller/tests/worker_lifecycle.rs index 7ef080c..24c9ed7 100644 --- a/crates/fleet-controller/tests/worker_lifecycle.rs +++ b/crates/fleet-controller/tests/worker_lifecycle.rs @@ -99,7 +99,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: None, - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index c9c29a3..16e324d 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -1778,7 +1778,6 @@ fn follow_review( account_id, node, vmid, - params, .. } = &invocation.command else { @@ -1789,10 +1788,9 @@ fn follow_review( message: "the review did not answer with a token".to_owned(), }); }; - let params_value: Value = params - .as_deref() - .and_then(|text| serde_json::from_str(text).ok()) - .unwrap_or(Value::Null); + // The reviewed params are the ones the controller echoed back: reusing + // them guarantees the run's bytes match the reviewed bytes. + let params_value = body["data"]["params"].clone(); let run_body = serde_json::json!({ "node": node, "reviewToken": token, @@ -1865,7 +1863,8 @@ fn follow_checkout_wait( | Command::MiseOperation { wait, timeout, .. } | Command::ProjectsReady { wait, timeout, .. } | Command::ApplyWorkflow { wait, timeout, .. } - | Command::ProxmoxLifecycle { wait, timeout, .. } => (*wait, *timeout), + | Command::ProxmoxLifecycle { wait, timeout, .. } + | Command::ProxmoxDestructive { wait, timeout, .. } => (*wait, *timeout), _ => return Ok(body), }; if !wait.0 { @@ -2577,19 +2576,25 @@ fn read_stdin_to_end(what: &str) -> Result { } fn read_stdin_line(what: &str) -> Result { - let mut line = String::new(); + read_stdin_to_eof(what) +} + +/// Reads standard input to EOF: JSON documents are multi-line. +fn read_stdin_to_eof(what: &str) -> Result { + use std::io::Read as _; + let mut text = String::new(); std::io::stdin() - .read_line(&mut line) + .read_to_string(&mut text) .map_err(|error| CliError { message: format!("cannot read {what} from stdin: {error}"), })?; - let line = line.trim().to_owned(); - if line.is_empty() { + let text = text.trim().to_owned(); + if text.is_empty() { return Err(CliError { message: format!("{what} must not be empty"), }); } - Ok(line) + Ok(text) } /// Wall-clock now, in epoch milliseconds. diff --git a/crates/fleetctl/tests/cli.rs b/crates/fleetctl/tests/cli.rs index b3c983d..7097f15 100644 --- a/crates/fleetctl/tests/cli.rs +++ b/crates/fleetctl/tests/cli.rs @@ -2206,3 +2206,65 @@ fn parsing_walks_the_proxmox_destructive_forms() { let error = fleetctl::parse(&args).unwrap_err(); assert!(error.message.contains("is required"), "{error}"); } + +#[test] +fn parsing_destructive_flags_and_vmid_validation() { + // --wait and --timeout parse for a destructive verb. + let args: Vec = [ + "proxmox", + "snapshot", + "--account", + "acc-1", + "--node", + "pve", + "--vmid", + "101", + "--wait", + "--timeout", + "120", + ] + .iter() + .map(ToString::to_string) + .collect(); + assert!(matches!( + fleetctl::parse(&args).unwrap().command, + fleetctl::Command::ProxmoxDestructive { + wait: true, + timeout: Some(120), + .. + } + )); + // A bad VMID refuses. + let args: Vec = [ + "proxmox", + "clone", + "--account", + "acc-1", + "--node", + "pve", + "--vmid", + "abc", + ] + .iter() + .map(ToString::to_string) + .collect(); + let error = fleetctl::parse(&args).unwrap_err(); + assert!(error.message.contains("must be a number"), "{error}"); + // An unknown flag refuses. + let args: Vec = [ + "proxmox", + "template", + "--account", + "acc-1", + "--node", + "pve", + "--vmid", + "101", + "--bogus", + ] + .iter() + .map(ToString::to_string) + .collect(); + let error = fleetctl::parse(&args).unwrap_err(); + assert!(error.message.contains("unknown flag"), "{error}"); +} diff --git a/crates/fleetd/tests/node_install.rs b/crates/fleetd/tests/node_install.rs index 0006a93..f110c0e 100644 --- a/crates/fleetd/tests/node_install.rs +++ b/crates/fleetd/tests/node_install.rs @@ -306,7 +306,7 @@ impl Harness { deadline_at: None, correlation_id: None, payload_json: Some(payload.to_string()), - reviewed: false, + review_token: None, }, ) .await diff --git a/crates/providers/fleet-provider-proxmox/src/lib.rs b/crates/providers/fleet-provider-proxmox/src/lib.rs index a5c3324..ccdf23d 100644 --- a/crates/providers/fleet-provider-proxmox/src/lib.rs +++ b/crates/providers/fleet-provider-proxmox/src/lib.rs @@ -192,10 +192,7 @@ pub trait PveTransport: fmt::Debug + Send + Sync { &self, request: PveHttpRequest, body: Vec, - ) -> Result { - let _ = body; - self.execute(request).await - } + ) -> Result; } /// The reqwest-backed transport: rustls with the pinned-fingerprint @@ -1063,8 +1060,20 @@ fn push_bounded(body: &mut Vec, chunk: &[u8]) -> Result<(), PveTransportErro /// The optional UPID string a mutating endpoint answered; a synchronous /// outcome carries no UPID. -fn upid_from_data(data: &serde_json::Value) -> Option { - data.as_str().and_then(|raw| Upid::parse(raw).ok()) +fn upid_from_data(data: &serde_json::Value) -> Result, PveApiError> { + match data { + // A synchronous outcome carries `null` or no UPID at all. + serde_json::Value::Null => Ok(None), + serde_json::Value::String(raw) => Upid::parse(raw) + .map(Some) + .map_err(|detail| PveApiError::InvalidPayload { detail }), + other => Err(PveApiError::InvalidPayload { + detail: format!( + "the mutating answer is neither a UPID string nor null (it is a {})", + type_name_of(other) + ), + }), + } } /// A bounded, credential-free body excerpt for error details. @@ -1247,11 +1256,11 @@ impl ProxmoxSource for ProxmoxClient { let data = self .call_with_body( request, - &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot"), + &format!("/api2/json/nodes/{}/qemu/{vmid}/snapshot", urlencode(node)), &body, ) .await?; - Ok(upid_from_data(&data)) + upid_from_data(&data) } async fn guest_snapshot_rollback( @@ -1264,11 +1273,15 @@ impl ProxmoxSource for ProxmoxClient { let data = self .call_with_body( request, - &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot/{snapshot}/rollback"), + &format!( + "/api2/json/nodes/{}/qemu/{vmid}/snapshot/{}/rollback", + urlencode(node), + urlencode(snapshot) + ), &serde_json::json!({}), ) .await?; - Ok(upid_from_data(&data)) + upid_from_data(&data) } async fn guest_snapshot_delete( @@ -1280,7 +1293,11 @@ impl ProxmoxSource for ProxmoxClient { ) -> Result<(), PveApiError> { self.call_delete( request, - &format!("/api2/json/nodes/{node}/qemu/{vmid}/snapshot/{snapshot}"), + &format!( + "/api2/json/nodes/{}/qemu/{vmid}/snapshot/{}", + urlencode(node), + urlencode(snapshot) + ), ) .await } @@ -1302,7 +1319,7 @@ impl ProxmoxSource for ProxmoxClient { let data = self .call_with_body( request, - &format!("/api2/json/nodes/{node}/qemu/{vmid}/clone"), + &format!("/api2/json/nodes/{}/qemu/{vmid}/clone", urlencode(node)), &body, ) .await?; @@ -1323,18 +1340,20 @@ impl ProxmoxSource for ProxmoxClient { let data = self .call_with_body( request, - &format!("/api2/json/nodes/{node}/qemu/{vmid}/template"), + &format!("/api2/json/nodes/{}/qemu/{vmid}/template", urlencode(node)), &serde_json::json!({}), ) .await?; - Ok(upid_from_data(&data)) + upid_from_data(&data) } async fn stop_task(&self, request: PveHttpRequest, upid: &Upid) -> Result<(), PveApiError> { self.call_delete( request, + // PVE's stop-task endpoint is the task itself, not its status + // subresource. &format!( - "/api2/json/nodes/{}/tasks/{}/status", + "/api2/json/nodes/{}/tasks/{}", urlencode(&upid.node), urlencode(&upid.raw) ), @@ -1546,7 +1565,7 @@ impl ProxmoxClient { /// # Errors /// /// Fails with [`PveApiError`]. - pub async fn list_qemu_resources( + pub async fn list_guest_resources( &self, request: PveHttpRequest, ) -> Result, PveApiError> { @@ -1560,7 +1579,7 @@ impl ProxmoxClient { } Ok(resources .into_iter() - .filter(|resource| resource.kind == "qemu" || resource.kind == "qemu-template") + .filter(|resource| matches!(resource.kind.as_str(), "qemu" | "qemu-template" | "lxc")) .collect()) } From a2e1b0d4d82a1d2984f17de83f84ad1e9cd3746c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Mon, 21 Sep 2026 11:51:12 +0000 Subject: [PATCH 3/4] =?UTF-8?q?FM-603:=20restore=20the=20single-line=20sec?= =?UTF-8?q?ret=20reader=20=E2=80=94=20only=20document=20inputs=20read=20st?= =?UTF-8?q?din=20to=20EOF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fleetctl/src/lib.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index 16e324d..1787e0f 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -2576,10 +2576,24 @@ fn read_stdin_to_end(what: &str) -> Result { } fn read_stdin_line(what: &str) -> Result { - read_stdin_to_eof(what) + let mut line = String::new(); + std::io::stdin() + .read_line(&mut line) + .map_err(|error| CliError { + message: format!("cannot read {what} from stdin: {error}"), + })?; + let line = line.trim().to_owned(); + if line.is_empty() { + return Err(CliError { + message: format!("{what} must not be empty"), + }); + } + Ok(line) } -/// Reads standard input to EOF: JSON documents are multi-line. +/// Reads standard input to EOF: JSON documents are multi-line. Only the +/// document inputs use this; single-line secrets keep the newline- +/// terminating reader. fn read_stdin_to_eof(what: &str) -> Result { use std::io::Read as _; let mut text = String::new(); From 07bae1e3ce7173768ed75e15f79ff8847b3eedb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Mon, 21 Sep 2026 11:54:58 +0000 Subject: [PATCH 4/4] FM-603: route the destructive params reader through the EOF reader --- crates/fleetctl/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index 1787e0f..4e722e5 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -2450,7 +2450,7 @@ fn request_for(command: &Command) -> Result { // in argv. let text = match params.as_deref() { Some(text) => text.to_owned(), - None => read_stdin_line("the action parameters as JSON")?, + None => read_stdin_to_eof("the action parameters as JSON")?, }; let params_value: serde_json::Value = serde_json::from_str(&text).map_err(|error| CliError {