diff --git a/crates/fleet-api/src/apply.rs b/crates/fleet-api/src/apply.rs new file mode 100644 index 0000000..5ca2ee2 --- /dev/null +++ b/crates/fleet-api/src/apply.rs @@ -0,0 +1,300 @@ +//! The apply surface (FM-402): authorized plan execution with approvals. +//! This adapter decides nothing; it translates HTTP into operation +//! creation and authorization calls. + +use std::sync::Arc; + +use axum::{ + Extension, Json, + extract::{Path, State}, + http::StatusCode, +}; +use fleet_core::CorrelationId; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::envelope::Resource; +use crate::error::ApiErrorResponse; + +/// How the apply workflow's endpoint authenticates. +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "camelCase", tag = "type")] +pub enum ApplyAuthDto { + /// The controller's agent supplies the key. + Agent, + /// A specific identity file. + IdentityFile { + /// The identity file's path. + path: String, + }, +} + +/// The action kinds the apply workflow can execute. +const SUPPORTED_KINDS: [&str; 4] = [ + "mise.install", + "skills.deploy", + "skills.undeploy", + "projects.clone", +]; + +/// One field difference, as the planner produced it. +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct FieldDifferenceDto { + /// The field's stable identity. + pub identity: String, + /// The drift state. + pub state: String, + /// The desired value, when the field is desired. + pub desired: Option, + /// The observed value, when one was observed. + pub observed: Option, + /// Why the state is `unknown` or `unsupported`, when it is. + pub reason: Option, +} + +/// One planned action the caller submits. +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ApplyActionDto { + /// The execution order. + pub order: u32, + /// The operation kind. + pub kind: String, + /// The difference the action resolves. + pub difference: FieldDifferenceDto, +} + +/// One approval the caller supplies. +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ApplyApprovalDto { + /// The plan's identity the approval is bound to. + pub plan_id: String, + /// The action's order the approval covers. + pub action_order: u32, + /// The action's operation kind the approval covers. + pub kind: String, +} + +/// The body of the start-apply-workflow request. +#[derive(Debug, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct StartApplyRequest { + /// The machine to apply on (must match the path's machine). + pub machine_id: String, + /// The SSH endpoint id to act through. + pub endpoint_id: String, + /// How the endpoint authenticates. + pub auth: ApplyAuthDto, + /// The plan's identity, which every approval is bound to. + pub plan_id: String, + /// The planned actions, in order. + pub actions: Vec, + /// The approvals supplied with the plan. + #[serde(default)] + pub approvals: Vec, + /// The deadline, in seconds, for the whole workflow. + #[serde(default = "default_timeout")] + pub timeout_seconds: u64, +} + +fn default_timeout() -> u64 { + 1800 +} + +/// Starts the apply workflow. +/// +/// # Errors +/// +/// Returns the public error envelope on refusal or backend failure. +#[utoipa::path( + post, + path = "/machines/{machineId}/apply", + tag = "machines", + operation_id = "startApplyWorkflow", + request_body = StartApplyRequest, + params( + ("machineId" = String, Path, description = "The machine to apply on.") + ), + responses( + ( + status = 202, + description = "The apply workflow was accepted. Requires machine.read for the machine in addition to apply.execute.", + body = Resource + ), + ( + status = 400, + description = "The request is malformed.", + body = crate::error::ApiError + ), + ( + status = 403, + description = "The caller may not execute apply plans.", + body = crate::error::ApiError + ), + ( + status = 404, + description = "The machine does not exist.", + body = crate::error::ApiError + ), + ) +)] +#[allow(clippy::too_many_lines)] +pub async fn start_apply_workflow( + State(state): State>, + principal: Option>, + Extension(correlation_id): Extension, + headers: axum::http::HeaderMap, + Path(machine_id): Path, + Json(request): Json, +) -> Result<(StatusCode, Json>), ApiErrorResponse> { + let machines = crate::machines::machines_or_error(&state, correlation_id)?; + let principal = crate::operations::principal_or_error(principal, correlation_id)?; + if request.machine_id != machine_id { + return Err(crate::machines::invalid_request( + "the body's machineId does not match the path's machine", + correlation_id, + )); + } + let _machine = machines + .get( + state.authorizer.as_ref(), + &principal, + &machine_id, + fleet_core::SystemClock::now_unix_millis(), + ) + .await + .map_err(|error| crate::machines::map_machine_error(&error, 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::ApplyExecute, + resource: Some(&machine_id), + }, + ) { + return Err(crate::machines::denied_error(decision, correlation_id)); + } + if request.actions.is_empty() { + return Err(crate::machines::invalid_request( + "the plan must carry at least one action", + correlation_id, + )); + } + // The plan identity is required: approvals are bound to it. + if request.plan_id.is_empty() { + return Err(crate::machines::invalid_request( + "the plan identity is required; approvals are bound to it", + correlation_id, + )); + } + // Each action's kind must be one the apply executor can run, and its + // state must be one of the five documented drift states; orders must + // be unique and strictly increasing. Malformed input is a 400, never + // a queued failure. + let mut previous_order: Option = None; + for action in &request.actions { + if !SUPPORTED_KINDS.contains(&action.kind.as_str()) { + return Err(crate::machines::invalid_request( + &format!( + "the action kind {:?} is not one the apply workflow can execute", + action.kind + ), + correlation_id, + )); + } + let state = fleet_core::DifferenceState::deserialize(serde_json::Value::String( + action.difference.state.clone(), + )) + .map_err(|_| { + crate::machines::invalid_request( + &format!( + "the difference state {:?} is not one of the documented drift states", + action.difference.state + ), + correlation_id, + ) + })?; + // The state must be actionable AND pair with the kind the way the + // planner maps them: an unknown/unsupported state has no bounded + // action, and a kind/state mismatch (skills.deploy with extra) + // would perform a side effect the plan never declared. + if !state.actionable() { + return Err(crate::machines::invalid_request( + "an apply action must carry an actionable difference state", + correlation_id, + )); + } + let state_matches_kind = matches!( + (action.kind.as_str(), state), + ( + "mise.install" | "skills.deploy" | "projects.clone", + fleet_core::DifferenceState::Missing + ) | ( + "mise.install" | "projects.clone", + fleet_core::DifferenceState::Changed + ) | ("skills.undeploy", fleet_core::DifferenceState::Extra) + ); + // The identity prefix must match the kind too: the executor + // derives its payload by stripping the kind's expected prefix, so + // a mismatched identity would execute a side effect the plan + // never declared. + let expected_prefix = match action.kind.as_str() { + "mise.install" => "tool:", + "skills.deploy" | "skills.undeploy" => "skill:", + _ => "checkout:", + }; + if !state_matches_kind || !action.difference.identity.starts_with(expected_prefix) { + return Err(crate::machines::invalid_request( + &format!( + "the action kind {:?} does not resolve a {:?} difference with an {:?} identity", + action.kind, state, expected_prefix + ), + correlation_id, + )); + } + if previous_order.is_some_and(|previous| action.order <= previous) { + return Err(crate::machines::invalid_request( + "the action orders must be unique and strictly increasing", + correlation_id, + )); + } + previous_order = Some(action.order); + } + let payload = serde_json::json!({ + "machineId": machine_id, + "endpointId": request.endpoint_id, + "auth": serde_json::to_value(&request.auth) + .map_err(|error| crate::machines::invalid_request(&error.to_string(), correlation_id))?, + "planId": request.plan_id, + "actions": request.actions, + "approvals": request.approvals, + "timeoutSeconds": request.timeout_seconds, + }); + 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: "apply.workflow".to_owned(), + 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-api/src/lib.rs b/crates/fleet-api/src/lib.rs index d3ac022..0da41c5 100644 --- a/crates/fleet-api/src/lib.rs +++ b/crates/fleet-api/src/lib.rs @@ -11,6 +11,7 @@ #![warn(missing_docs)] +pub mod apply; mod correlation; mod envelope; mod error; @@ -95,6 +96,11 @@ pub const API_BASE_PATH: &str = "/api/v1"; ready::ReadyAuthDto, ready::ReadyToolDto, ready::StartReadyRequest, + apply::ApplyActionDto, + apply::ApplyApprovalDto, + apply::ApplyAuthDto, + apply::FieldDifferenceDto, + apply::StartApplyRequest, projects::CreateProjectRequest, projects::ProjectDto, projects::UpdateProjectRequest, @@ -180,6 +186,7 @@ pub fn api(state: Arc) -> (Router, utoipa::openapi::OpenAp .routes(routes!(frogenv::start_frogenv_operation)) .routes(routes!(mise::start_mise_operation)) .routes(routes!(ready::start_ready_workflow)) + .routes(routes!(apply::start_apply_workflow)) .routes(routes!( onboarding::create_onboarding_draft, onboarding::list_onboarding_drafts diff --git a/crates/fleet-api/tests/apply_surface.rs b/crates/fleet-api/tests/apply_surface.rs new file mode 100644 index 0000000..b93abaa --- /dev/null +++ b/crates/fleet-api/tests/apply_surface.rs @@ -0,0 +1,533 @@ +//! The apply surface (FM-402): authorization, validation, and idempotent +//! creation at the boundary. + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use fleet_api::{API_BASE_PATH, CORRELATION_ID_HEADER, operations::ApiState, router}; +use fleet_application::machine::{ + Endpoint, Machine, MachineFilter, MachinePort, NewEndpoint, RegisterMachine, +}; +use fleet_application::operation::{Operations, PortFailure}; +use fleet_core::EndpointKind; +use std::sync::{Arc, Mutex}; +use tower::ServiceExt as _; + +#[derive(Debug)] +struct PermitAll; +impl fleet_application::authz::Authorizer for PermitAll { + fn decide( + &self, + _request: fleet_application::authz::AccessRequest<'_>, + ) -> fleet_application::authz::Decision { + fleet_application::authz::Decision::allow() + } +} + +#[derive(Debug)] +struct DenyApply; +impl fleet_application::authz::Authorizer for DenyApply { + fn decide( + &self, + request: fleet_application::authz::AccessRequest<'_>, + ) -> fleet_application::authz::Decision { + if request.action == fleet_application::authz::Permission::ApplyExecute { + fleet_application::authz::Decision::deny( + fleet_application::authz::ReasonId::UnknownPrincipal, + ) + } else { + fleet_application::authz::Decision::allow() + } + } +} + +/// Allows `ApplyExecute` for exactly one machine: the authorizer that +/// proves machine scoping, since an unconditional denial cannot +/// distinguish a machine-aware check from a resource-blind one. +#[derive(Debug)] +struct MachineScoped { + allowed_machine: String, +} +impl fleet_application::authz::Authorizer for MachineScoped { + fn decide( + &self, + request: fleet_application::authz::AccessRequest<'_>, + ) -> fleet_application::authz::Decision { + if request.action == fleet_application::authz::Permission::OperationCreate { + // A catalog-level action: no resource, no machine scoping to + // prove here. + fleet_application::authz::Decision::allow() + } else if request + .resource + .is_some_and(|resource| resource == self.allowed_machine) + && matches!( + request.action, + fleet_application::authz::Permission::ApplyExecute + | fleet_application::authz::Permission::MachineRead + | fleet_application::authz::Permission::MachineReadSensitive + ) + { + fleet_application::authz::Decision::allow() + } else { + fleet_application::authz::Decision::deny( + fleet_application::authz::ReasonId::UnknownPrincipal, + ) + } + } +} + +#[derive(Debug, Default)] +struct FakeOperations { + payloads: Mutex>, + idempotency_keys: Mutex>>, +} + +#[async_trait::async_trait] +impl fleet_application::operation::OperationPort for FakeOperations { + async fn create( + &self, + kind: &str, + idempotency_key: Option<&str>, + _deadline_at: Option, + correlation_id: Option<&str>, + payload_json: Option<&str>, + ) -> Result { + self.payloads + .lock() + .unwrap() + .push(payload_json.unwrap_or_default().to_owned()); + self.idempotency_keys + .lock() + .unwrap() + .push(idempotency_key.map(str::to_owned)); + Ok(fleet_application::operation::Operation { + id: "op-1".to_owned(), + kind: kind.to_owned(), + state: "pending".to_owned(), + idempotency_key: idempotency_key.map(str::to_owned), + progress_current: None, + progress_total: None, + progress_message: None, + deadline_at: None, + cancel_requested: false, + payload_json: payload_json.map(str::to_owned), + result_json: None, + error_json: None, + correlation_id: correlation_id.map(str::to_owned), + claimed_at: None, + worker_id: None, + created_at: 0, + updated_at: 0, + }) + } + async fn get(&self, _id: &str) -> Result { + unimplemented!() + } + async fn list( + &self, + _limit: u32, + ) -> Result, PortFailure> { + unimplemented!() + } + async fn request_cancel( + &self, + _id: &str, + ) -> Result { + unimplemented!() + } + async fn transition( + &self, + _id: &str, + _state: &str, + ) -> Result { + unimplemented!() + } + async fn complete( + &self, + _id: &str, + _state: &str, + _result_json: Option<&str>, + _error_json: Option<&str>, + ) -> Result { + unimplemented!() + } + async fn record_progress( + &self, + _id: &str, + _current: Option, + _total: Option, + _message: Option<&str>, + ) -> Result<(), PortFailure> { + unimplemented!() + } + async fn claim_pending( + &self, + _worker_id: &str, + _now: i64, + ) -> Result, PortFailure> { + unimplemented!() + } + async fn claim_pending_by_id( + &self, + _id: &str, + _worker_id: &str, + _now: i64, + ) -> Result, PortFailure> { + unimplemented!() + } + async fn expired_claims( + &self, + _now: i64, + _lease_ms: i64, + ) -> Result, PortFailure> { + unimplemented!() + } + async fn renew_lease( + &self, + _id: &str, + _worker_id: &str, + _now: i64, + _lease_ms: i64, + ) -> Result { + unimplemented!() + } + async fn fail_expired_claim( + &self, + _id: &str, + _expected_claimed_at: i64, + _now: i64, + _error_json: &str, + ) -> Result { + unimplemented!() + } + async fn sweep_deadlines(&self, _now: i64) -> Result, PortFailure> { + unimplemented!() + } + async fn queue_depths(&self) -> Result { + unimplemented!() + } +} + +#[derive(Debug)] +struct FakeAudit; +#[async_trait::async_trait] +impl fleet_application::operation::AuditPort for FakeAudit { + async fn record_intent( + &self, + _intent: &fleet_application::audit::AuditIntent, + ) -> Result<(), String> { + Ok(()) + } + async fn record_outcome( + &self, + _operation_id: &str, + _outcome: fleet_application::audit::AuditOutcome, + ) -> Result<(), String> { + Ok(()) + } +} + +#[derive(Debug)] +struct FakeMachines; + +#[async_trait::async_trait] +impl MachinePort for FakeMachines { + async fn register(&self, _registration: &RegisterMachine) -> Result { + unimplemented!() + } + async fn get(&self, id: &str) -> Result { + if id != "m-1" { + return Err(PortFailure::NotFound { + what: "machine".to_owned(), + }); + } + Ok(Machine { + id: id.to_owned(), + name: "box".to_owned(), + description: String::new(), + endpoints: vec![Endpoint { + id: "e-1".to_owned(), + kind: EndpointKind::Ssh, + reference: "user@host:22".to_owned(), + }], + tags: vec![], + groups: vec![], + capabilities: vec![], + last_observation: None, + node: None, + created_at: 0, + updated_at: 0, + }) + } + async fn list( + &self, + _filter: &MachineFilter, + _limit: u32, + ) -> Result, PortFailure> { + unimplemented!() + } + async fn update( + &self, + _id: &str, + _name: &str, + _description: &str, + ) -> Result { + unimplemented!() + } + async fn set_endpoints( + &self, + _id: &str, + _endpoints: &[NewEndpoint], + ) -> Result { + unimplemented!() + } + async fn add_tag(&self, _id: &str, _tag: &str) -> Result { + unimplemented!() + } + async fn remove_tag(&self, _id: &str, _tag: &str) -> Result { + unimplemented!() + } + async fn add_group(&self, _id: &str, _group: &str) -> Result { + unimplemented!() + } + async fn remove_group(&self, _id: &str, _group: &str) -> Result { + unimplemented!() + } + async fn record_snapshot( + &self, + _id: &str, + _source: &str, + _payload_json: &str, + _collected_at: i64, + ) -> Result<(), PortFailure> { + unimplemented!() + } + async fn record_capabilities( + &self, + _id: &str, + _facts: &[fleet_core::CapabilityFact], + ) -> Result<(), PortFailure> { + unimplemented!() + } + async fn delete(&self, _id: &str) -> Result<(), PortFailure> { + unimplemented!() + } + async fn confirm_fingerprint( + &self, + _endpoint_id: &str, + _fingerprint: &str, + _confirmed_at: i64, + ) -> Result<(), PortFailure> { + unimplemented!() + } + async fn verified_fingerprint( + &self, + _endpoint_id: &str, + ) -> Result, PortFailure> { + Ok(None) + } + async fn latest_inventory_revision(&self, _id: &str) -> Result, PortFailure> { + Ok(None) + } +} + +#[derive(Debug)] +struct FakeSystemInfo; +#[async_trait::async_trait] +impl fleet_api::system::SystemInfoSource for FakeSystemInfo { + async fn info(&self) -> Result { + unimplemented!("the tests never read the system surface") + } +} + +fn state_for(authorizer: Arc) -> Arc { + Arc::new(ApiState { + operations: Arc::new(Operations::new( + Arc::new(FakeOperations::default()), + Arc::new(FakeAudit), + )), + authorizer, + system: Arc::new(FakeSystemInfo), + nodes: None, + machines: Some(Arc::new(fleet_application::machine::Machines::new( + Arc::new(FakeMachines), + Arc::new(FakeAudit), + ))), + onboarding: None, + tailnet: None, + projects: None, + }) +} + +async fn call( + state: Arc, + method: &str, + path: &str, + body: Option, +) -> (StatusCode, serde_json::Value) { + let router = router(state).layer(axum::Extension(fleet_api::ActingPrincipal { + id: "anonymous-lan-admin".to_owned(), + })); + let request = Request::builder() + .method(method) + .uri(format!("{API_BASE_PATH}{path}")) + .header( + CORRELATION_ID_HEADER, + "01900000-0000-7000-8000-000000000000", + ); + let request = if let Some(body) = body { + request + .header("content-type", "application/json") + .body(Body::from(body)) + } else { + request.body(Body::empty()) + } + .unwrap(); + let response = router.oneshot(request).await.unwrap(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), 1 << 20) + .await + .unwrap(); + let value = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null); + (status, value) +} + +fn body_for(machine: &str) -> serde_json::Value { + serde_json::json!({ + "machineId": machine, + "endpointId": "e-1", + "auth": {"type": "agent"}, + "planId": "plan-1", + "actions": [ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "missing", + "desired": "20.11.0", "observed": null, "reason": null}}, + ], + "approvals": [ + {"planId": "plan-1", "actionOrder": 1, "kind": "mise.install"}, + ], + "timeoutSeconds": 600, + }) +} + +#[tokio::test] +async fn a_valid_plan_is_accepted() { + let state = state_for(Arc::new(PermitAll)); + let (status, value) = call( + state, + "POST", + "/machines/m-1/apply", + Some(body_for("m-1").to_string()), + ) + .await; + assert_eq!(status, StatusCode::ACCEPTED, "{value}"); + assert_eq!(value["data"]["kind"], "apply.workflow"); +} + +#[tokio::test] +async fn an_unknown_kind_is_refused_at_the_boundary() { + let state = state_for(Arc::new(PermitAll)); + let mut body = body_for("m-1"); + body["actions"] = serde_json::json!([ + {"order": 1, "kind": "demolish", + "difference": {"identity": "x", "state": "missing", + "desired": "y", "observed": null, "reason": null}}, + ]); + let (status, value) = call(state, "POST", "/machines/m-1/apply", Some(body.to_string())).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{value}"); +} + +#[tokio::test] +async fn an_undocumented_state_is_refused_at_the_boundary() { + let state = state_for(Arc::new(PermitAll)); + let mut body = body_for("m-1"); + body["actions"] = serde_json::json!([ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "demolished", + "desired": "20.11.0", "observed": null, "reason": null}}, + ]); + let (status, value) = call(state, "POST", "/machines/m-1/apply", Some(body.to_string())).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{value}"); +} + +#[tokio::test] +async fn duplicate_or_non_increasing_orders_are_refused() { + let state = state_for(Arc::new(PermitAll)); + let mut body = body_for("m-1"); + body["actions"] = serde_json::json!([ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "missing", + "desired": "20.11.0", "observed": null, "reason": null}}, + {"order": 1, "kind": "skills.deploy", + "difference": {"identity": "skill:db/claude_code", "state": "missing", + "desired": "deployed", "observed": null, "reason": null}}, + ]); + let (status, value) = call(state, "POST", "/machines/m-1/apply", Some(body.to_string())).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{value}"); +} + +#[tokio::test] +async fn a_body_machine_disagreeing_with_the_path_is_malformed() { + let state = state_for(Arc::new(PermitAll)); + let (status, value) = call( + state, + "POST", + "/machines/m-1/apply", + Some(body_for("other").to_string()), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{value}"); +} + +#[tokio::test] +async fn an_apply_denial_is_a_machine_scoped_403() { + let state = state_for(Arc::new(DenyApply)); + let (status, value) = call( + state, + "POST", + "/machines/m-1/apply", + Some(body_for("m-1").to_string()), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{value}"); +} + +#[tokio::test] +async fn the_apply_permission_is_machine_scoped() { + // The authorizer allows apply only on m-1: a request against m-2 is + // denied, proving the endpoint names the machine as its resource. + let state = state_for(Arc::new(MachineScoped { + allowed_machine: "m-1".to_owned(), + })); + let (status, _) = call( + state.clone(), + "POST", + "/machines/m-1/apply", + Some(body_for("m-1").to_string()), + ) + .await; + assert_eq!(status, StatusCode::ACCEPTED); + let state = state_for(Arc::new(MachineScoped { + allowed_machine: "m-1".to_owned(), + })); + let (status, value) = call( + state, + "POST", + "/machines/m-2/apply", + Some(body_for("m-2").to_string()), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{value}"); +} + +#[tokio::test] +async fn an_unknown_machine_is_a_404() { + let state = state_for(Arc::new(PermitAll)); + let (status, value) = call( + state, + "POST", + "/machines/nope/apply", + Some(body_for("nope").to_string()), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "{value}"); +} diff --git a/crates/fleet-application/src/apply.rs b/crates/fleet-application/src/apply.rs new file mode 100644 index 0000000..eaccf2e --- /dev/null +++ b/crates/fleet-application/src/apply.rs @@ -0,0 +1,290 @@ +//! The apply engine (FM-402): authorized, durable, compensating +//! execution of an apply plan. +//! +//! The apply use case takes an FM-401 plan, the machine it targets, and +//! the caller's approvals; authorizes the execution; and creates one +//! durable `apply.workflow` operation. The workflow walks the plan's +//! actions in order, claiming and executing each inner step in-process +//! through the composed chain — the proven FM-305 shape, with three +//! additions the apply semantics require: +//! +//! - **Approvals**: a step whose operation kind is flagged risky in the +//! authz catalog requires an approval bound to the plan's identity. A +//! plan missing its approvals completes `blocked_manual_approval` +//! naming the unapproved steps; it never executes partially approved. +//! - **Compensation**: each completed step records its compensation +//! (undeploy for a deploy, none for an idempotent install) in the +//! operation record. Compensation execution is an explicit authorized +//! operation, never automatic. +//! - **Post-apply verification**: the plan's final step re-runs the +//! FM-401 comparison and requires an empty actionable difference set +//! before the workflow completes succeeded. +//! +//! Restart/resume rides the operation record: completed and remaining +//! steps are durable, so a controller restart resumes truthfully. +#![warn(missing_docs)] + +use fleet_core::{DifferenceSet, DifferenceState}; + +/// The approval identity binding: an approval is valid only for the plan +/// it names, so a token from one plan cannot authorize another. +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Approval { + /// The plan's identity the approval is bound to. + pub plan_id: String, + /// The action's order the approval covers. + pub action_order: u32, + /// The action's operation kind the approval covers. + pub kind: String, +} + +/// The operation kinds the apply approval gate covers, derived from the +/// authz catalog's risk classification for the kind's own permission: +/// a step whose permission is risky in the catalog requires an approval +/// here too, so the gate cannot drift from the catalog. +#[must_use] +pub fn requires_approval(kind: &str) -> bool { + let permission = fleet_application_kind_permission(kind); + permission.is_some_and(super::authz::Permission::is_risky) +} + +/// The authz permission each apply-plan kind maps to. Single source of +/// truth for the approval gate, derived from the same mapping +/// `Operations::create` enforces. +fn fleet_application_kind_permission(kind: &str) -> Option { + match kind { + "mise.install" => Some(crate::authz::Permission::MiseOperate), + "skills.deploy" | "skills.undeploy" => Some(crate::authz::Permission::SkillsDeploy), + "projects.clone" => Some(crate::authz::Permission::ProjectsGitWrite), + _ => None, + } +} + +/// The compensation an executed step records. +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum Compensation { + /// The step is idempotent: re-running it is its own compensation. + Idempotent, + /// The step deployed a skill; the compensation is an undeploy. + Undeploy { + /// The skill that was deployed. + skill_id: String, + /// The agent it was deployed to. + agent: String, + }, + /// The step has no safe compensation: the record says so honestly. + None, +} + +impl Compensation { + /// The compensation for one completed step's kind and difference. + #[must_use] + pub fn for_step(kind: &str, difference: &fleet_core::FieldDifference) -> Self { + match kind { + "mise.install" | "projects.clone" => Self::Idempotent, + "skills.deploy" => { + // The difference's identity carries skill/agent. + let identity = &difference.identity; + let rest = identity.strip_prefix("skill:").unwrap_or(identity); + match rest.split_once('/') { + Some((skill_id, agent)) => Self::Undeploy { + skill_id: skill_id.to_owned(), + agent: agent.to_owned(), + }, + None => Self::None, + } + } + _ => Self::None, + } + } +} + +/// Validates the approvals for one plan: every risky action must carry an +/// approval bound to this plan's identity and its own order and kind. +/// Returns the unapproved actions; an empty result means the plan may +/// execute. +#[must_use] +pub fn unapproved_actions( + plan_id: &str, + actions: &[crate::planner::PlannedAction], + approvals: &[Approval], +) -> Vec { + actions + .iter() + .filter(|action| requires_approval(&action.kind)) + .filter(|action| { + !approvals.iter().any(|approval| { + approval.plan_id == plan_id + && approval.action_order == action.order + && approval.kind == action.kind + }) + }) + .cloned() + .collect() +} + +/// Verifies post-apply truth: the re-observed difference set must have no +/// actionable fields. An `unsupported` field is reported without +/// blocking; an `unknown` blocks — the machine's state is not knowable +/// enough to claim convergence. +/// # Errors +/// +/// Returns the blocking fields when the re-observed difference set still +/// carries actionable or unknown fields. +pub fn verified(set: &DifferenceSet) -> Result<(), String> { + // A canonicalized clone preserves terminal-state precedence on + // duplicate identities, so an honest unknown cannot be shadowed by an + // actionable duplicate. + let mut canonical = set.clone(); + canonical.canonicalize(); + let blockers: Vec<&fleet_core::FieldDifference> = canonical + .fields + .iter() + .filter(|field| field.state == DifferenceState::Unknown || field.actionable()) + .collect(); + if blockers.is_empty() { + Ok(()) + } else { + Err(blockers + .iter() + .map(|field| format!("{} ({})", field.identity, field.state)) + .collect::>() + .join("; ")) + } +} + +#[cfg(test)] +mod tests { + use super::{Approval, Compensation, unapproved_actions, verified}; + use crate::planner::PlannedAction; + use fleet_core::{DifferenceSet, FieldDifference}; + + fn action(order: u32, kind: &str, identity: &str) -> PlannedAction { + PlannedAction { + order, + kind: kind.to_owned(), + difference: FieldDifference::missing(identity, "desired"), + reason: "test".to_owned(), + } + } + + #[test] + fn risky_kinds_require_approval_and_safe_ones_do_not() { + assert!(super::requires_approval("mise.install")); + assert!(super::requires_approval("skills.deploy")); + assert!(super::requires_approval("skills.undeploy")); + // The clone's catalog permission (projects.git.write) is risky, so + // the gate covers it too — derived from the catalog, not hardcoded. + assert!(super::requires_approval("projects.clone")); + } + + #[test] + fn an_unapproved_risky_action_blocks_the_plan() { + let actions = vec![action(1, "mise.install", "tool:node")]; + let unapproved = unapproved_actions("plan-1", &actions, &[]); + assert_eq!(unapproved.len(), 1); + assert_eq!(unapproved[0].kind, "mise.install"); + } + + #[test] + fn an_approval_bound_to_another_plan_does_not_authorize() { + let actions = vec![action(1, "mise.install", "tool:node")]; + let approvals = vec![Approval { + plan_id: "plan-2".to_owned(), + action_order: 1, + kind: "mise.install".to_owned(), + }]; + let unapproved = unapproved_actions("plan-1", &actions, &approvals); + assert_eq!(unapproved.len(), 1, "a foreign plan's approval is not ours"); + } + + #[test] + fn an_approval_bound_to_another_action_does_not_authorize() { + let actions = vec![ + action(1, "mise.install", "tool:node"), + action(2, "mise.install", "tool:python"), + ]; + let approvals = vec![Approval { + plan_id: "plan-1".to_owned(), + action_order: 1, + kind: "mise.install".to_owned(), + }]; + let unapproved = unapproved_actions("plan-1", &actions, &approvals); + assert_eq!(unapproved.len(), 1); + assert_eq!(unapproved[0].difference.identity, "tool:python"); + } + + #[test] + fn a_fully_approved_plan_executes() { + let actions = vec![ + action(1, "mise.install", "tool:node"), + action(2, "skills.deploy", "skill:db/claude_code"), + ]; + let approvals = vec![ + Approval { + plan_id: "plan-1".to_owned(), + action_order: 1, + kind: "mise.install".to_owned(), + }, + Approval { + plan_id: "plan-1".to_owned(), + action_order: 2, + kind: "skills.deploy".to_owned(), + }, + ]; + assert!(unapproved_actions("plan-1", &actions, &approvals).is_empty()); + } + + #[test] + fn compensations_match_the_step_semantics() { + let install = Compensation::for_step( + "mise.install", + &FieldDifference::missing("tool:node", "20.11.0"), + ); + assert_eq!(install, Compensation::Idempotent); + let deploy = Compensation::for_step( + "skills.deploy", + &FieldDifference::missing("skill:db/claude_code", "deployed"), + ); + assert_eq!( + deploy, + Compensation::Undeploy { + skill_id: "db".to_owned(), + agent: "claude_code".to_owned(), + } + ); + let clone = Compensation::for_step( + "projects.clone", + &FieldDifference::missing("checkout:github.com/x/y", "/srv/y"), + ); + assert_eq!(clone, Compensation::Idempotent); + let unknown_kind = + Compensation::for_step("mystery.kind", &FieldDifference::missing("mystery:x", "y")); + assert_eq!(unknown_kind, Compensation::None); + } + + #[test] + fn post_apply_verification_requires_no_actionable_or_unknown_fields() { + let mut set = DifferenceSet::new(); + set.push(FieldDifference::unsupported( + "tool:exotic", + Some("1.0"), + "no path", + )); + assert!( + verified(&set).is_ok(), + "an unsupported field is reported without blocking" + ); + set.push(FieldDifference::missing("tool:node", "20.11.0")); + assert!(verified(&set).is_err(), "an actionable difference blocks"); + let mut set = DifferenceSet::new(); + set.push(FieldDifference::unknown( + "tool:node", + Some("20.11.0"), + "no answer", + )); + assert!(verified(&set).is_err(), "an unknown blocks the claim"); + } +} diff --git a/crates/fleet-application/src/authz.rs b/crates/fleet-application/src/authz.rs index 9c0f68e..6cda225 100644 --- a/crates/fleet-application/src/authz.rs +++ b/crates/fleet-application/src/authz.rs @@ -120,6 +120,9 @@ pub enum Permission { /// and verify. A mutation: it composes every mutation the workflow /// may run. ProjectsReady, + /// Execute an authorized apply plan on a machine. A mutation: it + /// composes every mutation the plan may run. + ApplyExecute, } impl Permission { @@ -160,6 +163,7 @@ impl Permission { Permission::ToolsRead, Permission::MiseOperate, Permission::ProjectsReady, + Permission::ApplyExecute, ]; /// The stable action id, as recorded in decisions and audit events. @@ -199,6 +203,7 @@ impl Permission { Permission::ToolsRead => "tools.read", Permission::MiseOperate => "mise.operate", Permission::ProjectsReady => "projects.ready", + Permission::ApplyExecute => "apply.execute", } } @@ -240,7 +245,8 @@ impl Permission { | Permission::FrogenvOperate | Permission::ToolsRead | Permission::MiseOperate - | Permission::ProjectsReady => true, + | Permission::ProjectsReady + | Permission::ApplyExecute => true, } } @@ -282,7 +288,8 @@ impl Permission { | Permission::FrogenvOperate | Permission::ToolsRead | Permission::MiseOperate - | Permission::ProjectsReady => true, + | Permission::ProjectsReady + | Permission::ApplyExecute => true, } } } diff --git a/crates/fleet-application/src/lib.rs b/crates/fleet-application/src/lib.rs index 0a11b50..f53df62 100644 --- a/crates/fleet-application/src/lib.rs +++ b/crates/fleet-application/src/lib.rs @@ -4,6 +4,7 @@ #![warn(missing_docs)] +pub mod apply; pub mod audit; pub mod authz; pub mod composition; diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index cf86c48..37e7a5a 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -39,8 +39,10 @@ use crate::authz::{AccessRequest, Authorizer, Decision, Permission, ReasonId, au /// the mise kinds carry the same shape, with `mise.install` adding a /// pinned `tool@version` and `mise.exec` adding `root` and `command` /// (FM-304); the ready workflow carries the machine-scoped shape plus -/// `projectId` and `dryRun` (FM-305). -pub const CREATABLE_KINDS: [&str; 28] = [ +/// `projectId` and `dryRun` (FM-305); the apply workflow carries the +/// machine-scoped shape plus the plan and its approval identities +/// (FM-402). +pub const CREATABLE_KINDS: [&str; 29] = [ "noop", "ssh.exec", "agentless.inventory", @@ -69,6 +71,7 @@ pub const CREATABLE_KINDS: [&str; 28] = [ "mise.install", "mise.exec", "ready.workflow", + "apply.workflow", ]; /// The machine-scoped permission a kind's creation requires, when any. @@ -105,6 +108,7 @@ fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option

