diff --git a/crates/fleet-api/src/lib.rs b/crates/fleet-api/src/lib.rs index 947b10d..831a80a 100644 --- a/crates/fleet-api/src/lib.rs +++ b/crates/fleet-api/src/lib.rs @@ -129,6 +129,7 @@ pub const API_BASE_PATH: &str = "/api/v1"; proxmox::AssociatedGuestDto, proxmox::AssociationCandidateDto, proxmox::ObserveProxmoxGuestRequest, + proxmox::StartProxmoxLifecycleRequest, proxmox::ProviderAgentDto, proxmox::ProviderInterfaceDto, node::CreateEnrollmentTokenRequest, @@ -234,6 +235,7 @@ pub fn api(state: Arc) -> (Router, utoipa::openapi::OpenAp .routes(routes!(proxmox::discover_proxmox_cluster)) .routes(routes!(proxmox::list_proxmox_guests)) .routes(routes!(proxmox::observe_proxmox_guest)) + .routes(routes!(proxmox::start_proxmox_lifecycle)) .with_state(state), ) .split_for_parts(); diff --git a/crates/fleet-api/src/proxmox.rs b/crates/fleet-api/src/proxmox.rs index 5582f22..83faa37 100644 --- a/crates/fleet-api/src/proxmox.rs +++ b/crates/fleet-api/src/proxmox.rs @@ -963,3 +963,139 @@ pub async fn observe_proxmox_guest( .map_err(|error| map_proxmox_error(&error, correlation_id))?; Ok(StatusCode::NO_CONTENT) } + +/// The lifecycle request: the guest's node and VMID. +#[derive(Debug, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct StartProxmoxLifecycleRequest { + /// 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, +} + +/// Runs a lifecycle action on one guest as a durable operation. The +/// operation is authorized through the catalog's `proxmox.operate` +/// (catalog-level, like the source kinds — a Proxmox guest is not a Fleet +/// machine) and executed by the worker with Fleet-owned UPID polling. +/// +/// # Errors +/// +/// Returns the public error envelope on refusal or a malformed request. +#[utoipa::path( + post, + path = "/proxmox/accounts/{accountId}/guests/{vmid}/{action}", + tag = "proxmox", + operation_id = "startProxmoxLifecycle", + params( + ( + "accountId" = String, + Path, + description = "The account's identity." + ), + ( + "vmid" = u32, + Path, + description = "The guest's VMID." + ), + ( + "action" = String, + Path, + description = "The lifecycle action: start, stop, shutdown, or reboot." + ), + ), + request_body = StartProxmoxLifecycleRequest, + responses( + ( + status = 202, + description = "The lifecycle operation was accepted and is durable.", + 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.", + body = crate::error::ApiError + ), + ) +)] +pub async fn start_proxmox_lifecycle( + 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)?; + // The body's VMID must agree with the path's: two names for one guest + // is a malformed request, not a fallback. + if request.vmid != vmid { + return Err(crate::machines::invalid_request( + "the body's vmid does not match the path's guest", + correlation_id, + )); + } + // The action is validated here so a malformed path is a 400, not an + // operation that fails later in the worker. The stable ids match the + // provider's `LifecycleAction` vocabulary; the executor re-validates. + if !matches!(action.as_str(), "start" | "stop" | "shutdown" | "reboot") { + return Err(crate::machines::invalid_request( + &format!("unrecognized lifecycle action {action:?}"), + correlation_id, + )); + } + // The executor re-applies the same gate at the account boundary; the + // catalog-level permission is checked here so a denial never creates + // an operation. + 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::ProxmoxOperate, + resource: None, + }, + ) { + return Err(crate::machines::denied_error(decision, correlation_id)); + } + let payload = serde_json::json!({ + "accountId": account_id, + "node": request.node, + "vmid": vmid, + "timeoutSeconds": request.timeout_seconds, + }); + let kind = format!("proxmox.guest.{action}"); + // A caller-scoped idempotency key makes a retried POST return the + // original operation instead of a second one. + 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()), + }, + ) + .await + .map_err(|error| crate::operations::map_use_case_error(&error, correlation_id))?; + Ok(( + StatusCode::ACCEPTED, + Json(Resource::new(crate::operations::OperationDto::from( + operation, + ))), + )) +} diff --git a/crates/fleet-application/src/authz.rs b/crates/fleet-application/src/authz.rs index b03ada6..8651b8e 100644 --- a/crates/fleet-application/src/authz.rs +++ b/crates/fleet-application/src/authz.rs @@ -136,6 +136,9 @@ pub enum Permission { /// Create, confirm trust for, or remove a Proxmox account. A /// mutation: it stores or removes a credential or a trust anchor. ProxmoxConfig, + /// Run a lifecycle action (start/stop/shutdown/reboot) on a Proxmox + /// guest. A mutation: it changes the guest's power state. + ProxmoxOperate, } impl Permission { @@ -181,6 +184,7 @@ impl Permission { Permission::SourceActivate, Permission::ProxmoxRead, Permission::ProxmoxConfig, + Permission::ProxmoxOperate, ]; /// The stable action id, as recorded in decisions and audit events. @@ -225,6 +229,7 @@ impl Permission { Permission::SourceActivate => "source.activate", Permission::ProxmoxRead => "proxmox.read", Permission::ProxmoxConfig => "proxmox.config", + Permission::ProxmoxOperate => "proxmox.operate", } } @@ -271,7 +276,8 @@ impl Permission { | Permission::SourceFetch | Permission::SourceActivate | Permission::ProxmoxRead - | Permission::ProxmoxConfig => true, + | Permission::ProxmoxConfig + | Permission::ProxmoxOperate => true, } } @@ -296,7 +302,8 @@ impl Permission { | Permission::SourceFetch | Permission::SourceActivate | Permission::ProxmoxRead - | Permission::ProxmoxConfig => false, + | Permission::ProxmoxConfig + | Permission::ProxmoxOperate => false, Permission::MachineReadSensitive | Permission::OperationCancel | Permission::SecretRead diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index 5fea37d..edf105a 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; 31] = [ +pub const CREATABLE_KINDS: [&str; 35] = [ "noop", "ssh.exec", "agentless.inventory", @@ -75,6 +75,10 @@ pub const CREATABLE_KINDS: [&str; 31] = [ "apply.workflow", "source.fetch", "source.activate", + "proxmox.guest.start", + "proxmox.guest.stop", + "proxmox.guest.shutdown", + "proxmox.guest.reboot", ]; /// The machine-scoped permission a kind's creation requires, when any. @@ -83,12 +87,23 @@ pub const CREATABLE_KINDS: [&str; 31] = [ /// governs both the dedicated endpoint and the generic one. #[must_use] fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option { - // The source kinds are catalog-level: their permission is enforced - // here with `resource: None`, never a machine id. + machine_scoped_kind_permission_inner(kind, payload) +} + +/// The catalog-level permission a kind's creation requires, when any. +/// The source and Proxmox lifecycle kinds act on infrastructure that is +/// not a Fleet machine, so their permission is enforced with +/// `resource: None` — never a machine id. +#[must_use] +fn catalog_scoped_kind_permission(kind: &str) -> Option { match kind { "source.fetch" => Some(Permission::SourceFetch), "source.activate" => Some(Permission::SourceActivate), - _ => machine_scoped_kind_permission_inner(kind, payload), + "proxmox.guest.start" + | "proxmox.guest.stop" + | "proxmox.guest.shutdown" + | "proxmox.guest.reboot" => Some(Permission::ProxmoxOperate), + _ => None, } } @@ -509,6 +524,16 @@ impl Operations { }, ) .map_err(OperationUseCaseError::Denied)?; + } else if let Some(permission) = catalog_scoped_kind_permission(&new.kind) { + authorize( + authorizer, + AccessRequest { + principal_id, + action: permission, + resource: None, + }, + ) + .map_err(OperationUseCaseError::Denied)?; } let operation = self .port @@ -634,6 +659,19 @@ impl Operations { Ok(operation.state) } + /// Whether cancellation has been requested for the operation. The + /// worker reads this between poll cycles; it bypasses authorization + /// like the other worker-side reads, because the worker already owns + /// the claimed operation. + /// + /// # Errors + /// + /// Fails when the operation is unknown or the backend errors. + pub async fn cancel_requested(&self, id: &str) -> Result { + let operation = self.port.get(id).await?; + Ok(operation.cancel_requested) + } + /// Reads one operation. /// /// # Errors diff --git a/crates/fleet-auth/tests/authz_adapter.rs b/crates/fleet-auth/tests/authz_adapter.rs index d19c3b9..0b6bedb 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(), 38); + assert_eq!(Permission::ALL.len(), 39); } #[test] diff --git a/crates/fleet-controller/src/lib.rs b/crates/fleet-controller/src/lib.rs index 3b0689d..23130b2 100644 --- a/crates/fleet-controller/src/lib.rs +++ b/crates/fleet-controller/src/lib.rs @@ -20,6 +20,7 @@ pub mod install; pub mod mise; pub mod node_crypto; pub mod onboard; +pub mod proxmox_exec; pub mod proxmox_store; pub mod ready; pub mod skills; diff --git a/crates/fleet-controller/src/main.rs b/crates/fleet-controller/src/main.rs index 8e91fc2..7f16db4 100644 --- a/crates/fleet-controller/src/main.rs +++ b/crates/fleet-controller/src/main.rs @@ -285,6 +285,41 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { )), )) }; + // The Proxmox lifecycle executor handles the FM-602 kinds over + // the same accounts and secret store the discovery surfaces + // use; it composes after the source dispatch so its kinds + // reach it and everything else falls through. + let with_proxmox: std::sync::Arc = { + let proxmox_client = fleet_provider_proxmox::ProxmoxClient::new( + std::sync::Arc::new(fleet_provider_proxmox::ReqwestPveTransport::new()), + ); + let accounts: std::sync::Arc = + std::sync::Arc::new(fleet_storage_sqlite::ProxmoxAccountRepository::new( + store.pool().clone(), + )); + let credentials: std::sync::Arc< + dyn fleet_application::proxmox::ProxmoxCredentialStore, + > = match &secrets { + Some(secrets) => std::sync::Arc::new( + fleet_controller::proxmox_store::SecretBackedProxmoxCredentials::new( + secrets.clone(), + ), + ), + None => std::sync::Arc::new( + fleet_controller::proxmox_store::AbsentProxmoxCredentials, + ), + }; + std::sync::Arc::new(fleet_controller::proxmox_exec::ProxmoxDispatch::new( + with_source.clone(), + std::sync::Arc::new( + fleet_controller::proxmox_exec::ProxmoxLifecycleExecutor::new( + accounts, + credentials, + proxmox_client, + ), + ), + )) + }; match &services { Some(services) => { let node_machines: std::sync::Arc = @@ -295,11 +330,11 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { std::sync::Arc::new(fleet_controller::gateway::NodeCommandExecutor::new( services.gateway.clone(), node_machines, - with_source.clone(), + with_proxmox.clone(), )); executor } - None => with_source.clone(), + None => with_proxmox.clone(), } }; let worker_host = WorkerHost::new(worker_operations, executor, 4); diff --git a/crates/fleet-controller/src/proxmox_exec.rs b/crates/fleet-controller/src/proxmox_exec.rs new file mode 100644 index 0000000..f3199ee --- /dev/null +++ b/crates/fleet-controller/src/proxmox_exec.rs @@ -0,0 +1,363 @@ +//! The Proxmox lifecycle executor: start/stop/shutdown/reboot as durable +//! operations with Fleet-owned UPID polling (FM-602). +//! +//! The executor resolves the account through the same composition the +//! discovery surfaces use, runs the lifecycle action, and polls the task's +//! status on a fixed interval against a deadline. The polling loop is a +//! sleep between polls, never a wall-clock race — the legacy `waitForTask` +//! flake is exactly what this avoids. Terminal states are honest: task +//! `OK` succeeds, `ERROR` fails with the bounded detail, a deadline expiry +//! fails with the last observed status, and an unreadable status fails +//! naming the uncertainty — never assumed success. +//! +//! Cancellation stops the *waiting*, not the remote task: PVE keeps +//! running the action, and the operation records that honestly. Remote +//! task cancellation is a destructive-adjacent action deferred to epic +//! #12's review. + +use std::sync::Arc; +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 serde::Deserialize; + +/// The interval between task-status polls. Fixed, not clock-derived. +const POLL_INTERVAL: Duration = Duration::from_secs(2); +/// The maximum lifecycle timeout the executor accepts from a payload. +pub const MAX_LIFECYCLE_TIMEOUT: u64 = 600; + +/// The payload every lifecycle kind carries: the machine-scoped shape plus +/// the account, the guest, and its node. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LifecyclePayload { + /// 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 kind-dispatching Proxmox lifecycle executor. +#[derive(Debug)] +pub struct ProxmoxLifecycleExecutor { + accounts: Arc, + credentials: Arc, + client: fleet_provider_proxmox::ProxmoxClient, +} + +impl ProxmoxLifecycleExecutor { + /// 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, + } + } + + /// The trusted account and its resolved secret: the same explicit-trust + /// gate the read surfaces apply, without duplicating their checks. + 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, + } + } + + /// Runs one action and polls its task to a terminal state, with the + /// injected sleep keeping the loop deterministic under test. The + /// deadline starts before the mutation and bounds every poll; a + /// cancellation observed mid-poll stops the *waiting* — the remote PVE + /// task keeps running, and the operation records that honestly. + async fn run_action( + &self, + operations: &Operations, + params: RunParams, + ) -> Result { + let RunParams { + operation_id, + request, + node, + vmid, + action, + deadline, + sleep, + } = params; + let started = std::time::Instant::now(); + let upid = self + .client + .guest_lifecycle(request.clone(), &node, vmid, action) + .await + .map_err(|error| format!("the lifecycle action failed: {error}"))?; + loop { + // Cancellation is honored between polls: the remote task keeps + // running, and the operation says so. + let cancelled = operations + .cancel_requested(&operation_id) + .await + .unwrap_or(false); + if cancelled { + return Ok(TaskStatus::Error { + detail: format!( + "cancelled while waiting; the remote task on node {} keeps running and its outcome is unknown", + upid.node + ), + }); + } + 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 Ok(terminal), + } + if started.elapsed() >= deadline { + // The deadline expired while the task still runs: the last + // observed status is the honest terminal, and it names the + // uncertainty. + return Ok(TaskStatus::Error { + detail: format!( + "the deadline expired while the task still runs; its final state is unknown (task on node {})", + upid.node + ), + }); + } + sleep(POLL_INTERVAL).await; + } + } + + /// Completes the operation from the terminal task status. + async fn finish( + &self, + operations: &Operations, + operation_id: &str, + action: LifecycleAction, + status: TaskStatus, + ) -> Result<(), String> { + match status { + TaskStatus::Ok => operations + .complete( + operation_id, + "succeeded", + Some( + &serde_json::json!({ + "action": action.id(), + "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 } => { + let error_json = + serde_json::json!({ "reason": "task_failed", "detail": detail }).to_string(); + operations + .complete(operation_id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + TaskStatus::Unknown => { + let error_json = serde_json::json!({ + "reason": "task_unknown", + "detail": "the task's status could not be read; its outcome is unknown, not assumed" + }) + .to_string(); + operations + .complete(operation_id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + } + } + + async fn lifecycle( + &self, + operations: &Operations, + operation: &Operation, + action: LifecycleAction, + ) -> Result<(), String> { + let payload: LifecyclePayload = payload(operation)?; + if payload.node.is_empty() || payload.node.len() > 128 { + return Err("the node must be 1..=128 characters".to_owned()); + } + let (account, secret) = self.bound(&payload.account_id).await?; + let request = self.request(&account, &secret); + operations + .record_progress( + &operation.id, + Some(0), + Some(1), + Some(&format!("{} on qemu/{}", action.id(), payload.vmid)), + ) + .await + .map_err(|error| error.to_string())?; + let deadline = Duration::from_secs(payload.timeout_seconds.min(MAX_LIFECYCLE_TIMEOUT)); + let status = self + .run_action( + operations, + RunParams { + operation_id: operation.id.clone(), + request, + node: payload.node, + vmid: payload.vmid, + action, + deadline, + sleep: Arc::new(|duration| { + Box::pin(tokio::time::sleep(duration)) + as futures_util::future::BoxFuture<'static, ()> + }), + }, + ) + .await?; + self.finish(operations, &operation.id, action, status).await + } +} + +#[async_trait::async_trait] +impl OperationExecutor for ProxmoxLifecycleExecutor { + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + let action = match operation.kind.as_str() { + "proxmox.guest.start" => LifecycleAction::Start, + "proxmox.guest.stop" => LifecycleAction::Stop, + "proxmox.guest.shutdown" => LifecycleAction::Shutdown, + "proxmox.guest.reboot" => LifecycleAction::Reboot, + _ => return Err("not a Proxmox lifecycle kind".to_owned()), + }; + self.lifecycle(operations, operation, action).await + } +} + +/// The parameters of one lifecycle run, grouped so the poll loop's +/// signature stays readable. +struct RunParams { + /// The operation whose cancellation is observed between polls. + operation_id: String, + /// The provider request carrying the account and pin. + request: fleet_provider_proxmox::PveHttpRequest, + /// The guest's hosting node. + node: String, + /// The guest's VMID. + vmid: u32, + /// The action to run. + action: LifecycleAction, + /// The polling deadline. + deadline: Duration, + /// The injected sleep, for deterministic tests. + sleep: Arc futures_util::future::BoxFuture<'static, ()> + Send + Sync>, +} + +/// Decodes and validates an operation's payload. +fn payload(operation: &Operation) -> Result { + serde_json::from_str( + operation + .payload_json + .as_deref() + .ok_or("the operation carries no payload")?, + ) + .map_err(|error| format!("the payload is not a valid lifecycle record: {error}")) +} + +/// The kind-dispatching Proxmox executor: lifecycle kinds route to the +/// lifecycle executor, everything else falls through to the next executor +/// in the chain. +#[derive(Debug)] +pub struct ProxmoxDispatch { + fallback: Arc, + lifecycle: Arc, +} + +impl ProxmoxDispatch { + /// Composes the dispatch from its parts. + #[must_use] + pub fn new( + fallback: Arc, + lifecycle: Arc, + ) -> Self { + Self { + fallback, + lifecycle, + } + } +} + +#[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.") { + self.lifecycle.execute(operations, operation).await + } else { + self.fallback.execute(operations, operation).await + } + } +} diff --git a/crates/fleet-controller/src/proxmox_store.rs b/crates/fleet-controller/src/proxmox_store.rs index 79e8d3f..9db6375 100644 --- a/crates/fleet-controller/src/proxmox_store.rs +++ b/crates/fleet-controller/src/proxmox_store.rs @@ -186,6 +186,7 @@ impl ProxmoxTrustProbe for ProviderTrustProbe { token_id: "observe-only".to_owned(), token: SensitiveString::new("observe-only"), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; match self.transport.execute(request).await { Err(fleet_provider_proxmox::PveTransportError::ObserveRefused { observed }) => { @@ -248,6 +249,7 @@ impl ProxmoxDiscoverPort for ProviderDiscovery { token_id: account.token_id.clone(), token: SensitiveString::new(secret.expose().to_owned()), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; match self.client.discover(request).await { Ok(discovery) => Ok(RawDiscovery { @@ -300,6 +302,7 @@ impl fleet_application::proxmox::ProxmoxGuestDiscoverPort for ProviderDiscovery token_id: account.token_id.clone(), token: SensitiveString::new(secret.expose().to_owned()), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; match self.client.guest_discover(request).await { Ok(discovery) => Ok(fleet_application::proxmox::RawGuestDiscovery { @@ -414,3 +417,36 @@ fn machines_for( audit, )) } + +/// The credential store answering nothing, composed when the controller +/// runs without a secret store: lifecycle operations then fail honestly +/// instead of the composition panicking. +#[derive(Debug)] +pub struct AbsentProxmoxCredentials; + +#[async_trait] +impl ProxmoxCredentialStore for AbsentProxmoxCredentials { + async fn load( + &self, + _account_id: &str, + ) -> Result, fleet_application::proxmox::CredentialStoreError> { + Ok(None) + } + + async fn store( + &self, + _account_id: &str, + _secret: &str, + ) -> Result<(), fleet_application::proxmox::CredentialStoreError> { + Err(fleet_application::proxmox::CredentialStoreError::Backend { + detail: "the controller runs without a secret store".to_owned(), + }) + } + + async fn clear( + &self, + _account_id: &str, + ) -> Result<(), fleet_application::proxmox::CredentialStoreError> { + Ok(()) + } +} diff --git a/crates/fleet-controller/tests/proxmox.rs b/crates/fleet-controller/tests/proxmox.rs index 02db30a..592829b 100644 --- a/crates/fleet-controller/tests/proxmox.rs +++ b/crates/fleet-controller/tests/proxmox.rs @@ -128,6 +128,8 @@ struct Harness { pool: sqlx::SqlitePool, address: std::net::SocketAddr, shutdown: Option>, + /// The lifecycle harness's worker; aborted on drop. + _worker: Option>, } async fn harness_with(transport: Arc) -> Harness { @@ -192,6 +194,7 @@ async fn harness_with(transport: Arc) -> Harness { pool: store.pool().clone(), address, shutdown: Some(shutdown_tx), + _worker: None, } } @@ -562,3 +565,337 @@ async fn the_guest_surface_walks_list_and_observe() { assert_eq!(status, axum::http::StatusCode::NOT_FOUND, "{body}"); assert_eq!(body["code"], "not_found"); } + +const LIFECYCLE_UPID_BODY: &str = + r#"{"data":"UPID:pve:0015523F:0C6DF532:6AAFE1EC:qmstart:101:root@pam!GLM-AGENT:"}"#; + +const TASK_RUNNING_BODY: &str = r#"{"data":{"status":"running"}}"#; + +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"}}"#; + +/// What the lifecycle fixture answers per poll. +#[derive(Debug, Clone, Copy)] +enum TaskOutcome { + OkAfterOne, + Error, + AlwaysRunning, +} + +#[derive(Debug)] +struct LifecycleTransport { + outcome: TaskOutcome, +} + +impl LifecycleTransport { + fn with(outcome: TaskOutcome) -> Arc { + Arc::new(Self { outcome }) + } +} + +#[async_trait] +impl PveTransport for LifecycleTransport { + async fn execute(&self, request: PveHttpRequest) -> Result { + // The observe-only trust probe: capture and refuse. + if request.pinned_fingerprint.is_none() { + return Err(PveTransportError::ObserveRefused { + observed: FP.to_owned(), + }); + } + if request.path.contains("/status/start") { + return Ok(PveHttpResponse { + status: 200, + body: LIFECYCLE_UPID_BODY.as_bytes().to_vec(), + }); + } + if request.path.contains("/tasks/") && request.path.contains("/status") { + let body = match self.outcome { + TaskOutcome::OkAfterOne => TASK_OK_BODY, + TaskOutcome::Error => TASK_ERROR_BODY, + TaskOutcome::AlwaysRunning => TASK_RUNNING_BODY, + }; + return Ok(PveHttpResponse { + status: 200, + body: body.as_bytes().to_vec(), + }); + } + if request.path.contains("/version") { + return Ok(PveHttpResponse { + status: 200, + body: VERSION_BODY.as_bytes().to_vec(), + }); + } + Err(PveTransportError::Connect { + detail: format!("unexpected path {}", request.path), + }) + } +} + +async fn lifecycle_harness(outcome: TaskOutcome) -> Harness { + let dist = tempfile::tempdir().unwrap(); + std::fs::write(dist.path().join("index.html"), "fleet").unwrap(); + let store_dir = tempfile::tempdir().unwrap(); + let store = Store::open(&store_dir.path().join("fleet.db")) + .await + .unwrap(); + let key_dir = tempfile::tempdir().unwrap(); + let key_path = key_dir.path().join("master.key"); + std::fs::write( + &key_path, + "1 0a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20212223242526272829\n", + ) + .unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).unwrap(); + } + let secrets = Arc::new(SecretStore::open(store.pool().clone(), &key_path).unwrap()); + // One shared transport: the HTTP surface and the worker see the same + // fixture, so the test exercises one consistent PVE. + let transport = LifecycleTransport::with(outcome); + let proxmox = Arc::new(compose_proxmox( + store.pool().clone(), + secrets.clone(), + transport.clone(), + Arc::new(fleet_storage_sqlite::AuditSink::new(store.pool().clone())), + )); + + // The worker runs the lifecycle executor against the same transport, + // so the durable operation executes end to end in the test. + let worker_operations = Arc::new(fleet_application::operation::Operations::new( + Arc::new(fleet_storage_sqlite::OperationRepository::new( + store.pool().clone(), + )), + Arc::new(fleet_storage_sqlite::AuditSink::new(store.pool().clone())), + )); + let accounts: Arc = Arc::new( + fleet_storage_sqlite::ProxmoxAccountRepository::new(store.pool().clone()), + ); + let credentials: Arc = + Arc::new(fleet_controller::proxmox_store::SecretBackedProxmoxCredentials::new(secrets)); + let executor = Arc::new(fleet_controller::proxmox_exec::ProxmoxDispatch::new( + Arc::new(fleet_application::worker::NoopExecutor), + Arc::new( + fleet_controller::proxmox_exec::ProxmoxLifecycleExecutor::new( + accounts, + credentials, + fleet_provider_proxmox::ProxmoxClient::new(transport), + ), + ), + )); + let worker_host = fleet_controller::worker::WorkerHost::new(worker_operations, executor, 4); + // The shutdown future must never resolve while the test runs: an + // immediately-ready `async {}` would drain the worker on the first + // tick. + let _worker_handle = tokio::spawn(async move { + worker_host.run(std::future::pending::<()>()).await; + }); + + let settings = Settings { + listen: "127.0.0.1:0".parse().unwrap(), + web_dist: dist.path().to_path_buf(), + artifacts_dir: None, + }; + let router = build_router( + &settings, + Some(store.pool().clone()), + None, + None, + None, + None, + Some(&proxmox), + ); + let listener = TokioListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + tokio::spawn(async move { + axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + }) + .await + .expect("the test server must serve"); + }); + Harness { + _dist: dist, + _store_dir: store_dir, + _key_dir: key_dir, + pool: store.pool().clone(), + address, + shutdown: Some(shutdown_tx), + _worker: None, + } +} + +/// The setup shared by the lifecycle tests: a trusted account. +async fn trusted_account(harness: &Harness) -> String { + let (_, body) = harness + .post( + "/api/v1/proxmox/accounts", + json!({ + "name": "pve-main", + "host": "192.168.68.223", + "tokenId": "root@pam!GLM-AGENT", + "tokenSecret": "the-token-secret-material" + }), + ) + .await; + let account_id = body["data"]["id"].as_str().unwrap().to_owned(); + let (status, _) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/observe"), + json!({}), + ) + .await; + assert_eq!(status, axum::http::StatusCode::OK, "observe must answer"); + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/confirm"), + json!({"fingerprint": FP}), + ) + .await; + assert_eq!( + status, + axum::http::StatusCode::OK, + "confirm must answer: {body}" + ); + account_id +} + +#[tokio::test] +async fn a_lifecycle_operation_runs_to_task_ok() { + let harness = lifecycle_harness(TaskOutcome::OkAfterOne).await; + let account_id = trusted_account(&harness).await; + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/start"), + json!({"node": "pve", "vmid": 101, "timeoutSeconds": 30}), + ) + .await; + assert_eq!(status, axum::http::StatusCode::ACCEPTED, "{body}"); + let operation_id = body["data"]["id"].as_str().unwrap().to_owned(); + eprintln!("created operation {operation_id}"); + + // The worker picks it up; wait for the terminal state. + 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" { + eprintln!("terminal {terminal}: {body}"); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert_eq!(terminal, "succeeded", "terminal={terminal} body={body}"); + let (_, body) = harness + .get(&format!("/api/v1/operations/{operation_id}")) + .await; + let result: Value = serde_json::from_str(body["data"]["resultJson"].as_str().unwrap()) + .expect("the result is JSON"); + assert_eq!(result["taskState"], "ok", "{body}"); +} + +#[tokio::test] +async fn a_failing_task_fails_the_operation_with_the_detail() { + let harness = lifecycle_harness(TaskOutcome::Error).await; + let account_id = trusted_account(&harness).await; + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/start"), + json!({"node": "pve", "vmid": 101, "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(); + let mut error_detail = 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" { + // The error detail rides the operation's errorJson string. + if let Some(error_json) = body["data"]["errorJson"].as_str() { + let parsed: Value = serde_json::from_str(error_json).unwrap_or(Value::Null); + error_detail = parsed["detail"].as_str().unwrap_or_default().to_owned(); + } + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert_eq!(terminal, "failed"); + assert!( + error_detail.contains("KVM is not available"), + "{error_detail}" + ); +} + +#[tokio::test] +async fn a_deadline_expiry_fails_honestly_naming_the_uncertainty() { + let harness = lifecycle_harness(TaskOutcome::AlwaysRunning).await; + let account_id = trusted_account(&harness).await; + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/start"), + // A 1-second deadline: the poll loop expires while the task + // still runs. + json!({"node": "pve", "vmid": 101, "timeoutSeconds": 1}), + ) + .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(); + let mut error_detail = String::new(); + for _ in 0..100 { + 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" { + if let Some(error_json) = body["data"]["errorJson"].as_str() { + let parsed: Value = serde_json::from_str(error_json).unwrap_or(Value::Null); + error_detail = parsed["detail"].as_str().unwrap_or_default().to_owned(); + } + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert_eq!(terminal, "failed"); + assert!( + error_detail.contains("final state is unknown"), + "{error_detail}" + ); +} + +#[tokio::test] +async fn an_unrecognized_action_refuses_with_invalid_request() { + let harness = lifecycle_harness(TaskOutcome::OkAfterOne).await; + let account_id = trusted_account(&harness).await; + let (status, body) = harness + .post( + &format!("/api/v1/proxmox/accounts/{account_id}/guests/101/destroy"), + json!({"node": "pve", "vmid": 101, "timeoutSeconds": 30}), + ) + .await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["code"], "invalid_request"); +} diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index 19e91a1..2531a94 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -542,6 +542,21 @@ pub enum Command { /// The machine the guest is confirmed to be. machine_id: String, }, + /// Run a lifecycle action on a guest as a durable operation. + ProxmoxLifecycle { + /// The action: start, stop, shutdown, or reboot. + action: String, + /// The account's identity. + account_id: String, + /// The guest's hosting node. + node: String, + /// The guest's VMID. + vmid: u32, + /// Wait for the operation to finish. + wait: bool, + /// How long to wait, in seconds. + timeout: Option, + }, /// Start the audited "Install Fleet Node" bootstrap on an agentless /// machine: download the checksummed service package on the node, /// install the systemd service, enroll, and wait for the gateway @@ -1208,6 +1223,77 @@ fn parse_proxmox_command(verb: &str, rest: &[&str]) -> Result } _ => Err(CliError { message: usage() }), }, + "start" | "stop" | "shutdown" | "reboot" => { + 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() + ), + }); + } + } + } + Ok(Command::ProxmoxLifecycle { + 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(), + })?, + wait, + timeout, + }) + } _ => Err(CliError { message: usage() }), } } @@ -1638,7 +1724,8 @@ fn follow_checkout_wait( | Command::FrogenvOperation { wait, timeout, .. } | Command::MiseOperation { wait, timeout, .. } | Command::ProjectsReady { wait, timeout, .. } - | Command::ApplyWorkflow { wait, timeout, .. } => (*wait, *timeout), + | Command::ApplyWorkflow { wait, timeout, .. } + | Command::ProxmoxLifecycle { wait, timeout, .. } => (*wait, *timeout), _ => return Ok(body), }; if !wait.0 { @@ -2210,6 +2297,23 @@ fn request_for(command: &Command) -> Result { Vec::new(), Some(serde_json::json!({ "machineId": machine_id })), ), + Command::ProxmoxLifecycle { + action, + account_id, + node, + vmid, + timeout, + .. + } => ( + reqwest::Method::POST, + format!("/api/v1/proxmox/accounts/{account_id}/guests/{vmid}/{action}"), + Vec::new(), + Some(serde_json::json!({ + "node": node, + "vmid": vmid, + "timeoutSeconds": timeout.unwrap_or(300), + })), + ), Command::TailnetImport { node_id, user, diff --git a/crates/fleetctl/tests/cli.rs b/crates/fleetctl/tests/cli.rs index d0a3545..02a73ee 100644 --- a/crates/fleetctl/tests/cli.rs +++ b/crates/fleetctl/tests/cli.rs @@ -2093,3 +2093,77 @@ fn an_empty_guests_page_reports_no_guests_not_no_accounts() { assert!(text.contains("(no guests reported)"), "{text}"); assert!(!text.contains("no Proxmox accounts"), "{text}"); } + +#[test] +fn parsing_walks_the_proxmox_lifecycle_forms() { + for verb in ["start", "stop", "shutdown", "reboot"] { + 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::ProxmoxLifecycle { action, .. } => { + assert_eq!(action, verb); + } + other => panic!("unexpected command {other:?}"), + } + } + // --wait and --timeout parse. + let args: Vec = [ + "proxmox", + "start", + "--account", + "acc-1", + "--node", + "pve", + "--vmid", + "101", + "--wait", + "--timeout", + "60", + ] + .iter() + .map(ToString::to_string) + .collect(); + assert!(matches!( + fleetctl::parse(&args).unwrap().command, + fleetctl::Command::ProxmoxLifecycle { + wait: true, + timeout: Some(60), + .. + } + )); + // Missing flags refuse. + let args: Vec = ["proxmox", "start", "--account", "acc-1"] + .iter() + .map(ToString::to_string) + .collect(); + let error = fleetctl::parse(&args).unwrap_err(); + assert!(error.message.contains("is required"), "{error}"); + // A bad VMID refuses. + let args: Vec = [ + "proxmox", + "start", + "--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}"); +} diff --git a/crates/providers/fleet-provider-proxmox/src/lib.rs b/crates/providers/fleet-provider-proxmox/src/lib.rs index 55892b0..3b8812b 100644 --- a/crates/providers/fleet-provider-proxmox/src/lib.rs +++ b/crates/providers/fleet-provider-proxmox/src/lib.rs @@ -70,6 +70,19 @@ pub struct PveHttpRequest { pub pinned_fingerprint: Option, /// The credentials for the call. pub credentials: Arc, + /// The HTTP method; `GET` for reads, `POST` for mutations. PVE's + /// lifecycle endpoints require `POST`. + pub method: PveHttpMethod, +} + +/// The HTTP methods the transport speaks. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum PveHttpMethod { + /// A read. + #[default] + Get, + /// A mutation. + Post, } impl PveHttpRequest { @@ -341,8 +354,11 @@ impl PveTransport for ReqwestPveTransport { request.port, request.path ); - let response = client - .get(&url) + let request_builder = match request.method { + PveHttpMethod::Get => client.get(&url), + PveHttpMethod::Post => client.post(&url), + }; + let response = request_builder .header( "Authorization", format!( @@ -543,6 +559,155 @@ pub struct PveGuestDiscovery { pub warnings: Vec, } +/// One lifecycle action on one guest. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LifecycleAction { + /// Start a stopped guest. + Start, + /// Stop a running guest immediately (no guest-side shutdown). + Stop, + /// ACPI-shutdown a running guest. + Shutdown, + /// Reboot a running guest. + Reboot, +} + +impl LifecycleAction { + /// The URL path segment PVE expects for the action. + #[must_use] + pub const fn path_segment(self) -> &'static str { + match self { + Self::Start => "start", + Self::Stop => "stop", + Self::Shutdown => "shutdown", + Self::Reboot => "reboot", + } + } + + /// Parses the stable string used in payloads and audit events. + /// + /// # Errors + /// + /// Fails on an unrecognized action id. + pub fn from_id(id: &str) -> Result { + match id { + "start" => Ok(Self::Start), + "stop" => Ok(Self::Stop), + "shutdown" => Ok(Self::Shutdown), + "reboot" => Ok(Self::Reboot), + other => Err(format!("unrecognized lifecycle action {other:?}")), + } + } + + /// The stable string used in payloads and audit events. + #[must_use] + pub const fn id(self) -> &'static str { + match self { + Self::Start => "start", + Self::Stop => "stop", + Self::Shutdown => "shutdown", + Self::Reboot => "reboot", + } + } +} + +/// A parsed UPID. Fleet parses the string itself — the node it polls comes +/// from the parse, never from trust in the caller. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Upid { + /// The hosting node the task runs on. + pub node: String, + /// The task type, e.g. `qmstart`. + pub task_type: String, + /// The task's target id (the VMID for guest tasks). + pub target: String, + /// The user the task runs as. + pub user: String, + /// The raw UPID string, for API round trips. + pub raw: String, +} + +impl Upid { + /// Parses `UPID::::::::`. + /// + /// # Errors + /// + /// Fails on a malformed UPID, with a bounded detail and no echo of the + /// raw value beyond its shape. + pub fn parse(raw: &str) -> Result { + let body = raw + .strip_prefix("UPID:") + .ok_or_else(|| "the UPID is missing its UPID: prefix".to_owned())?; + let parts: Vec<&str> = body.split(':').collect(); + if parts.len() != 8 { + return Err(format!( + "the UPID carries {} fields, expected 8", + parts.len() + )); + } + let [ + node, + pid, + pstart, + starttime, + task_type, + target, + user, + trailing, + ] = parts[..] + else { + return Err("the UPID fields did not destructure".to_owned()); + }; + for (label, part) in [ + ("node", node), + ("pid", pid), + ("pstart", pstart), + ("starttime", starttime), + ("type", task_type), + ("id", target), + ("user", user), + ] { + if part.is_empty() { + return Err(format!("the UPID's {label} field is empty")); + } + } + if !trailing.is_empty() { + return Err("the UPID carries trailing material".to_owned()); + } + if raw.len() > 256 { + return Err(format!( + "the UPID is {} bytes, over the 256-byte bound", + raw.len() + )); + } + let bounded = |part: &str, max: usize| part.chars().take(max).collect::(); + Ok(Self { + node: bounded(node, 128), + task_type: bounded(task_type, 64), + target: bounded(target, 64), + user: bounded(user, 128), + raw: raw.to_owned(), + }) + } +} + +/// The status of one PVE task, as the API reports it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TaskStatus { + /// The task is still running. + Running, + /// The task finished successfully. + Ok, + /// The task finished with an error, carrying the bounded exit status. + Error { + /// The bounded exit status string. + detail: String, + }, + /// The task is unknown to the node: it may have been rotated out of + /// the task list. Honest uncertainty, never assumed success. + Unknown, +} + /// The discovery port. The provider implements this over the PVE API; /// tests implement it over recorded fixtures. #[async_trait] @@ -568,6 +733,36 @@ pub trait ProxmoxSource: fmt::Debug + Send + Sync { &self, request: PveHttpRequest, ) -> Result; + + /// Runs one lifecycle action on one QEMU guest, returning the parsed + /// UPID of the task PVE started. Read-only until this point; this is + /// the first mutating surface in the provider. + /// + /// # Errors + /// + /// Fails with [`PveApiError`] on auth, privilege, HTTP, payload, or + /// transport failures. + async fn guest_lifecycle( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + action: LifecycleAction, + ) -> Result; + + /// Reads one task's status by node and UPID. An unknown task is an + /// honest [`TaskStatus::Unknown`], not an error: PVE rotates old task + /// entries out, and assuming success would be a lie. + /// + /// # Errors + /// + /// Fails with [`PveApiError`] on auth, privilege, HTTP, payload, or + /// transport failures. + async fn task_status( + &self, + request: PveHttpRequest, + upid: &Upid, + ) -> Result; } /// The provider client: transport plus normalization. Stateless — every @@ -802,6 +997,40 @@ impl ProxmoxSource for ProxmoxClient { warnings, }) } + + async fn guest_lifecycle( + &self, + request: PveHttpRequest, + node: &str, + vmid: u32, + action: LifecycleAction, + ) -> Result { + let lifecycle_request = PveHttpRequest { + path: format!( + "/api2/json/nodes/{}/qemu/{vmid}/status/{}", + urlencode(node), + action.path_segment() + ), + method: PveHttpMethod::Post, + ..request.clone() + }; + let data = self.call(lifecycle_request).await?; + // The mutating API answers `{"data": ""}`. + let Some(upid_raw) = data.as_str() else { + return Err(PveApiError::InvalidPayload { + detail: "the lifecycle answer carries no UPID string".to_owned(), + }); + }; + Upid::parse(upid_raw).map_err(|detail| PveApiError::InvalidPayload { detail }) + } + + async fn task_status( + &self, + request: PveHttpRequest, + upid: &Upid, + ) -> Result { + self.task_status_impl(&request, upid).await + } } /// The config's `netN` entries, parsed for MAC addresses. The value shape @@ -947,6 +1176,44 @@ impl ProxmoxClient { } agent } + + /// Reads one task's status. + async fn task_status_impl( + &self, + request: &PveHttpRequest, + upid: &Upid, + ) -> Result { + let status_request = PveHttpRequest { + path: format!( + "/api2/json/nodes/{}/tasks/{}/status", + urlencode(&upid.node), + urlencode(&upid.raw) + ), + ..request.clone() + }; + let data = self.call(status_request).await?; + // The task-status payload: `status: running|stopped`, + // `exitstatus: OK|ERROR ...`. `data: null` means the task entry is + // unknown to the node — honest uncertainty. + let Some(object) = data.as_object() else { + return Ok(TaskStatus::Unknown); + }; + match object.get("status").and_then(serde_json::Value::as_str) { + Some("running") => Ok(TaskStatus::Running), + Some("stopped") => { + match object.get("exitstatus").and_then(serde_json::Value::as_str) { + Some("OK") => Ok(TaskStatus::Ok), + Some(detail) => Ok(TaskStatus::Error { + detail: detail.chars().take(256).collect(), + }), + // A stopped task without an exit status is honest + // uncertainty, not an empty error. + None => Ok(TaskStatus::Unknown), + } + } + _ => Ok(TaskStatus::Unknown), + } + } } /// Normalizes one agent network interface. `Ok(None)` skips loopback-style @@ -1132,6 +1399,35 @@ mod tests { assert_eq!(macs[1], "de:ad:be:ef:00:02"); } + #[test] + fn upids_parse_and_refuse_malformed_shapes() { + let raw = "UPID:pve:0015523F:0C6DF532:6AAFE1EC:qmreboot:101:root@pam!GLM-AGENT:"; + let upid = Upid::parse(raw).unwrap(); + assert_eq!(upid.node, "pve"); + assert_eq!(upid.task_type, "qmreboot"); + assert_eq!(upid.target, "101"); + assert_eq!(upid.user, "root@pam!GLM-AGENT"); + + assert!(Upid::parse("not-a-upid").is_err()); + assert!(Upid::parse("UPID:pve:0015:0C6D:6AAF:qmreboot:101:").is_err()); + // A field is empty: refused. + assert!(Upid::parse("UPID:pve::0C6DF532:6AAFE1EC:qmreboot:101:user:").is_err()); + // Trailing material: refused. + assert!( + Upid::parse("UPID:pve:0015523F:0C6DF532:6AAFE1EC:qmreboot:101:user:extra").is_err() + ); + } + + #[test] + fn lifecycle_actions_round_trip_their_ids() { + for id in ["start", "stop", "shutdown", "reboot"] { + let action = LifecycleAction::from_id(id).unwrap(); + assert_eq!(action.id(), id); + assert_eq!(action.path_segment(), id); + } + assert!(LifecycleAction::from_id("destroy").is_err()); + } + #[test] fn a_net_entry_without_a_mac_warns() { let config = serde_json::json!({"net0": "bridge=vmbr0,firewall=1"}); @@ -1196,6 +1492,7 @@ mod tests { token_id: "t".to_owned(), token: SensitiveString::new("s"), }), + method: PveHttpMethod::Get, }; assert_eq!(request("2001:db8::1").authority(), "[2001:db8::1]"); assert_eq!(request("192.168.68.223").authority(), "192.168.68.223"); diff --git a/crates/providers/fleet-provider-proxmox/tests/pin_live.rs b/crates/providers/fleet-provider-proxmox/tests/pin_live.rs index 3349623..62de024 100644 --- a/crates/providers/fleet-provider-proxmox/tests/pin_live.rs +++ b/crates/providers/fleet-provider-proxmox/tests/pin_live.rs @@ -41,6 +41,7 @@ async fn pinned_transport_converses_with_the_live_host() { token_id, token: SensitiveString::new(key), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; let client = ProxmoxClient::new(transport); let discovery = client.discover(request).await.unwrap(); @@ -68,6 +69,7 @@ async fn a_wrong_fingerprint_is_refused_at_the_handshake() { token_id, token: SensitiveString::new(key), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; let error = transport.execute(request).await.unwrap_err(); match error { @@ -99,6 +101,7 @@ async fn an_unpinned_host_is_observed_not_conversed_with() { token_id, token: SensitiveString::new(key), }), + method: fleet_provider_proxmox::PveHttpMethod::Get, }; let error = transport.execute(request).await.unwrap_err(); match error { diff --git a/packages/api-client/openapi.json b/packages/api-client/openapi.json index ed66ab1..ed0c67a 100644 --- a/packages/api-client/openapi.json +++ b/packages/api-client/openapi.json @@ -2417,6 +2417,89 @@ } } }, + "/api/v1/proxmox/accounts/{accountId}/guests/{vmid}/{action}": { + "post": { + "tags": [ + "proxmox" + ], + "summary": "Runs a lifecycle action on one guest as a durable operation. The\noperation is authorized through the catalog's `proxmox.operate`\n(catalog-level, like the source kinds — a Proxmox guest is not a Fleet\nmachine) and executed by the worker with Fleet-owned UPID polling.", + "description": "# Errors\n\nReturns the public error envelope on refusal or a malformed request.", + "operationId": "startProxmoxLifecycle", + "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 lifecycle action: start, stop, shutdown, or reboot.", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StartProxmoxLifecycleRequest" + } + } + }, + "required": true + }, + "responses": { + "202": { + "description": "The lifecycle operation was accepted and is durable.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Resource_OperationDto" + } + } + } + }, + "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.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + } + } + } + }, "/api/v1/proxmox/accounts/{accountId}/observe": { "post": { "tags": [ @@ -6896,6 +6979,33 @@ } } }, + "StartProxmoxLifecycleRequest": { + "type": "object", + "description": "The lifecycle request: the guest's node and VMID.", + "required": [ + "node", + "vmid", + "timeoutSeconds" + ], + "properties": { + "node": { + "type": "string", + "description": "The guest's hosting node." + }, + "timeoutSeconds": { + "type": "integer", + "format": "int64", + "description": "The deadline, in seconds. Bounded by the executor.", + "minimum": 0 + }, + "vmid": { + "type": "integer", + "format": "int32", + "description": "The guest's VMID.", + "minimum": 0 + } + } + }, "StartReadyRequest": { "type": "object", "description": "The body of the start-ready-workflow request.", diff --git a/packages/api-client/src/generated/fleet.ts b/packages/api-client/src/generated/fleet.ts index c2504ce..d2da351 100644 --- a/packages/api-client/src/generated/fleet.ts +++ b/packages/api-client/src/generated/fleet.ts @@ -2289,6 +2289,24 @@ export interface StartMiseOperationRequest { version?: string | null; } +/** + * The lifecycle request: the guest's node and VMID. + */ +export interface StartProxmoxLifecycleRequest { + /** The guest's hosting node. */ + node: string; + /** + * The deadline, in seconds. Bounded by the executor. + * @minimum 0 + */ + timeoutSeconds: number; + /** + * The guest's VMID. + * @minimum 0 + */ + vmid: number; +} + /** * The body of the start-ready-workflow request. */ @@ -5032,6 +5050,78 @@ const res = await fetch(getObserveProxmoxGuestUrl(accountId,vmid), +export type startProxmoxLifecycleResponse202 = { + data: ResourceOperationDto + status: 202 +} + +export type startProxmoxLifecycleResponse400 = { + data: ApiError + status: 400 +} + +export type startProxmoxLifecycleResponse403 = { + data: ApiError + status: 403 +} + +export type startProxmoxLifecycleResponseSuccess = (startProxmoxLifecycleResponse202) & { + headers: Headers; +}; +export type startProxmoxLifecycleResponseError = (startProxmoxLifecycleResponse400 | startProxmoxLifecycleResponse403) & { + headers: Headers; +}; + +export type startProxmoxLifecycleResponse = (startProxmoxLifecycleResponseSuccess | startProxmoxLifecycleResponseError) + +export const getStartProxmoxLifecycleUrl = (accountId: string, + vmid: number, + action: string,) => { + + + + + return `/api/v1/proxmox/accounts/${accountId}/guests/${vmid}/${action}` +} + +/** + * # Errors + * + * Returns the public error envelope on refusal or a malformed request. + * @summary Runs a lifecycle action on one guest as a durable operation. The +operation is authorized through the catalog's `proxmox.operate` +(catalog-level, like the source kinds — a Proxmox guest is not a Fleet +machine) and executed by the worker with Fleet-owned UPID polling. + */ +export const startProxmoxLifecycle = async (accountId: string, + vmid: number, + action: string, + startProxmoxLifecycleRequest: StartProxmoxLifecycleRequest, 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(getStartProxmoxLifecycleUrl(accountId,vmid,action), + { + ...options, + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getHeaders(options?.headers) }, + body: JSON.stringify(startProxmoxLifecycleRequest) + } +) + + + const body = [204, 205, 304].includes(res.status) ? null : await res.text(); + + const data: startProxmoxLifecycleResponse['data'] = body ? JSON.parse(body) : {} + return { data, status: res.status, headers: res.headers } as startProxmoxLifecycleResponse +} + + + export type observeProxmoxFingerprintResponse200 = { data: ResourceProxmoxFingerprintDto status: 200