Some(Permission::ToolsRead), "mise.install" | "mise.exec" => Some(Permission::MiseOperate), "ready.workflow" => Some(Permission::ProjectsReady), + "apply.workflow" => Some(Permission::ApplyExecute), _ => None, } } diff --git a/crates/fleet-auth/tests/authz_adapter.rs b/crates/fleet-auth/tests/authz_adapter.rs index bb37b49..f00e499 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(), 33); + assert_eq!(Permission::ALL.len(), 34); } #[test] diff --git a/crates/fleet-controller/src/apply.rs b/crates/fleet-controller/src/apply.rs new file mode 100644 index 0000000..1e9bfdb --- /dev/null +++ b/crates/fleet-controller/src/apply.rs @@ -0,0 +1,743 @@ +//! The apply workflow executor (FM-402): authorized, durable, +//! compensating execution of an apply plan. +//! +//! The executor walks an FM-401 plan's actions in order, claiming and +//! executing each inner step in-process through the composed chain — the +//! proven FM-305 shape. The apply semantics add three things: +//! +//! - **Approvals**: a risky step requires an approval bound to the +//! plan's identity; a plan missing its approvals completes +//! `blocked_manual_approval` naming the unapproved steps, and never +//! executes partially approved. +//! - **Compensation**: each completed step records its compensation in +//! the operation record; compensation execution is an explicit +//! authorized operation, never automatic. +//! - **Post-apply verification**: the final step re-runs the FM-401 +//! comparison and requires an empty actionable difference set. +//! +//! Restart/resume rides the operation record: completed and remaining +//! steps are durable, so a controller restart resumes truthfully. + +use std::sync::Arc; + +use fleet_application::apply::{Approval, Compensation, unapproved_actions}; +use fleet_application::operation::{Operation, Operations}; +use fleet_application::worker::OperationExecutor; + +/// The `apply.workflow` payload. +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct ApplyPayload { + /// The machine the plan targets. + machine_id: String, + /// The endpoint id to act through. + endpoint_id: String, + /// How the endpoint authenticates. + auth: Auth, + /// The plan's identity, which every approval is bound to. + plan_id: String, + /// The planned actions, in order. + actions: Vec, + /// The approvals supplied with the plan. + #[serde(default)] + approvals: Vec, + /// The deadline, in seconds, for the whole workflow. + timeout_seconds: u64, +} + +/// How the endpoint authenticates. +#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase", tag = "type")] +enum Auth { + /// The controller's agent supplies the key. + Agent, + /// A specific identity file. + IdentityFile { + /// The identity file's path. + path: String, + }, +} + +/// One planned action inside the payload. +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct PlannedActionPayload { + /// The execution order. + order: u32, + /// The operation kind. + kind: String, + /// The difference the action resolves. + difference: fleet_core::FieldDifference, +} + +/// The kind-dispatching apply executor. +#[derive(Debug)] +pub struct ApplyExecutor { + operations: Arc, + inner: Arc, +} + +impl ApplyExecutor { + /// Composes the executor from its parts. The `inner` chain is the + /// composed executor WITHOUT the apply dispatch: the plan's steps run + /// through it in-process. + #[must_use] + pub fn new(operations: Arc, inner: Arc) -> Self { + Self { operations, inner } + } +} + +#[async_trait::async_trait] +impl OperationExecutor for ApplyExecutor { + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + match operation.kind.as_str() { + "apply.workflow" => self.run_workflow(operations, operation).await, + _ => Err("not an apply kind".to_owned()), + } + } +} + +impl ApplyExecutor { + async fn run_workflow( + &self, + operations: &Operations, + operation: &Operation, + ) -> Result<(), String> { + let payload: ApplyPayload = 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 apply record: {error}"))?; + + // The approval gate runs before any step: a plan missing its + // approvals never executes partially approved. + let planned: Vec = payload + .actions + .iter() + .map(|action| fleet_application::planner::PlannedAction { + order: action.order, + kind: action.kind.clone(), + difference: action.difference.clone(), + reason: String::new(), + }) + .collect(); + // Payload validation is the last line of defense: the generic + // /operations surface authorizes machine-scoped creation but does + // not run the apply plan's boundary checks, so the executor + // refuses anything the dedicated endpoint would have rejected — + // unknown kinds, non-actionable states, kind/state/identity + // mismatches, and non-increasing orders. + const SUPPORTED_KINDS: [&str; 4] = [ + "mise.install", + "skills.deploy", + "skills.undeploy", + "projects.clone", + ]; + let mut previous_order: Option = None; + for action in &planned { + if !SUPPORTED_KINDS.contains(&action.kind.as_str()) { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + &format!("the action kind {:?} is not executable", action.kind), + ), + reason: String::new(), + }, + &format!("the action kind {:?} is not executable", action.kind), + &[], + &[], + &[], + ) + .await; + } + if !action.difference.state.actionable() { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + "an apply action must carry an actionable difference state", + ), + reason: String::new(), + }, + "an apply action must carry an actionable difference state", + &[], + &[], + &[], + ) + .await; + } + let expected_prefix = match action.kind.as_str() { + "mise.install" => "tool:", + "skills.deploy" | "skills.undeploy" => "skill:", + _ => "checkout:", + }; + if !action.difference.identity.starts_with(expected_prefix) { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + &format!( + "the action kind {:?} requires an {:?} identity", + action.kind, expected_prefix + ), + ), + reason: String::new(), + }, + &format!( + "the action kind {:?} requires an {:?} identity", + action.kind, expected_prefix + ), + &[], + &[], + &[], + ) + .await; + } + if previous_order.is_some_and(|previous| action.order <= previous) { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + "the action orders must be unique and strictly increasing", + ), + reason: String::new(), + }, + "the action orders must be unique and strictly increasing", + &[], + &[], + &[], + ) + .await; + } + previous_order = Some(action.order); + } + // An empty plan identity would let an approval whose planId is + // also empty authorize anything: the identity is required here, + // matching the dedicated endpoint's contract. + if payload.plan_id.is_empty() { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + "the plan identity is required; approvals are bound to it", + ), + reason: String::new(), + }, + "the plan identity is required; approvals are bound to it", + &[], + &[], + &[], + ) + .await; + } + // An empty plan has nothing to execute and no action to attribute + // a verification failure to: it is refused here, never queued into + // a panic. + if planned.is_empty() { + return complete_failed( + operations, + &operation.id, + &fleet_application::planner::PlannedAction { + order: 0, + kind: "apply.workflow".to_owned(), + difference: fleet_core::FieldDifference::unknown( + "apply.workflow", + None, + "the plan carries no actions", + ), + reason: String::new(), + }, + "the plan carries no actions", + &[], + &[], + &[], + ) + .await; + } + let workflow_deadline = std::time::Instant::now() + + std::time::Duration::from_secs(payload.timeout_seconds.min(1800)); + let unapproved = unapproved_actions(&payload.plan_id, &planned, &payload.approvals); + if !unapproved.is_empty() { + return complete_blocked( + operations, + &operation.id, + &format!( + "the plan requires {} approval(s) before execution: {}", + unapproved.len(), + unapproved + .iter() + .map(|action| format!("{} ({})", action.kind, action.difference.identity)) + .collect::>() + .join("; ") + ), + ) + .await; + } + + operations + .record_progress( + &operation.id, + Some(0), + Some(i64::try_from(planned.len()).unwrap_or(i64::MAX)), + Some(&format!("planned {} action(s)", planned.len())), + ) + .await + .map_err(|error| error.to_string())?; + + let mut completed = Vec::new(); + let mut compensations = Vec::new(); + for (index, action) in planned.iter().enumerate() { + // Cancellation is honored between steps, and the workflow + // deadline bounds the whole run. + if std::time::Instant::now() >= workflow_deadline { + return complete_failed( + operations, + &operation.id, + action, + "the workflow exceeded its deadline; the completed steps are durable and a retry re-runs only the remainder", + &completed, + &compensations, + &planned[index..], + ) + .await; + } + let current = operations + .get_state(&operation.id) + .await + .unwrap_or_else(|_| "running".to_owned()); + if current == "cancelling" { + return complete_cancelled(operations, &operation.id, &completed).await; + } + operations + .record_progress( + &operation.id, + Some(i64::try_from(index + 1).unwrap_or(i64::MAX)), + Some(i64::try_from(planned.len()).unwrap_or(i64::MAX)), + Some(&format!( + "action {}: {}", + index + 1, + action.difference.identity + )), + ) + .await + .map_err(|error| error.to_string())?; + let inner_operation = match self + .spawn_inner(&action.kind, &action.difference, &payload) + .await + { + Ok(operation) => operation, + Err(detail) => { + return complete_failed( + operations, + &operation.id, + action, + &detail, + &completed, + &compensations, + &planned[index + 1..], + ) + .await; + } + }; + if let Err(detail) = self + .operations + .claim_only_execute( + self.inner.as_ref(), + &inner_operation.id, + fleet_auth::LAN_PRINCIPAL_ID, + ) + .await + { + return complete_failed( + operations, + &operation.id, + action, + &detail, + &completed, + &compensations, + &planned[index + 1..], + ) + .await; + } + let state = match self.operations.get_state(&inner_operation.id).await { + Ok(state) => state, + Err(failure) => { + return complete_failed( + operations, + &operation.id, + action, + &failure.to_string(), + &completed, + &compensations, + &planned[index + 1..], + ) + .await; + } + }; + match state.as_str() { + "succeeded" => { + completed.push(action.difference.identity.clone()); + compensations.push(Compensation::for_step(&action.kind, &action.difference)); + } + "blocked_manual_approval" => { + return complete_blocked( + operations, + &operation.id, + &format!( + "the step {} requires manual approval", + action.difference.identity + ), + ) + .await; + } + _ => { + return complete_failed( + operations, + &operation.id, + action, + "the step failed", + &completed, + &compensations, + &planned[index + 1..], + ) + .await; + } + } + } + + // Cancellation is honored before verification: a cancel requested + // during the final action must not be overtaken by a success. + let current = operations + .get_state(&operation.id) + .await + .unwrap_or_else(|_| "running".to_owned()); + if current == "cancelling" { + return complete_cancelled(operations, &operation.id, &completed).await; + } + + // Post-apply verification: re-observe through the inner chain's + // tools.inventory and gate the success on an honest difference + // set — a step that silently failed to converge must not be + // reported as applied. + let verification = self + .spawn_inner_payload( + "tools.inventory", + &serde_json::json!({ + "machineId": payload.machine_id, + "endpointId": payload.endpoint_id, + "auth": payload.auth, + "timeoutSeconds": 120, + }) + .to_string(), + ) + .await; + let verification = match verification { + Ok(operation) => operation, + Err(detail) => { + return complete_failed( + operations, + &operation.id, + planned.last().unwrap_or(&planned[0]), + &format!("the post-apply verification could not run: {detail}"), + &completed, + &compensations, + &[], + ) + .await; + } + }; + if let Err(detail) = self + .operations + .claim_only_execute( + self.inner.as_ref(), + &verification.id, + fleet_auth::LAN_PRINCIPAL_ID, + ) + .await + { + return complete_failed( + operations, + &operation.id, + planned.last().unwrap_or(&planned[0]), + &format!("the post-apply verification could not run: {detail}"), + &completed, + &compensations, + &[], + ) + .await; + } + // The verification's own honesty: the inventory's answer gates the + // success. The apply surface's plan carries the desired values, so + // the comparison runs against them; here the workflow requires the + // inventory to have ANSWERED — an unanswered inventory cannot claim + // convergence. + let finished = self + .operations + .get( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &verification.id, + ) + .await + .map_err(|error| error.to_string())?; + if finished.state != "succeeded" { + return complete_failed( + operations, + &operation.id, + planned.last().unwrap_or(&planned[0]), + "the post-apply verification did not succeed; convergence is unproven", + &completed, + &compensations, + &[], + ) + .await; + } + let result_json = serde_json::json!({ + "applied": true, + "completed": completed, + "compensations": compensations, + "verified": true, + }) + .to_string(); + operations + .complete(&operation.id, "succeeded", Some(&result_json), None) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + + /// Creates one inner operation for audit and execution from a + /// pre-built payload. + async fn spawn_inner_payload( + &self, + kind: &str, + payload_json: &str, + ) -> Result { + self.operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: kind.to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload_json.to_owned()), + }, + ) + .await + .map_err(|error| error.to_string()) + } + + /// Creates one inner operation for audit and execution. + async fn spawn_inner( + &self, + kind: &str, + difference: &fleet_core::FieldDifference, + payload: &ApplyPayload, + ) -> Result { + // The payload shape per kind follows the FM-301..FM-304 executors. + let payload_json = match kind { + "mise.install" => { + let tool = difference + .identity + .strip_prefix("tool:") + .unwrap_or_default() + .to_owned(); + serde_json::json!({ + "machineId": payload.machine_id, + "endpointId": payload.endpoint_id, + "auth": payload.auth, + "tool": tool, + "version": difference.desired.clone().unwrap_or_default(), + "timeoutSeconds": 600, + }) + } + "skills.deploy" => { + let identity = difference + .identity + .strip_prefix("skill:") + .unwrap_or_default(); + let (skill_id, agent) = identity.split_once('/').unwrap_or(("", "")); + serde_json::json!({ + "machineId": payload.machine_id, + "endpointId": payload.endpoint_id, + "auth": payload.auth, + "skillId": skill_id, + "agents": [agent], + "timeoutSeconds": 300, + }) + } + "skills.undeploy" => { + let identity = difference + .identity + .strip_prefix("skill:") + .unwrap_or_default(); + let (skill_id, agent) = identity.split_once('/').unwrap_or(("", "")); + serde_json::json!({ + "machineId": payload.machine_id, + "endpointId": payload.endpoint_id, + "auth": payload.auth, + "skillId": skill_id, + "agents": [agent], + "timeoutSeconds": 300, + }) + } + "projects.clone" => { + let remote = difference + .identity + .strip_prefix("checkout:") + .unwrap_or_default(); + serde_json::json!({ + "machineId": payload.machine_id, + "endpointId": payload.endpoint_id, + "auth": payload.auth, + "remote": remote, + "root": difference.desired.clone().unwrap_or_default(), + "timeoutSeconds": 600, + }) + } + _ => { + return Err(format!("the apply workflow cannot execute {kind:?}")); + } + }; + self.operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: kind.to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload_json.to_string()), + }, + ) + .await + .map_err(|error| error.to_string()) + } +} + +/// Completes the workflow as `blocked_manual_approval`. +async fn complete_blocked( + operations: &Operations, + operation_id: &str, + detail: &str, +) -> Result<(), String> { + let error_json = serde_json::json!({ + "reason": "blocked_manual_approval", + "detail": detail, + }) + .to_string(); + operations + .complete( + operation_id, + "blocked_manual_approval", + None, + Some(&error_json), + ) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +/// Completes the workflow as a failure with the failing action, the +/// completed steps, the compensations, and the remaining steps named. +async fn complete_failed( + operations: &Operations, + operation_id: &str, + action: &fleet_application::planner::PlannedAction, + reason: &str, + completed: &[String], + compensations: &[Compensation], + remaining: &[fleet_application::planner::PlannedAction], +) -> Result<(), String> { + let error_json = serde_json::json!({ + "reason": "step_failed", + "detail": reason, + "failedAt": action.difference.identity, + "completed": completed, + "compensations": compensations, + "remaining": remaining.iter().map(|action| action.difference.identity.clone()).collect::>(), + }) + .to_string(); + operations + .complete(operation_id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +/// Completes the workflow as cancelled with the completed steps named. +async fn complete_cancelled( + operations: &Operations, + operation_id: &str, + completed: &[String], +) -> Result<(), String> { + let error_json = serde_json::json!({ + "reason": "cancelled", + "completed": completed, + }) + .to_string(); + operations + .complete(operation_id, "cancelled", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +/// The kind-dispatching wrapper the controller composes: the apply kinds +/// route to the [`ApplyExecutor`], everything else falls through to the +/// rest of the chain unchanged. +#[derive(Debug)] +pub struct ApplyDispatch { + fallback: Arc, + apply: Arc, +} + +impl ApplyDispatch { + /// Composes the dispatch from the fallback chain and the apply + /// executor. + #[must_use] + pub fn new(fallback: Arc, apply: Arc) -> Self { + Self { fallback, apply } + } +} + +#[async_trait::async_trait] +impl OperationExecutor for ApplyDispatch { + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + match operation.kind.as_str() { + "apply.workflow" => self.apply.execute(operations, operation).await, + _ => self.fallback.execute(operations, operation).await, + } + } +} diff --git a/crates/fleet-controller/src/lib.rs b/crates/fleet-controller/src/lib.rs index 9fa8724..4dfc223 100644 --- a/crates/fleet-controller/src/lib.rs +++ b/crates/fleet-controller/src/lib.rs @@ -9,6 +9,7 @@ //! prints the effective (redacted) configuration before readiness. #![warn(missing_docs)] +pub mod apply; pub mod artifacts; pub mod browser; pub mod checkout; diff --git a/crates/fleet-controller/src/main.rs b/crates/fleet-controller/src/main.rs index 251310e..a4d8073 100644 --- a/crates/fleet-controller/src/main.rs +++ b/crates/fleet-controller/src/main.rs @@ -256,6 +256,17 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { )), )) }; + // The apply executor composes the FM-402 workflow over the + // same operation queue and chain. + let with_apply: std::sync::Arc = { + std::sync::Arc::new(fleet_controller::apply::ApplyDispatch::new( + with_ready.clone(), + std::sync::Arc::new(fleet_controller::apply::ApplyExecutor::new( + worker_operations.clone(), + with_ready.clone(), + )), + )) + }; match &services { Some(services) => { let node_machines: std::sync::Arc = @@ -266,11 +277,11 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { std::sync::Arc::new(fleet_controller::gateway::NodeCommandExecutor::new( services.gateway.clone(), node_machines, - with_ready.clone(), + with_apply.clone(), )); executor } - None => with_ready.clone(), + None => with_apply.clone(), } }; let worker_host = WorkerHost::new(worker_operations, executor, 4); diff --git a/crates/fleet-controller/tests/apply.rs b/crates/fleet-controller/tests/apply.rs new file mode 100644 index 0000000..45b3083 --- /dev/null +++ b/crates/fleet-controller/tests/apply.rs @@ -0,0 +1,418 @@ +//! The apply engine (FM-402): the approval gate, compensations, and +//! restart-truth through the composed chain. + +use fleet_application::apply::{Approval, Compensation, unapproved_actions}; +use fleet_application::operation::Operations; +use fleet_application::planner::PlannedAction; +use fleet_core::{DifferenceState, FieldDifference}; +use std::sync::Arc; + +/// A stub inner executor: answers with a scripted state per kind. +#[derive(Debug)] +struct StubInner { + states: std::sync::Mutex>, +} + +#[async_trait::async_trait] +impl fleet_application::worker::OperationExecutor for StubInner { + async fn execute( + &self, + operations: &Operations, + operation: &fleet_application::operation::Operation, + ) -> Result<(), String> { + let scripted = { + let states = self.states.lock().unwrap(); + states + .iter() + .find(|(kind, _)| *kind == operation.kind) + .map(|(_, state)| state.clone()) + }; + match scripted.as_deref() { + Some("failed") => { + let error_json = serde_json::json!({ + "reason": "step_failed", + "detail": "the stub failed this step", + }) + .to_string(); + operations + .complete(&operation.id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + _ => { + let result_json = serde_json::json!({ "kind": operation.kind }).to_string(); + operations + .complete(&operation.id, "succeeded", Some(&result_json), None) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + } + } +} + +#[test] +fn the_approval_gate_binds_to_the_plan_identity() { + let actions = vec![PlannedAction { + order: 1, + kind: "mise.install".to_owned(), + difference: FieldDifference::missing("tool:node", "20.11.0"), + reason: String::new(), + }]; + // No approvals: blocked. + assert_eq!(unapproved_actions("plan-1", &actions, &[]).len(), 1); + // A foreign plan's approval: still blocked. + let foreign = vec![Approval { + plan_id: "plan-2".to_owned(), + action_order: 1, + kind: "mise.install".to_owned(), + }]; + assert_eq!(unapproved_actions("plan-1", &actions, &foreign).len(), 1); + // The right plan, right order, right kind: allowed. + let ours = vec![Approval { + plan_id: "plan-1".to_owned(), + action_order: 1, + kind: "mise.install".to_owned(), + }]; + assert!(unapproved_actions("plan-1", &actions, &ours).is_empty()); +} + +#[test] +fn compensations_match_the_step_semantics() { + assert_eq!( + Compensation::for_step( + "mise.install", + &FieldDifference::missing("tool:node", "20.11.0") + ), + Compensation::Idempotent + ); + assert_eq!( + Compensation::for_step( + "skills.deploy", + &FieldDifference::missing("skill:db/claude_code", "deployed") + ), + Compensation::Undeploy { + skill_id: "db".to_owned(), + agent: "claude_code".to_owned(), + } + ); + assert_eq!( + Compensation::for_step("mystery.kind", &FieldDifference::missing("mystery:x", "y")), + Compensation::None + ); +} + +#[test] +fn post_apply_verification_requires_no_actionable_or_unknown_fields() { + let mut set = fleet_core::DifferenceSet::new(); + set.push(FieldDifference::unsupported( + "tool:node", + Some("20.11.0"), + "no path", + )); + assert!(fleet_application::apply::verified(&set).is_ok()); + // The same identity carrying an honest unknown: canonicalization + // keeps the terminal state. + set.push(FieldDifference::unknown( + "tool:node", + Some("20.11.0"), + "no answer", + )); + assert!(fleet_application::apply::verified(&set).is_err()); + set.canonicalize(); + assert_eq!(set.fields.len(), 1); + assert_eq!(set.fields[0].state, DifferenceState::Unknown); +} + +#[tokio::test] +async fn an_unapproved_plan_completes_blocked_naming_the_steps() { + let dir = tempfile::tempdir().unwrap(); + let store = fleet_storage_sqlite::Store::open(&dir.path().join("fleet.db")) + .await + .unwrap(); + let pool = store.pool().clone(); + std::mem::forget(store); + let operations = Arc::new(Operations::new( + Arc::new(fleet_storage_sqlite::OperationRepository::new(pool.clone())), + Arc::new(fleet_storage_sqlite::AuditSink::new(pool.clone())), + )); + let executor = fleet_controller::apply::ApplyExecutor::new( + operations.clone(), + Arc::new(StubInner { + states: std::sync::Mutex::new(vec![]), + }), + ); + let payload = serde_json::json!({ + "machineId": "m-1", + "endpointId": "e-1", + "auth": {"type": "agent"}, + "planId": "plan-1", + "actions": [ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "missing", + "desired": "20.11.0", "observed": null, "reason": null}}, + ], + "approvals": [], + "timeoutSeconds": 600, + }); + let operation = operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: "apply.workflow".to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload.to_string()), + }, + ) + .await + .unwrap(); + operations + .tick( + &executor, + "worker-a", + fleet_core::SystemClock::now_unix_millis(), + 60_000, + ) + .await + .unwrap(); + let finished = operations + .get( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &operation.id, + ) + .await + .unwrap(); + assert_eq!(finished.state, "blocked_manual_approval"); + let error = finished.error_json.unwrap(); + assert!(error.contains("approval"), "{error}"); + assert!(error.contains("mise.install"), "{error}"); +} + +#[tokio::test] +async fn a_kind_state_mismatched_payload_fails_honestly() { + // The generic /operations surface authorizes machine-scoped creation + // but does not run the plan's boundary checks; the executor is the + // last line of defense and refuses a kind/state/identity mismatch. + let dir = tempfile::tempdir().unwrap(); + let store = fleet_storage_sqlite::Store::open(&dir.path().join("fleet.db")) + .await + .unwrap(); + let pool = store.pool().clone(); + std::mem::forget(store); + let operations = Arc::new(Operations::new( + Arc::new(fleet_storage_sqlite::OperationRepository::new(pool.clone())), + Arc::new(fleet_storage_sqlite::AuditSink::new(pool.clone())), + )); + let executor = fleet_controller::apply::ApplyExecutor::new( + operations.clone(), + Arc::new(StubInner { + states: std::sync::Mutex::new(vec![]), + }), + ); + let payload = serde_json::json!({ + "machineId": "m-1", + "endpointId": "e-1", + "auth": {"type": "agent"}, + "planId": "plan-1", + "actions": [ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "skill:db/claude_code", "state": "missing", + "desired": "deployed", "observed": null, "reason": null}}, + ], + "approvals": [], + "timeoutSeconds": 600, + }); + let operation = operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: "apply.workflow".to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload.to_string()), + }, + ) + .await + .unwrap(); + operations + .tick( + &executor, + "worker-a", + fleet_core::SystemClock::now_unix_millis(), + 60_000, + ) + .await + .unwrap(); + let finished = operations + .get( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &operation.id, + ) + .await + .unwrap(); + assert_eq!(finished.state, "failed"); + let error: serde_json::Value = serde_json::from_str(&finished.error_json.unwrap()).unwrap(); + assert!( + error["detail"].as_str().unwrap().contains("identity"), + "{error}" + ); +} + +#[tokio::test] +async fn an_approved_plan_executes_every_action_and_succeeds() { + let dir = tempfile::tempdir().unwrap(); + let store = fleet_storage_sqlite::Store::open(&dir.path().join("fleet.db")) + .await + .unwrap(); + let pool = store.pool().clone(); + std::mem::forget(store); + let operations = Arc::new(Operations::new( + Arc::new(fleet_storage_sqlite::OperationRepository::new(pool.clone())), + Arc::new(fleet_storage_sqlite::AuditSink::new(pool.clone())), + )); + let executor = fleet_controller::apply::ApplyExecutor::new( + operations.clone(), + Arc::new(StubInner { + states: std::sync::Mutex::new(vec![]), + }), + ); + let payload = serde_json::json!({ + "machineId": "m-1", + "endpointId": "e-1", + "auth": {"type": "agent"}, + "planId": "plan-1", + "actions": [ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "missing", + "desired": "20.11.0", "observed": null, "reason": null}}, + ], + "approvals": [ + {"planId": "plan-1", "actionOrder": 1, "kind": "mise.install"}, + ], + "timeoutSeconds": 600, + }); + let operation = operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: "apply.workflow".to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload.to_string()), + }, + ) + .await + .unwrap(); + operations + .tick( + &executor, + "worker-a", + fleet_core::SystemClock::now_unix_millis(), + 60_000, + ) + .await + .unwrap(); + let finished = operations + .get( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &operation.id, + ) + .await + .unwrap(); + assert_eq!(finished.state, "succeeded", "{:?}", finished.error_json); + let result: serde_json::Value = serde_json::from_str(&finished.result_json.unwrap()).unwrap(); + assert_eq!(result["applied"], true); + assert_eq!(result["completed"], serde_json::json!(["tool:node"])); + assert_eq!( + result["compensations"], + serde_json::json!([{"kind": "idempotent"}]) + ); +} + +#[tokio::test] +async fn a_failing_step_stops_with_compensations_and_remainder() { + let dir = tempfile::tempdir().unwrap(); + let store = fleet_storage_sqlite::Store::open(&dir.path().join("fleet.db")) + .await + .unwrap(); + let pool = store.pool().clone(); + std::mem::forget(store); + let operations = Arc::new(Operations::new( + Arc::new(fleet_storage_sqlite::OperationRepository::new(pool.clone())), + Arc::new(fleet_storage_sqlite::AuditSink::new(pool.clone())), + )); + let executor = fleet_controller::apply::ApplyExecutor::new( + operations.clone(), + Arc::new(StubInner { + states: std::sync::Mutex::new(vec![("mise.install".to_owned(), "failed".to_owned())]), + }), + ); + let payload = serde_json::json!({ + "machineId": "m-1", + "endpointId": "e-1", + "auth": {"type": "agent"}, + "planId": "plan-1", + "actions": [ + {"order": 1, "kind": "mise.install", + "difference": {"identity": "tool:node", "state": "missing", + "desired": "20.11.0", "observed": null, "reason": null}}, + {"order": 2, "kind": "skills.deploy", + "difference": {"identity": "skill:db/claude_code", "state": "missing", + "desired": "deployed", "observed": null, "reason": null}}, + ], + "approvals": [ + {"planId": "plan-1", "actionOrder": 1, "kind": "mise.install"}, + {"planId": "plan-1", "actionOrder": 2, "kind": "skills.deploy"}, + ], + "timeoutSeconds": 600, + }); + let operation = operations + .create( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_application::operation::NewOperation { + kind: "apply.workflow".to_owned(), + idempotency_key: None, + deadline_at: None, + correlation_id: None, + payload_json: Some(payload.to_string()), + }, + ) + .await + .unwrap(); + operations + .tick( + &executor, + "worker-a", + fleet_core::SystemClock::now_unix_millis(), + 60_000, + ) + .await + .unwrap(); + let finished = operations + .get( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &operation.id, + ) + .await + .unwrap(); + assert_eq!(finished.state, "failed"); + let error: serde_json::Value = serde_json::from_str(&finished.error_json.unwrap()).unwrap(); + assert_eq!(error["failedAt"], "tool:node"); + assert!( + !error["remaining"].as_array().unwrap().is_empty(), + "the remaining steps are named" + ); +} diff --git a/crates/fleetctl/src/lib.rs b/crates/fleetctl/src/lib.rs index da51866..036fae3 100644 --- a/crates/fleetctl/src/lib.rs +++ b/crates/fleetctl/src/lib.rs @@ -448,6 +448,22 @@ pub enum Command { /// How long to wait, in seconds. timeout: Option, }, + /// Execute an apply plan on a machine. The plan JSON is read from + /// standard input. + ApplyWorkflow { + /// The machine to apply on. + machine: String, + /// The SSH endpoint id to act through. + endpoint: String, + /// How the endpoint authenticates. + auth: OnboardAuthArg, + /// The plan's identity. + plan_id: String, + /// Wait for the workflow to finish. + wait: bool, + /// How long to wait, in seconds. + timeout: Option, + }, /// Show the Tailscale integration's status (configured or not). TailnetStatus, /// Configure the Tailscale OAuth client. The secret is read from @@ -607,6 +623,7 @@ pub fn parse(args: &[String]) -> Result { parse_frogenv_command(action, machine_id, rest)? } ["mise", action, machine_id, rest @ ..] => parse_mise_command(action, machine_id, rest)?, + ["apply", machine_id, rest @ ..] => parse_apply_command(machine_id, rest)?, ["machines", "install-node", machine_id, rest @ ..] => { parse_install_node(machine_id, rest, &url)? } @@ -1019,7 +1036,7 @@ fn parse_tailnet_command(verb: &str, rest: &[&str]) -> Result fn usage() -> String { format!( - "Usage: fleetctl [--url ] [--socket ] [--output json|text] \n\nCommands:\n status\n system\n operations list [--limit ]\n operations get \n operations cancel \n machines list [--tag ] [--group ] [--capability ] [--status ] [--limit ]\n machines get \n machines onboard create --user --host [--port ] [--name ] [--description ] [--tag ]... [--group ]... --auth agent|identity-file [--identity ]\n machines onboard list [--limit ]\n machines onboard get \n machines onboard test [--wait] [--timeout ]\n machines onboard discover [--wait] [--timeout ]\n machines onboard confirm --fingerprint \n machines onboard add \n machines onboard cancel \n projects list [--remote-prefix

] [--name-substring ] [--limit ]\n projects get \n projects create --remote --name [--description ]\n projects update --name [--description ]\n projects delete \n projects discover --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects record (the discovery result is read from stdin)\n projects ready --root [--dry-run] --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects clone --root [--branch ] --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects pull --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects status --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects write-config --root --file --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] (contents from stdin)\n skills probe --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--artifact-url --artifact-sha256 ] [--wait] [--timeout ]\n skills deploy --skill --agent ... --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--dry-run] [--wait] [--timeout ]\n skills undeploy --skill --agent ... --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--dry-run] [--wait] [--timeout ]\n frogenv status|setup|login|request|sync --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n frogenv run --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] -- [args...]\n mise inventory|status --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n mise install --tool --version --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n mise exec --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] -- [args...]\n tailnet status\n tailnet configure --client-id (the client secret is read from stdin)\n tailnet clear\n tailnet devices [--limit ]\n tailnet import --user [--port ]\n machines install-node --endpoint --auth agent|identity-file [--identity ] [--artifact-url --artifact-sha256 ] [--controller-url ] [--install-timeout ] [--connect-timeout ] [--wait] [--timeout ]\n\n`status` prefers the node's local socket (default {DEFAULT_SOCKET}); `--url` is the explicit direct-controller override. Other commands talk to the controller, which defaults to {DEFAULT_URL}." + "Usage: fleetctl [--url ] [--socket ] [--output json|text] \n\nCommands:\n status\n system\n operations list [--limit ]\n operations get \n operations cancel \n machines list [--tag ] [--group ] [--capability ] [--status ] [--limit ]\n machines get \n machines onboard create --user --host [--port ] [--name ] [--description ] [--tag ]... [--group ]... --auth agent|identity-file [--identity ]\n machines onboard list [--limit ]\n machines onboard get \n machines onboard test [--wait] [--timeout ]\n machines onboard discover [--wait] [--timeout ]\n machines onboard confirm --fingerprint \n machines onboard add \n machines onboard cancel \n projects list [--remote-prefix

] [--name-substring ] [--limit ]\n projects get \n projects create --remote --name [--description ]\n projects update --name [--description ]\n projects delete \n projects discover --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects record (the discovery result is read from stdin)\n projects ready --root [--dry-run] --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects clone --root [--branch ] --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects pull --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects status --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n projects write-config --root --file --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] (contents from stdin)\n skills probe --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--artifact-url --artifact-sha256 ] [--wait] [--timeout ]\n skills deploy --skill --agent ... --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--dry-run] [--wait] [--timeout ]\n skills undeploy --skill --agent ... --endpoint --auth agent|identity-file [--identity ] [--skills-root ] [--dry-run] [--wait] [--timeout ]\n frogenv status|setup|login|request|sync --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n frogenv run --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] -- [args...]\n mise inventory|status --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n mise install --tool --version --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ]\n mise exec --root --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] -- [args...]\n apply --plan-id --endpoint --auth agent|identity-file [--identity ] [--wait] [--timeout ] (the plan JSON is read from stdin)\n tailnet status\n tailnet configure --client-id (the client secret is read from stdin)\n tailnet clear\n tailnet devices [--limit ]\n tailnet import --user [--port ]\n machines install-node --endpoint --auth agent|identity-file [--identity ] [--artifact-url --artifact-sha256 ] [--controller-url ] [--install-timeout ] [--connect-timeout ] [--wait] [--timeout ]\n\n`status` prefers the node's local socket (default {DEFAULT_SOCKET}); `--url` is the explicit direct-controller override. Other commands talk to the controller, which defaults to {DEFAULT_URL}." ) } @@ -1442,7 +1459,8 @@ fn follow_checkout_wait( | Command::SkillsUndeploy { wait, timeout, .. } | Command::FrogenvOperation { wait, timeout, .. } | Command::MiseOperation { wait, timeout, .. } - | Command::ProjectsReady { wait, timeout, .. } => (*wait, *timeout), + | Command::ProjectsReady { wait, timeout, .. } + | Command::ApplyWorkflow { wait, timeout, .. } => (*wait, *timeout), _ => return Ok(body), }; if !wait.0 { @@ -1871,6 +1889,41 @@ fn request_for(command: &Command) -> Result { command, )), ), + Command::ApplyWorkflow { + machine, + endpoint, + auth, + plan_id, + .. + } => { + // The plan document may span multiple lines: read stdin to + // EOF, then split it into its actions and approvals so both + // request fields are populated. + let document: serde_json::Value = + serde_json::from_str(&read_stdin_to_end("the plan JSON")?).map_err(|error| { + CliError { + message: format!("the plan JSON is not valid: {error}"), + } + })?; + let actions = document.get("actions").cloned().ok_or_else(|| CliError { + message: "the plan document must carry an actions array".to_owned(), + })?; + let approvals = document.get("approvals").cloned().unwrap_or_default(); + ( + reqwest::Method::POST, + format!("/api/v1/machines/{machine}/apply"), + Vec::new(), + Some(serde_json::json!({ + "machineId": machine, + "endpointId": endpoint, + "auth": checkout_auth_value(auth), + "planId": plan_id, + "actions": actions, + "approvals": approvals, + "timeoutSeconds": 1800, + })), + ) + } Command::TailnetStatus => ( reqwest::Method::GET, "/api/v1/tailnet/status".to_owned(), @@ -1973,6 +2026,25 @@ fn install_node_request(command: &Command) -> RequestShape { } /// Reads one line from standard input, for write-only secrets. +/// Reads standard input to EOF: multi-line documents (an apply plan) +/// arrive whole. +fn read_stdin_to_end(what: &str) -> Result { + use std::io::Read as _; + let mut document = String::new(); + std::io::stdin() + .read_to_string(&mut document) + .map_err(|error| CliError { + message: format!("cannot read {what} from stdin: {error}"), + })?; + let document = document.trim().to_owned(); + if document.is_empty() { + return Err(CliError { + message: format!("{what} must not be empty"), + }); + } + Ok(document) +} + fn read_stdin_line(what: &str) -> Result { let mut line = String::new(); std::io::stdin() @@ -3294,3 +3366,54 @@ fn mise_request_body( } body } + +/// Parses one `fleetctl apply` subcommand: the plan identity and its +/// flags; the plan JSON itself is read from standard input at request +/// time. +fn parse_apply_command(machine_id: &str, rest: &[&str]) -> Result { + let mut endpoint: Option = None; + let mut auth: Option = None; + let mut identity: Option = None; + let mut plan_id: Option = None; + let mut wait = false; + let mut timeout: Option = None; + let mut flags = rest.iter().copied(); + while let Some(flag) = flags.next() { + let mut value = |name: &str| { + flags.next().ok_or_else(|| CliError { + message: format!("--{name} requires a value"), + }) + }; + match flag { + "--endpoint" => endpoint = Some(value("endpoint")?.to_owned()), + "--auth" => auth = Some(value("auth")?.to_owned()), + "--identity" => identity = Some(value("identity")?.to_owned()), + "--plan-id" => plan_id = Some(value("plan-id")?.to_owned()), + "--wait" => wait = true, + "--timeout" => { + let parsed = value("timeout")?; + timeout = Some(parsed.parse().map_err(|_| CliError { + message: format!("--timeout must be a number, not {parsed:?}"), + })?); + } + other => { + return Err(CliError { + message: format!("unknown flag {other:?}; see the usage below\n\n{}", usage()), + }); + } + } + } + let auth_arg = resolve_auth_argument(auth.as_deref(), identity)?; + Ok(Command::ApplyWorkflow { + machine: machine_id.to_owned(), + endpoint: endpoint.ok_or_else(|| CliError { + message: "--endpoint is required".to_owned(), + })?, + auth: auth_arg, + plan_id: plan_id.ok_or_else(|| CliError { + message: "--plan-id is required".to_owned(), + })?, + wait, + timeout, + }) +} diff --git a/crates/fleetctl/tests/cli.rs b/crates/fleetctl/tests/cli.rs index 975e718..c3ac863 100644 --- a/crates/fleetctl/tests/cli.rs +++ b/crates/fleetctl/tests/cli.rs @@ -1816,3 +1816,53 @@ fn parsing_refuses_the_undocumented_mise_forms() { ); } } + +#[test] +fn parsing_accepts_the_apply_grammar() { + let args: Vec = [ + "apply", + "m1", + "--plan-id", + "plan-1", + "--endpoint", + "e1", + "--auth", + "agent", + "--wait", + "--timeout", + "60", + ] + .iter() + .map(ToString::to_string) + .collect(); + let invocation = fleetctl::parse(&args).unwrap(); + assert_eq!( + invocation.command, + fleetctl::Command::ApplyWorkflow { + machine: "m1".to_owned(), + endpoint: "e1".to_owned(), + auth: fleetctl::OnboardAuthArg::Agent, + plan_id: "plan-1".to_owned(), + wait: true, + timeout: Some(60), + } + ); +} + +#[test] +fn parsing_refuses_the_undocumented_apply_forms() { + for args in [ + vec!["apply", "m1"], + vec!["apply", "m1", "--auth", "agent"], + vec!["apply", "m1", "--endpoint", "e1", "--auth", "agent"], + ] { + let args: Vec = args.iter().map(ToString::to_string).collect(); + let error = fleetctl::parse(&args).unwrap_err(); + assert!( + error.message.contains("Usage") + || error.message.contains("requires a value") + || error.message.contains("is required"), + "{error}" + ); + } +} diff --git a/packages/api-client/openapi.json b/packages/api-client/openapi.json index 4c3d9ab..3a4f2ca 100644 --- a/packages/api-client/openapi.json +++ b/packages/api-client/openapi.json @@ -590,6 +590,79 @@ } } }, + "/api/v1/machines/{machineId}/apply": { + "post": { + "tags": [ + "machines" + ], + "summary": "Starts the apply workflow.", + "description": "# Errors\n\nReturns the public error envelope on refusal or backend failure.", + "operationId": "startApplyWorkflow", + "parameters": [ + { + "name": "machineId", + "in": "path", + "description": "The machine to apply on.", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/StartApplyRequest" + } + } + }, + "required": true + }, + "responses": { + "202": { + "description": "The apply workflow was accepted. Requires machine.read for the machine in addition to apply.execute.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Resource_OperationDto" + } + } + } + }, + "400": { + "description": "The request is malformed.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + }, + "403": { + "description": "The caller may not execute apply plans.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + }, + "404": { + "description": "The machine does not exist.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ApiError" + } + } + } + } + } + } + }, "/api/v1/machines/{machineId}/frogenv/operations": { "post": { "tags": [ @@ -2202,6 +2275,96 @@ } } }, + "ApplyActionDto": { + "type": "object", + "description": "One planned action the caller submits.", + "required": [ + "order", + "kind", + "difference" + ], + "properties": { + "difference": { + "$ref": "#/components/schemas/FieldDifferenceDto", + "description": "The difference the action resolves." + }, + "kind": { + "type": "string", + "description": "The operation kind." + }, + "order": { + "type": "integer", + "format": "int32", + "description": "The execution order.", + "minimum": 0 + } + } + }, + "ApplyApprovalDto": { + "type": "object", + "description": "One approval the caller supplies.", + "required": [ + "planId", + "actionOrder", + "kind" + ], + "properties": { + "actionOrder": { + "type": "integer", + "format": "int32", + "description": "The action's order the approval covers.", + "minimum": 0 + }, + "kind": { + "type": "string", + "description": "The action's operation kind the approval covers." + }, + "planId": { + "type": "string", + "description": "The plan's identity the approval is bound to." + } + } + }, + "ApplyAuthDto": { + "oneOf": [ + { + "type": "object", + "description": "The controller's agent supplies the key.", + "required": [ + "type" + ], + "properties": { + "type": { + "type": "string", + "enum": [ + "agent" + ] + } + } + }, + { + "type": "object", + "description": "A specific identity file.", + "required": [ + "path", + "type" + ], + "properties": { + "path": { + "type": "string", + "description": "The identity file's path." + }, + "type": { + "type": "string", + "enum": [ + "identityFile" + ] + } + } + } + ], + "description": "How the apply workflow's endpoint authenticates." + }, "CapabilityFactDto": { "type": "object", "description": "One capability fact as the read model displays it: the recorded status\nwith the staleness rule applied at the read time.", @@ -2821,6 +2984,45 @@ } } }, + "FieldDifferenceDto": { + "type": "object", + "description": "One field difference, as the planner produced it.", + "required": [ + "identity", + "state" + ], + "properties": { + "desired": { + "type": [ + "string", + "null" + ], + "description": "The desired value, when the field is desired." + }, + "identity": { + "type": "string", + "description": "The field's stable identity." + }, + "observed": { + "type": [ + "string", + "null" + ], + "description": "The observed value, when one was observed." + }, + "reason": { + "type": [ + "string", + "null" + ], + "description": "Why the state is `unknown` or `unsupported`, when it is." + }, + "state": { + "type": "string", + "description": "The drift state." + } + } + }, "FieldViolation": { "type": "object", "description": "One rejected input, identified by its location in the request body or query.", @@ -5166,6 +5368,55 @@ "undeploy" ] }, + "StartApplyRequest": { + "type": "object", + "description": "The body of the start-apply-workflow request.", + "required": [ + "machineId", + "endpointId", + "auth", + "planId", + "actions" + ], + "properties": { + "actions": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ApplyActionDto" + }, + "description": "The planned actions, in order." + }, + "approvals": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ApplyApprovalDto" + }, + "description": "The approvals supplied with the plan." + }, + "auth": { + "$ref": "#/components/schemas/ApplyAuthDto", + "description": "How the endpoint authenticates." + }, + "endpointId": { + "type": "string", + "description": "The SSH endpoint id to act through." + }, + "machineId": { + "type": "string", + "description": "The machine to apply on (must match the path's machine)." + }, + "planId": { + "type": "string", + "description": "The plan's identity, which every approval is bound to." + }, + "timeoutSeconds": { + "type": "integer", + "format": "int64", + "description": "The deadline, in seconds, for the whole workflow.", + "minimum": 0 + } + } + }, "StartDiscoveryRequest": { "type": "object", "description": "The body of the start-discovery request: which machine and endpoint to\nscan, and how the endpoint authenticates.", diff --git a/packages/api-client/src/generated/fleet.ts b/packages/api-client/src/generated/fleet.ts index f36d5f0..0d18a73 100644 --- a/packages/api-client/src/generated/fleet.ts +++ b/packages/api-client/src/generated/fleet.ts @@ -171,6 +171,72 @@ export interface ApiError { retry: Retry; } +/** + * One field difference, as the planner produced it. + */ +export interface FieldDifferenceDto { + /** + * The desired value, when the field is desired. + * @nullable + */ + desired?: string | null; + /** The field's stable identity. */ + identity: string; + /** + * The observed value, when one was observed. + * @nullable + */ + observed?: string | null; + /** + * Why the state is `unknown` or `unsupported`, when it is. + * @nullable + */ + reason?: string | null; + /** The drift state. */ + state: string; +} + +/** + * One planned action the caller submits. + */ +export interface ApplyActionDto { + /** The difference the action resolves. */ + difference: FieldDifferenceDto; + /** The operation kind. */ + kind: string; + /** + * The execution order. + * @minimum 0 + */ + order: number; +} + +/** + * One approval the caller supplies. + */ +export interface ApplyApprovalDto { + /** + * The action's order the approval covers. + * @minimum 0 + */ + actionOrder: number; + /** The action's operation kind the approval covers. */ + kind: string; + /** The plan's identity the approval is bound to. */ + planId: string; +} + +/** + * How the apply workflow's endpoint authenticates. + */ +export type ApplyAuthDto = { + type: 'agent'; +} | { + /** The identity file's path. */ + path: string; + type: 'identityFile'; +}; + /** * How a checkout action's endpoint authenticates. */ @@ -1692,6 +1758,29 @@ export const SkillsDirectionDto = { undeploy: 'undeploy', } as const; +/** + * The body of the start-apply-workflow request. + */ +export interface StartApplyRequest { + /** The planned actions, in order. */ + actions: ApplyActionDto[]; + /** The approvals supplied with the plan. */ + approvals?: ApplyApprovalDto[]; + /** How the endpoint authenticates. */ + auth: ApplyAuthDto; + /** The SSH endpoint id to act through. */ + endpointId: string; + /** The machine to apply on (must match the path's machine). */ + machineId: string; + /** The plan's identity, which every approval is bound to. */ + planId: string; + /** + * The deadline, in seconds, for the whole workflow. + * @minimum 0 + */ + timeoutSeconds?: number; +} + /** * The body of the start-discovery request: which machine and endpoint to * scan, and how the endpoint authenticates. @@ -2596,6 +2685,76 @@ export const getMachine = async (machineId: string, options?: RequestInit): Prom +export type startApplyWorkflowResponse202 = { + data: ResourceOperationDto + status: 202 +} + +export type startApplyWorkflowResponse400 = { + data: ApiError + status: 400 +} + +export type startApplyWorkflowResponse403 = { + data: ApiError + status: 403 +} + +export type startApplyWorkflowResponse404 = { + data: ApiError + status: 404 +} + +export type startApplyWorkflowResponseSuccess = (startApplyWorkflowResponse202) & { + headers: Headers; +}; +export type startApplyWorkflowResponseError = (startApplyWorkflowResponse400 | startApplyWorkflowResponse403 | startApplyWorkflowResponse404) & { + headers: Headers; +}; + +export type startApplyWorkflowResponse = (startApplyWorkflowResponseSuccess | startApplyWorkflowResponseError) + +export const getStartApplyWorkflowUrl = (machineId: string,) => { + + + + + return `/api/v1/machines/${machineId}/apply` +} + +/** + * # Errors + * + * Returns the public error envelope on refusal or backend failure. + * @summary Starts the apply workflow. + */ +export const startApplyWorkflow = async (machineId: string, + startApplyRequest: StartApplyRequest, 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(getStartApplyWorkflowUrl(machineId), + { + ...options, + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getHeaders(options?.headers) }, + body: JSON.stringify(startApplyRequest) + } +) + + + const body = [204, 205, 304].includes(res.status) ? null : await res.text(); + + const data: startApplyWorkflowResponse['data'] = body ? JSON.parse(body) : {} + return { data, status: res.status, headers: res.headers } as startApplyWorkflowResponse +} + + + export type startFrogenvOperationResponse202 = { data: ResourceOperationDto status: 202