diff --git a/Cargo.lock b/Cargo.lock index 147c6b8..e18fdba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -620,10 +620,12 @@ dependencies = [ "fleet-core", "fleet-protocol", "fleet-provider-frogenv", + "fleet-provider-git", "fleet-provider-mise", "fleet-provider-skills-manager", "fleet-provider-ssh", "fleet-provider-tailscale", + "fleet-schema", "fleet-secrets", "fleet-storage-sqlite", "fleetd", @@ -685,12 +687,24 @@ dependencies = [ "serde_json", ] +[[package]] +name = "fleet-provider-git" +version = "0.1.0" +dependencies = [ + "fleet-core", + "sha2 0.10.9", +] + [[package]] name = "fleet-provider-github" version = "0.1.0" dependencies = [ + "async-trait", "fleet-application", "fleet-core", + "serde", + "serde_json", + "tokio", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a840ec0..124e30a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ members = [ "crates/providers/fleet-provider-tailscale", "crates/providers/fleet-provider-docker", "crates/providers/fleet-provider-proxmox", + "crates/providers/fleet-provider-git", "crates/providers/fleet-provider-github", "crates/providers/fleet-provider-skills-manager", "crates/providers/fleet-provider-frogenv", diff --git a/crates/fleet-application/src/authz.rs b/crates/fleet-application/src/authz.rs index 6cda225..3d3debb 100644 --- a/crates/fleet-application/src/authz.rs +++ b/crates/fleet-application/src/authz.rs @@ -123,6 +123,12 @@ pub enum Permission { /// Execute an authorized apply plan on a machine. A mutation: it /// composes every mutation the plan may run. ApplyExecute, + /// Fetch and inspect candidates from the desired-state Git source. A + /// read, but a topology-revealing one. + SourceFetch, + /// Activate a validated candidate as the desired revision. A + /// mutation: it changes what Fleet converges machines toward. + SourceActivate, } impl Permission { @@ -164,6 +170,8 @@ impl Permission { Permission::MiseOperate, Permission::ProjectsReady, Permission::ApplyExecute, + Permission::SourceFetch, + Permission::SourceActivate, ]; /// The stable action id, as recorded in decisions and audit events. @@ -204,6 +212,8 @@ impl Permission { Permission::MiseOperate => "mise.operate", Permission::ProjectsReady => "projects.ready", Permission::ApplyExecute => "apply.execute", + Permission::SourceFetch => "source.fetch", + Permission::SourceActivate => "source.activate", } } @@ -246,7 +256,9 @@ impl Permission { | Permission::ToolsRead | Permission::MiseOperate | Permission::ProjectsReady - | Permission::ApplyExecute => true, + | Permission::ApplyExecute + | Permission::SourceFetch + | Permission::SourceActivate => true, } } @@ -267,7 +279,9 @@ impl Permission { | Permission::TailnetRead | Permission::TailnetConfig | Permission::ProjectsRead - | Permission::ProjectsCreate => false, + | Permission::ProjectsCreate + | Permission::SourceFetch + | Permission::SourceActivate => false, Permission::MachineReadSensitive | Permission::OperationCancel | Permission::SecretRead diff --git a/crates/fleet-application/src/lib.rs b/crates/fleet-application/src/lib.rs index f53df62..a570ee1 100644 --- a/crates/fleet-application/src/lib.rs +++ b/crates/fleet-application/src/lib.rs @@ -16,5 +16,6 @@ pub mod operation; pub mod planner; pub mod project; pub mod ready; +pub mod source; pub mod tailnet; pub mod worker; diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index 37e7a5a..5fea37d 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -41,8 +41,9 @@ use crate::authz::{AccessRequest, Authorizer, Decision, Permission, ReasonId, au /// (FM-304); the ready workflow carries the machine-scoped shape plus /// `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] = [ +/// (FM-402); the source kinds carry the remote/commit payloads and are +/// catalog-level (FM-403). +pub const CREATABLE_KINDS: [&str; 31] = [ "noop", "ssh.exec", "agentless.inventory", @@ -72,6 +73,8 @@ pub const CREATABLE_KINDS: [&str; 29] = [ "mise.exec", "ready.workflow", "apply.workflow", + "source.fetch", + "source.activate", ]; /// The machine-scoped permission a kind's creation requires, when any. @@ -80,6 +83,16 @@ pub const CREATABLE_KINDS: [&str; 29] = [ /// governs both the dedicated endpoint and the generic one. #[must_use] fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option { + // The source kinds are catalog-level: their permission is enforced + // here with `resource: None`, never a machine id. + match kind { + "source.fetch" => Some(Permission::SourceFetch), + "source.activate" => Some(Permission::SourceActivate), + _ => machine_scoped_kind_permission_inner(kind, payload), + } +} + +fn machine_scoped_kind_permission_inner(kind: &str, payload: Option<&str>) -> Option { match kind { "projects.discover" => Some(Permission::ProjectsDiscover), "projects.clone" | "projects.pull" | "projects.status" => { @@ -109,6 +122,7 @@ fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option

Some(Permission::MiseOperate), "ready.workflow" => Some(Permission::ProjectsReady), "apply.workflow" => Some(Permission::ApplyExecute), + _ => None, } } diff --git a/crates/fleet-application/src/source.rs b/crates/fleet-application/src/source.rs new file mode 100644 index 0000000..c0cc2db --- /dev/null +++ b/crates/fleet-application/src/source.rs @@ -0,0 +1,507 @@ +//! The desired-source use cases (FM-403): candidate fetching, activation +//! gating, and explicit rollback. +//! +//! A Git repository becomes the canonical source of desired resources +//! only when validation gates activation: a candidate carrying +//! diagnostics can never become active, the last valid revision stays +//! active on failure, activation is serialized and audited, and rollback +//! is an explicit authorized operation naming a prior valid revision — +//! never automatic. +//! +//! The active revision's identity is durable state: (commit SHA + content +//! digest), so a controller restart resumes truthfully. +#![warn(missing_docs)] + +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; + +/// One active revision: the candidate digest that was activated. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ActiveRevision { + /// The commit SHA of the active revision. + pub commit_sha: String, + /// The content digest of the active revision. + pub content_digest: String, +} + +/// The storage contract for the desired source's durable state. +#[async_trait::async_trait] +pub trait SourcePort: std::fmt::Debug + Send + Sync { + /// Reads the active revision, when one has been activated. + /// + /// # Errors + /// + /// Fails when the backend errors. + async fn active_revision(&self) -> Result, String>; + /// The digests of prior valid revisions, for manual rollback. + /// + /// # Errors + /// + /// Fails when the backend errors. + async fn prior_revisions(&self) -> Result, String>; + /// Records a candidate's digest as a prior valid revision (only valid + /// candidates are recorded — a candidate carrying diagnostics is + /// reported and forgotten). + /// + /// # Errors + /// + /// Fails when the backend errors. + async fn record_valid_revision(&self, revision: &ActiveRevision) -> Result<(), String>; + /// Atomically activates a revision: the backend serializes the + /// check-and-set so concurrent activations cannot race — the method + /// returns the revision that is active AFTER the call, which is the + /// caller's requested revision only if the activation won. + /// + /// # Errors + /// + /// Fails when the backend errors. + async fn activate_serialized( + &self, + revision: &ActiveRevision, + ) -> Result; +} + +/// The outcome of one fetch: the candidate's digest and validation +/// diagnostics, plus whether the fetch could complete at all. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum FetchOutcome { + /// The candidate was fetched and validated; `diagnostics` empty means + /// it may be activated. + Candidate { + /// The candidate's digest. + digest: fleet_core::CandidateDigest, + /// The validation diagnostics, empty when valid. + diagnostics: Vec, + }, + /// The fetch failed at the transport level (clone/checkout); the + /// active revision is untouched. + TransportFailed { + /// The bounded, redacted failure detail. + detail: String, + }, +} + +/// The authorized desired-source use cases. +#[derive(Clone, Debug)] +pub struct DesiredSource { + port: std::sync::Arc, + audit: std::sync::Arc, +} + +impl DesiredSource { + /// Composes the service from its ports. + #[must_use] + pub fn new( + port: std::sync::Arc, + audit: std::sync::Arc, + ) -> Self { + Self { port, audit } + } + + /// Handles one fetch outcome: a valid candidate is recorded as a + /// prior valid revision (available for manual rollback); a candidate + /// carrying diagnostics is reported and forgotten; a transport + /// failure leaves everything untouched. The active revision only + /// changes through [`activate`](Self::activate). + /// + /// # Errors + /// + /// Fails when the audit or backend errors. + pub async fn handle_fetch( + &self, + authorizer: &dyn crate::authz::Authorizer, + principal_id: &str, + outcome: FetchOutcome, + ) -> Result { + use crate::authz::{AccessRequest, Permission, authorize}; + authorize( + authorizer, + AccessRequest { + principal_id, + action: Permission::SourceFetch, + resource: None, + }, + ) + .map_err(crate::project::ProjectUseCaseError::Denied)?; + match outcome { + FetchOutcome::Candidate { + ref digest, + ref diagnostics, + } => { + if diagnostics.is_empty() { + self.port + .record_valid_revision(&ActiveRevision { + commit_sha: digest.commit_sha.clone(), + content_digest: digest.content_digest.clone(), + }) + .await + .map_err(|detail| crate::project::ProjectUseCaseError::Backend { + context: "source_record", + detail, + })?; + } + Ok(outcome) + } + transport_failure @ FetchOutcome::TransportFailed { .. } => Ok(transport_failure), + } + } + + /// Activates one revision by digest: the candidate must be valid and + /// known (fetched before), activation is serialized and audited, and + /// the active revision is replaced only on success. + /// + /// # Errors + /// + /// Fails on denial, an unknown or invalid revision, or a backend + /// failure. + pub async fn activate( + &self, + authorizer: &dyn crate::authz::Authorizer, + principal_id: &str, + digest: &fleet_core::CandidateDigest, + candidate_valid: bool, + candidate_known: bool, + operation_id: Option<&str>, + ) -> Result { + use crate::authz::{AccessRequest, Permission, authorize}; + authorize( + authorizer, + AccessRequest { + principal_id, + action: Permission::SourceActivate, + resource: None, + }, + ) + .map_err(crate::project::ProjectUseCaseError::Denied)?; + // An invalid candidate can never become active, and an unknown + // digest was never fetched: both are refusals, not errors. The + // caller's claims are verified against the recorded candidate + // state: only a candidate the fetch path recorded as valid AND + // known may activate. + let recorded = self.port.prior_revisions().await.map_err(|detail| { + crate::project::ProjectUseCaseError::Backend { + context: "source_verify", + detail, + } + })?; + let known_and_valid = recorded + .iter() + .any(|revision| revision.commit_sha == digest.commit_sha); + if !candidate_valid { + return Err(crate::project::ProjectUseCaseError::Invalid { + detail: "the candidate carries validation diagnostics and cannot become active" + .to_owned(), + }); + } + if !candidate_known || !known_and_valid { + return Err(crate::project::ProjectUseCaseError::Invalid { + detail: "the candidate was never fetched as valid; fetch it before activating" + .to_owned(), + }); + } + // The audit intent lands BEFORE the mutation: a failure to audit + // prevents the activation, so durable state can never exist + // without its intent. + let mut metadata = crate::audit::AuditMetadata::default(); + metadata + .insert("event", "source_activation") + .map_err(|error| crate::project::ProjectUseCaseError::Backend { + context: "audit", + detail: error.to_string(), + })?; + metadata + .insert("commitSha", &digest.commit_sha) + .map_err(|error| crate::project::ProjectUseCaseError::Backend { + context: "audit", + detail: error.to_string(), + })?; + self.audit + .record_intent(&crate::audit::AuditIntent { + actor: principal_id.to_owned(), + action: Permission::SourceActivate.id().to_owned(), + resource: None, + decision: crate::authz::Decision::allow(), + correlation_id: None, + operation_id: operation_id.map(str::to_owned), + metadata, + }) + .await + .map_err(|detail| crate::project::ProjectUseCaseError::Backend { + context: "audit", + detail, + })?; + let revision = ActiveRevision { + commit_sha: digest.commit_sha.clone(), + content_digest: digest.content_digest.clone(), + }; + // The backend serializes the check-and-set: a concurrent + // activation cannot race. + self.port + .activate_serialized(&revision) + .await + .map_err(|detail| crate::project::ProjectUseCaseError::Backend { + context: "source_activate", + detail, + }) + } + + /// The prior valid revisions available for manual rollback. + /// + /// # Errors + /// + /// Fails on denial or a backend failure. + pub async fn prior_revisions( + &self, + authorizer: &dyn crate::authz::Authorizer, + principal_id: &str, + ) -> Result, crate::project::ProjectUseCaseError> { + use crate::authz::{AccessRequest, Permission, authorize}; + authorize( + authorizer, + AccessRequest { + principal_id, + action: Permission::SourceFetch, + resource: None, + }, + ) + .map_err(crate::project::ProjectUseCaseError::Denied)?; + self.port.prior_revisions().await.map_err(|detail| { + crate::project::ProjectUseCaseError::Backend { + context: "source_prior", + detail, + } + }) + } +} + +/// Reports the conflict between two revisions as data: the two digests +/// and the divergent file paths. Fleet never auto-resolves conflicts. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ConflictReport { + /// The active revision's digest. + pub active: ActiveRevision, + /// The fetched revision's digest. + pub fetched: ActiveRevision, + /// The files that differ between the two revisions. + pub divergent_files: BTreeSet, +} + +#[cfg(test)] +mod tests { + use super::{ActiveRevision, DesiredSource, FetchOutcome, SourcePort}; + use crate::authz::{AccessRequest, Authorizer, Decision}; + use std::sync::{Arc, Mutex}; + + #[derive(Debug)] + struct AllowAll; + impl Authorizer for AllowAll { + fn decide(&self, _request: AccessRequest<'_>) -> Decision { + Decision::allow() + } + } + + #[derive(Debug, Default)] + struct FakePort { + active: Mutex>, + valid: Mutex>, + } + + #[async_trait::async_trait] + impl SourcePort for FakePort { + async fn active_revision(&self) -> Result, String> { + Ok(self.active.lock().unwrap().clone()) + } + async fn prior_revisions(&self) -> Result, String> { + Ok(self.valid.lock().unwrap().clone()) + } + async fn record_valid_revision(&self, revision: &ActiveRevision) -> Result<(), String> { + self.valid.lock().unwrap().push(revision.clone()); + Ok(()) + } + async fn activate_serialized( + &self, + revision: &ActiveRevision, + ) -> Result { + *self.active.lock().unwrap() = Some(revision.clone()); + Ok(revision.clone()) + } + } + + #[derive(Debug, Default)] + struct FakeAudit { + intents: Mutex>, + } + #[async_trait::async_trait] + impl crate::operation::AuditPort for FakeAudit { + async fn record_intent(&self, intent: &crate::audit::AuditIntent) -> Result<(), String> { + self.intents.lock().unwrap().push(intent.clone()); + Ok(()) + } + async fn record_outcome( + &self, + _operation_id: &str, + _outcome: crate::audit::AuditOutcome, + ) -> Result<(), String> { + Ok(()) + } + } + + fn digest(sha: &str, content: &str) -> fleet_core::CandidateDigest { + fleet_core::CandidateDigest { + commit_sha: sha.to_owned(), + content_digest: content.to_owned(), + } + } + + fn service() -> (DesiredSource, Arc, Arc) { + let port = Arc::new(FakePort::default()); + let audit = Arc::new(FakeAudit::default()); + (DesiredSource::new(port.clone(), audit.clone()), audit, port) + } + + #[tokio::test] + async fn a_valid_candidate_is_recorded_for_rollback() { + let (service, _, port) = service(); + let outcome = FetchOutcome::Candidate { + digest: digest("abc", "digest-1"), + diagnostics: vec![], + }; + let handled = service + .handle_fetch(&AllowAll, "anonymous-lan-admin", outcome) + .await + .unwrap(); + let FetchOutcome::Candidate { diagnostics, .. } = handled else { + panic!("the candidate outcome is preserved"); + }; + assert!(diagnostics.is_empty()); + assert_eq!(port.prior_revisions().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn a_candidate_with_diagnostics_is_reported_and_forgotten() { + let (service, _, port) = service(); + let outcome = FetchOutcome::Candidate { + digest: digest("abc", "digest-bad"), + diagnostics: vec!["the document does not match".to_owned()], + }; + service + .handle_fetch(&AllowAll, "anonymous-lan-admin", outcome) + .await + .unwrap(); + assert!( + port.prior_revisions().await.unwrap().is_empty(), + "an invalid candidate is never a rollback point" + ); + } + + #[tokio::test] + async fn activation_refuses_an_invalid_candidate() { + let (service, _, _) = service(); + let error = service + .activate( + &AllowAll, + "anonymous-lan-admin", + &digest("abc", "bad"), + false, + true, + None, + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("cannot become active"), + "{error}" + ); + assert!(service.port.active_revision().await.unwrap().is_none()); + } + + #[tokio::test] + async fn activation_refuses_an_unknown_candidate() { + let (service, _, _) = service(); + let error = service + .activate( + &AllowAll, + "anonymous-lan-admin", + &digest("abc", "d"), + true, + false, + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("never fetched"), "{error}"); + } + + #[tokio::test] + async fn activation_is_audited_and_durable() { + let (service, audit, port) = service(); + // The candidate must have been fetched as valid before activation: + // the fetch records it. + service + .handle_fetch( + &AllowAll, + "anonymous-lan-admin", + FetchOutcome::Candidate { + digest: digest("abc", "digest-1"), + diagnostics: vec![], + }, + ) + .await + .unwrap(); + let revision = service + .activate( + &AllowAll, + "anonymous-lan-admin", + &digest("abc", "digest-1"), + true, + true, + Some("op-1"), + ) + .await + .unwrap(); + assert_eq!(revision.commit_sha, "abc"); + let active = port.active_revision().await.unwrap().unwrap(); + assert_eq!(active.content_digest, "digest-1"); + // The activation is audited: the intent names the action, carries + // the commit SHA, and is correlated with the operation. + let intents = audit.intents.lock().unwrap(); + assert_eq!(intents.len(), 1); + assert_eq!(intents[0].action, "source.activate"); + assert!( + intents[0] + .metadata + .entries() + .any(|(key, value)| key == "commitSha" && value == "abc"), + "the intent carries the commit SHA" + ); + assert!(intents[0].operation_id.is_some()); + } + + #[tokio::test] + async fn a_denied_fetch_or_activation_never_touches_state() { + #[derive(Debug)] + struct DenyAll; + impl Authorizer for DenyAll { + fn decide(&self, _request: AccessRequest<'_>) -> Decision { + Decision::deny(crate::authz::ReasonId::UnknownPrincipal) + } + } + let (service, _, port) = service(); + let outcome = FetchOutcome::Candidate { + digest: digest("abc", "d"), + diagnostics: vec![], + }; + assert!(service.handle_fetch(&DenyAll, "x", outcome).await.is_err()); + assert!( + service + .activate(&DenyAll, "x", &digest("abc", "d"), true, true, None) + .await + .is_err() + ); + assert!(port.active_revision().await.unwrap().is_none()); + assert!(port.prior_revisions().await.unwrap().is_empty()); + } +} diff --git a/crates/fleet-auth/tests/authz_adapter.rs b/crates/fleet-auth/tests/authz_adapter.rs index f00e499..f50c9da 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(), 34); + assert_eq!(Permission::ALL.len(), 36); } #[test] diff --git a/crates/fleet-controller/Cargo.toml b/crates/fleet-controller/Cargo.toml index 19d53a4..02df0e7 100644 --- a/crates/fleet-controller/Cargo.toml +++ b/crates/fleet-controller/Cargo.toml @@ -18,6 +18,8 @@ fleet-core = { version = "0.1.0", path = "../fleet-core" } fleet-protocol = { path = "../fleet-protocol" } fleet-provider-ssh = { version = "0.1.0", path = "../providers/fleet-provider-ssh" } fleet-provider-frogenv = { version = "0.1.0", path = "../providers/fleet-provider-frogenv" } +fleet-provider-git = { version = "0.1.0", path = "../providers/fleet-provider-git" } +fleet-schema = { path = "../../schemas" } fleet-provider-mise = { version = "0.1.0", path = "../providers/fleet-provider-mise" } fleet-provider-skills-manager = { version = "0.1.0", path = "../providers/fleet-provider-skills-manager" } fleet-provider-tailscale = { path = "../providers/fleet-provider-tailscale" } diff --git a/crates/fleet-controller/src/lib.rs b/crates/fleet-controller/src/lib.rs index 4dfc223..a45d0f3 100644 --- a/crates/fleet-controller/src/lib.rs +++ b/crates/fleet-controller/src/lib.rs @@ -22,6 +22,7 @@ pub mod node_crypto; pub mod onboard; pub mod ready; pub mod skills; +pub mod source; pub mod tailnet_store; pub mod worker; diff --git a/crates/fleet-controller/src/main.rs b/crates/fleet-controller/src/main.rs index a4d8073..f8d4e44 100644 --- a/crates/fleet-controller/src/main.rs +++ b/crates/fleet-controller/src/main.rs @@ -267,6 +267,24 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { )), )) }; + // The source executor handles the FM-403 kinds over the git + // work root and the desired-source use cases. + let with_source: std::sync::Arc = { + std::sync::Arc::new(fleet_controller::source::SourceDispatch::new( + with_apply.clone(), + std::sync::Arc::new(fleet_controller::source::SourceExecutor::new( + config.data_dir.join("git-source"), + std::sync::Arc::new(fleet_application::source::DesiredSource::new( + std::sync::Arc::new(fleet_storage_sqlite::SourceRepository::new( + store.pool().clone(), + )), + std::sync::Arc::new(fleet_storage_sqlite::AuditSink::new( + store.pool().clone(), + )), + )), + )), + )) + }; match &services { Some(services) => { let node_machines: std::sync::Arc = @@ -277,11 +295,11 @@ fn run_serve(config: fleet_config::ControllerConfig) -> ExitCode { std::sync::Arc::new(fleet_controller::gateway::NodeCommandExecutor::new( services.gateway.clone(), node_machines, - with_apply.clone(), + with_source.clone(), )); executor } - None => with_apply.clone(), + None => with_source.clone(), } }; let worker_host = WorkerHost::new(worker_operations, executor, 4); diff --git a/crates/fleet-controller/src/source.rs b/crates/fleet-controller/src/source.rs new file mode 100644 index 0000000..379e4c8 --- /dev/null +++ b/crates/fleet-controller/src/source.rs @@ -0,0 +1,272 @@ +//! The desired-source executor (FM-403): durable fetch/activate +//! operations over the Git source provider. +//! +//! `source.fetch` clones the pinned commit, validates, and records the +//! outcome through the authorized use case (valid candidates become +//! rollback points; invalid ones are reported and forgotten). +//! `source.activate` activates a fetched candidate through the authorized +//! use case — the activation is serialized and audited there. + +use std::sync::Arc; + +use fleet_application::operation::{Operation, Operations}; +use fleet_application::worker::OperationExecutor; +use fleet_provider_git::GitSource; + +/// The `source.fetch` payload. +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct FetchPayload { + /// The desired-state repository's remote. + remote: String, + /// The commit SHA to fetch. + commit_sha: String, +} + +/// The `source.activate` payload. +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct ActivatePayload { + /// The commit SHA to activate. + commit_sha: String, + /// The content digest the candidate was fetched with. + content_digest: String, +} + +/// The kind-dispatching source executor. +#[derive(Debug)] +pub struct SourceExecutor { + source: GitSource, + desired_source: Arc, +} + +impl SourceExecutor { + /// Composes the executor from its parts. + /// + /// # Panics + /// + /// Panics only if the git work root cannot be prepared, which the + /// controller's data-directory preparation already ensures. + #[must_use] + pub fn new( + work_root: std::path::PathBuf, + desired_source: Arc, + ) -> Self { + Self { + source: GitSource::new(work_root), + desired_source, + } + } +} + +#[async_trait::async_trait] +impl OperationExecutor for SourceExecutor { + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + match operation.kind.as_str() { + "source.fetch" => self.fetch(operations, operation).await, + "source.activate" => self.activate(operations, operation).await, + _ => Err("not a source kind".to_owned()), + } + } +} + +impl SourceExecutor { + async fn fetch(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + let payload: FetchPayload = 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 source record: {error}"))?; + operations + .record_progress( + &operation.id, + Some(0), + Some(1), + Some(&format!("fetching {}", payload.commit_sha)), + ) + .await + .map_err(|error| error.to_string())?; + // The provider runs on the controller's own machine; the + // validation closure rides the schemas crate's validate_paths. + let outcome = { + let source = self.source.clone(); + let remote = payload.remote.clone(); + let commit_sha = payload.commit_sha.clone(); + tokio::task::spawn_blocking(move || { + source.fetch_candidate(&remote, &commit_sha, |sources| { + fleet_schema::validate_paths(sources) + .unwrap_or_default() + .iter() + .map(ToString::to_string) + .collect() + }) + }) + .await + .map_err(|join_error| format!("the fetch thread failed: {join_error}"))? + }; + let outcome = match outcome { + Ok(candidate) => fleet_application::source::FetchOutcome::Candidate { + digest: candidate.digest, + diagnostics: candidate.diagnostics, + }, + Err(detail) => fleet_application::source::FetchOutcome::TransportFailed { detail }, + }; + let handled = self + .desired_source + .handle_fetch( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + outcome, + ) + .await + .map_err(|error| error.to_string())?; + let result_json = match handled { + fleet_application::source::FetchOutcome::Candidate { + digest, + diagnostics, + } => serde_json::json!({ + "commitSha": digest.commit_sha, + "contentDigest": digest.content_digest, + "diagnostics": diagnostics, + "valid": diagnostics.is_empty(), + }) + .to_string(), + fleet_application::source::FetchOutcome::TransportFailed { detail } => { + return complete_failed(operations, &operation.id, &detail).await; + } + }; + operations + .complete(&operation.id, "succeeded", Some(&result_json), None) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } + + async fn activate(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + let payload: ActivatePayload = 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 source record: {error}"))?; + // The candidate must have been fetched: the worktree named by the + // SHA proves it. Validation re-runs against the materialized + // worktree so the activation gate holds even across restarts. + let worktree = self + .source + .work_root() + .join(format!("candidate-{}", payload.commit_sha)); + let candidate_known = worktree.exists(); + let diagnostics = if candidate_known { + let mut sources = Vec::new(); + collect_yaml(&worktree, &worktree, &mut sources); + fleet_schema::validate_paths(&sources) + .unwrap_or_default() + .iter() + .map(ToString::to_string) + .collect() + } else { + vec!["the candidate was never fetched".to_owned()] + }; + let revision = self + .desired_source + .activate( + &fleet_auth::LanAllowAllAuthorizer, + fleet_auth::LAN_PRINCIPAL_ID, + &fleet_core::CandidateDigest { + commit_sha: payload.commit_sha.clone(), + content_digest: payload.content_digest.clone(), + }, + diagnostics.is_empty(), + candidate_known, + Some(&operation.id), + ) + .await + .map_err(|error| { + // A refusal (invalid/unknown candidate) is the operation's + // public failure, not a backend error. + error.to_string() + })?; + let result_json = serde_json::json!({ + "activated": true, + "commitSha": revision.commit_sha, + "contentDigest": revision.content_digest, + }) + .to_string(); + operations + .complete(&operation.id, "succeeded", Some(&result_json), None) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) + } +} + +/// Completes the workflow as a failure with a stable reason. +async fn complete_failed( + operations: &Operations, + operation_id: &str, + detail: &str, +) -> Result<(), String> { + let error_json = serde_json::json!({ "reason": "fetch_failed", "detail": detail }).to_string(); + operations + .complete(operation_id, "failed", None, Some(&error_json)) + .await + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +/// Collects YAML files under a root, as relative paths. +fn collect_yaml( + _root: &std::path::Path, + base: &std::path::Path, + sources: &mut Vec, +) { + let Ok(entries) = std::fs::read_dir(base) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + if path.file_name().is_some_and(|name| name == ".git") { + continue; + } + collect_yaml(_root, &path, sources); + } else if path + .extension() + .is_some_and(|extension| extension == "yaml") + { + sources.push(path); + } + } +} + +/// The kind-dispatching wrapper the controller composes: the source kinds +/// route to the [`SourceExecutor`], everything else falls through to the +/// rest of the chain unchanged. +#[derive(Debug)] +pub struct SourceDispatch { + fallback: Arc, + source: Arc, +} + +impl SourceDispatch { + /// Composes the dispatch from the fallback chain and the source + /// executor. + #[must_use] + pub fn new(fallback: Arc, source: Arc) -> Self { + Self { fallback, source } + } +} + +#[async_trait::async_trait] +impl OperationExecutor for SourceDispatch { + async fn execute(&self, operations: &Operations, operation: &Operation) -> Result<(), String> { + match operation.kind.as_str() { + "source.fetch" | "source.activate" => self.source.execute(operations, operation).await, + _ => self.fallback.execute(operations, operation).await, + } + } +} diff --git a/crates/fleet-core/src/lib.rs b/crates/fleet-core/src/lib.rs index 0b72eeb..3af3b7c 100644 --- a/crates/fleet-core/src/lib.rs +++ b/crates/fleet-core/src/lib.rs @@ -14,6 +14,7 @@ mod operation; mod project; mod redact; mod sensitive; +mod source; mod time; mod value; @@ -29,5 +30,6 @@ pub use redact::{ flatten_control_characters, redact_schemeless_credentials, redact_url_credentials, }; pub use sensitive::{SecretReference, SensitiveString}; +pub use source::CandidateDigest; pub use time::{Clock, Deadline, FixedClock, SystemClock, Timestamp}; pub use value::{ParseSlugError, Revision, Slug}; diff --git a/crates/fleet-core/src/source.rs b/crates/fleet-core/src/source.rs new file mode 100644 index 0000000..a5ab5ae --- /dev/null +++ b/crates/fleet-core/src/source.rs @@ -0,0 +1,23 @@ +//! The desired-source primitives (FM-403): the candidate digest shared +//! by the Git provider and the desired-source use cases. + +use serde::{Deserialize, Serialize}; + +/// The digest of one candidate: the commit SHA plus the content digest of +/// its file set, so two candidates are equal only when both match. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CandidateDigest { + /// The commit SHA the candidate was fetched at. + pub commit_sha: String, + /// The SHA-256 of the candidate's tracked file set (paths + contents). + pub content_digest: String, +} + +impl CandidateDigest { + /// Whether two digests describe the same candidate. + #[must_use] + pub fn matches(&self, other: &Self) -> bool { + self.commit_sha == other.commit_sha && self.content_digest == other.content_digest + } +} diff --git a/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql new file mode 100644 index 0000000..1f4cc5e --- /dev/null +++ b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql @@ -0,0 +1,31 @@ +-- FM-403: the desired-state source's durable state — the active revision +-- (single row, upsert) and the prior valid revisions (append-only +-- history for manual rollback). Only VALID candidates are recorded; a +-- candidate carrying diagnostics is reported and forgotten. +CREATE TABLE source_active_revision ( + singleton TEXT PRIMARY KEY CHECK (singleton = 'active'), + commit_sha TEXT NOT NULL, + content_digest TEXT NOT NULL, + activated_at INTEGER NOT NULL +) STRICT; + +CREATE TABLE source_revision_history ( + id TEXT PRIMARY KEY, + commit_sha TEXT NOT NULL, + content_digest TEXT NOT NULL, + activated_at INTEGER NOT NULL +) STRICT; + +CREATE INDEX source_revision_history_at ON source_revision_history (activated_at); + +CREATE TRIGGER source_revision_history_no_update + BEFORE UPDATE ON source_revision_history +BEGIN + SELECT RAISE(ABORT, 'source_revision_history is append-only'); +END; + +CREATE TRIGGER source_revision_history_no_delete + BEFORE DELETE ON source_revision_history +BEGIN + SELECT RAISE(ABORT, 'source_revision_history is append-only'); +END; diff --git a/crates/fleet-storage-sqlite/src/lib.rs b/crates/fleet-storage-sqlite/src/lib.rs index 43b4596..e046460 100644 --- a/crates/fleet-storage-sqlite/src/lib.rs +++ b/crates/fleet-storage-sqlite/src/lib.rs @@ -18,6 +18,7 @@ pub mod nodes; pub mod onboarding; pub mod operations; pub mod projects; +pub mod source; pub use audit::{AuditLedger, AuditSink}; pub use machines::MachineRepository; @@ -25,6 +26,7 @@ pub use nodes::NodeRepository; pub use onboarding::OnboardingRepository; pub use operations::OperationRepository; pub use projects::ProjectRepository; +pub use source::SourceRepository; use std::path::{Path, PathBuf}; use std::time::Duration; diff --git a/crates/fleet-storage-sqlite/src/source.rs b/crates/fleet-storage-sqlite/src/source.rs new file mode 100644 index 0000000..2fa47e7 --- /dev/null +++ b/crates/fleet-storage-sqlite/src/source.rs @@ -0,0 +1,106 @@ +//! The desired-source adapter (FM-403): the active revision and prior +//! valid revisions, durable in SQLite. +//! +//! The active revision is a single-row state table (upsert semantics); +//! prior valid revisions are an append-only history for manual rollback. + +use sqlx::SqlitePool; + +use fleet_application::source::ActiveRevision; + +/// The desired-source repository over the controller's pool. +#[derive(Debug, Clone)] +pub struct SourceRepository { + pool: SqlitePool, +} + +impl SourceRepository { + /// Composes the repository over the pool. + #[must_use] + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } +} + +#[async_trait::async_trait] +impl fleet_application::source::SourcePort for SourceRepository { + async fn active_revision(&self) -> Result, String> { + let row: Option<(String, String)> = + sqlx::query_as("SELECT commit_sha, content_digest FROM source_active_revision LIMIT 1") + .fetch_optional(&self.pool) + .await + .map_err(|error| format!("the active revision read failed: {error}"))?; + Ok(row.map(|(commit_sha, content_digest)| ActiveRevision { + commit_sha, + content_digest, + })) + } + + async fn prior_revisions(&self) -> Result, String> { + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT commit_sha, content_digest FROM source_revision_history \ + ORDER BY activated_at DESC LIMIT 100", + ) + .fetch_all(&self.pool) + .await + .map_err(|error| format!("the revision history read failed: {error}"))?; + Ok(rows + .into_iter() + .map(|(commit_sha, content_digest)| ActiveRevision { + commit_sha, + content_digest, + }) + .collect()) + } + + async fn activate_serialized( + &self, + revision: &ActiveRevision, + ) -> Result { + // The critical section is serialized by BEGIN IMMEDIATE: only one + // activation's check-and-set can run at a time, so a concurrent + // activation cannot race. + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(|error| format!("the activation lock failed: {error}"))?; + sqlx::query( + "INSERT INTO source_active_revision (singleton, commit_sha, content_digest, activated_at) \ + VALUES ('active', ?1, ?2, ?3) \ + ON CONFLICT(singleton) DO UPDATE SET \ + commit_sha = excluded.commit_sha, \ + content_digest = excluded.content_digest, \ + activated_at = excluded.activated_at", + ) + .bind(&revision.commit_sha) + .bind(&revision.content_digest) + .bind(fleet_core::SystemClock::now_unix_millis()) + .execute(&mut *tx) + .await + .map_err(|error| format!("the active revision write failed: {error}"))?; + tx.commit() + .await + .map_err(|error| format!("the activation commit failed: {error}"))?; + Ok(revision.clone()) + } + + async fn record_valid_revision(&self, revision: &ActiveRevision) -> Result<(), String> { + sqlx::query( + "INSERT OR IGNORE INTO source_revision_history \ + (id, commit_sha, content_digest, activated_at) \ + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(format!( + "rev-{}-{}", + revision.commit_sha, revision.content_digest + )) + .bind(&revision.commit_sha) + .bind(&revision.content_digest) + .bind(fleet_core::SystemClock::now_unix_millis()) + .execute(&self.pool) + .await + .map_err(|error| format!("the revision history write failed: {error}"))?; + Ok(()) + } +} diff --git a/crates/providers/fleet-provider-git/Cargo.toml b/crates/providers/fleet-provider-git/Cargo.toml new file mode 100644 index 0000000..8964052 --- /dev/null +++ b/crates/providers/fleet-provider-git/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "fleet-provider-git" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +publish.workspace = true + +[dependencies] +fleet-core = { path = "../../fleet-core" } +sha2 = "0.10" + +[lints] +workspace = true diff --git a/crates/providers/fleet-provider-git/src/lib.rs b/crates/providers/fleet-provider-git/src/lib.rs new file mode 100644 index 0000000..07878aa --- /dev/null +++ b/crates/providers/fleet-provider-git/src/lib.rs @@ -0,0 +1,238 @@ +//! The Git source provider (FM-403): an isolated clone/worktree per +//! candidate, immutable candidates by SHA + content digest, and hooks +//! that never run. +//! +//! A Git repository becomes the canonical source of desired resources +//! only when validation gates activation. The provider clones a pinned +//! commit into an isolated directory (never shared with Fleet's runtime +//! state), records the candidate as (commit SHA + content digest), and +//! hands the file set to the schemas crate for validation. Every git +//! invocation passes `-c core.hooksPath=/nonexistent-fleet-hooks` — hook +//! execution is remote code execution by another name (the FM-301 rule). +//! +//! The provider runs git locally (the controller's own machine), not +//! over SSH: the desired repository is infrastructure, not a managed +//! machine's state. +#![warn(missing_docs)] + +use fleet_core::CandidateDigest; +use std::path::{Path, PathBuf}; +use std::process::Command; + +/// One candidate: the digest plus the isolated worktree path and the +/// diagnostics validation produced. +#[derive(Clone, Debug)] +pub struct Candidate { + /// The candidate's digest. + pub digest: CandidateDigest, + /// The isolated worktree the candidate was materialized into. + pub worktree: PathBuf, + /// The validation diagnostics: empty means the candidate is valid and + /// may be activated. + pub diagnostics: Vec, +} + +impl Candidate { + /// Whether the candidate may be activated. + #[must_use] + pub fn valid(&self) -> bool { + self.diagnostics.is_empty() + } +} + +/// The git transport: one isolated clone/worktree per candidate. +#[derive(Clone, Debug)] +pub struct GitSource { + /// The root directory the provider materializes worktrees under. + work_root: PathBuf, +} + +impl GitSource { + /// Composes the provider over a work root. + /// + /// # Panics + /// + /// Panics only if the work root cannot be prepared, which the + /// controller's data-directory preparation already ensures. + #[must_use] + pub fn new(work_root: PathBuf) -> Self { + std::fs::create_dir_all(&work_root).expect("the git work root must prepare"); + Self { work_root } + } + + /// The work root. + #[must_use] + pub fn work_root(&self) -> &Path { + &self.work_root + } + + /// Runs one git command with hooks disabled; output is bounded by + /// the caller's use. + fn git(&self, arguments: &[&str]) -> Result { + let _ = self; + let mut command = Command::new("git"); + command + .arg("-c") + .arg("core.hooksPath=/nonexistent-fleet-hooks") + .args(arguments); + let output = command + .output() + .map_err(|error| format!("git could not start: {error}"))?; + if !output.status.success() { + // The stderr is bounded and redacted: a credential-bearing + // remote can echo its URL in a failure message. + let stderr = fleet_core::redact_schemeless_credentials( + &fleet_core::redact_url_credentials(&String::from_utf8_lossy(&output.stderr)), + ); + let bounded = stderr.trim().chars().take(500).collect::(); + return Err(format!( + "git {} failed: {bounded}", + arguments.first().unwrap_or(&"") + )); + } + // The stdout is bounded: a large repository cannot consume + // unbounded controller memory. A truncated listing would hash only + // part of the file set, so oversized output is refused outright. + let stdout = String::from_utf8_lossy(&output.stdout); + if output.stdout.len() > 1024 * 1024 { + return Err( + "the tracked file listing exceeds its 1 MiB bound; the candidate is refused rather than partially hashed".to_owned(), + ); + } + Ok(stdout.into_owned()) + } + + /// Fetches one candidate: clones the repository at the pinned commit + /// into an isolated worktree, computes the digest, and returns the + /// candidate with its validation diagnostics. + /// + /// # Errors + /// + /// Fails on transport errors (clone/checkout failures); a candidate + /// whose validation fails is a valid return carrying diagnostics. + pub fn fetch_candidate( + &self, + remote: &str, + commit_sha: &str, + validate: impl FnOnce(&[PathBuf]) -> Vec, + ) -> Result { + // The SHA must be a full hexadecimal commit id: anything else + // could escape the isolated work root through the path. + if commit_sha.len() != 40 + || !commit_sha + .chars() + .all(|c| c.is_ascii_hexdigit() && c.is_ascii_lowercase() || c.is_ascii_digit()) + { + return Err("the commit SHA must be a full 40-character hexadecimal id".to_owned()); + } + // The worktree directory is named by the SHA: the same commit + // materializes to the same isolated directory. An existing + // directory is discarded and recreated — a stale or partial clone + // must never be validated under the requested SHA. + let worktree = self.work_root.join(format!("candidate-{commit_sha}")); + if worktree.exists() { + std::fs::remove_dir_all(&worktree) + .map_err(|error| format!("the stale candidate could not be removed: {error}"))?; + } + self.git(&[ + "clone", + "--quiet", + "--no-recurse-submodules", + remote, + worktree.to_str().unwrap_or_default(), + ])?; + self.git(&[ + "-C", + worktree.to_str().unwrap_or_default(), + "checkout", + "--quiet", + "--detach", + commit_sha, + ])?; + // Symlinks escape the isolated candidate: a repository carrying + // one is refused before any file is read. + reject_symlinks(&worktree)?; + // The digest covers the tracked file set: paths + contents, so + // two candidates are equal only when both the commit and the + // content match. + let content_digest = self.digest_worktree(&worktree)?; + // Collect the YAML files for validation. + let mut sources = Vec::new(); + collect_yaml(&worktree, &mut sources)?; + let diagnostics = validate(&sources); + Ok(Candidate { + digest: CandidateDigest { + commit_sha: commit_sha.to_owned(), + content_digest, + }, + worktree, + diagnostics, + }) + } + + /// Computes the SHA-256 of the tracked file set: sorted paths plus + /// contents, hashed as one stream. + /// Computes the digest of the tracked file set. + /// + /// # Errors + /// + /// Fails when a tracked file is unreadable. + pub fn digest_worktree(&self, worktree: &Path) -> Result { + use sha2::Digest as _; + let files = self.git(&["-C", worktree.to_str().unwrap_or_default(), "ls-files"])?; + let mut hasher = sha2::Sha256::new(); + for path in files.lines() { + hasher.update(path.as_bytes()); + hasher.update([0]); + let contents = std::fs::read(worktree.join(path)) + .map_err(|error| format!("the candidate file {path} is unreadable: {error}"))?; + hasher.update(&contents); + } + Ok(format!("{:x}", hasher.finalize())) + } +} + +/// Collects YAML files under a root, as relative paths. +fn collect_yaml(base: &Path, sources: &mut Vec) -> Result<(), String> { + let entries = std::fs::read_dir(base).map_err(|error| error.to_string())?; + for entry in entries { + let entry = entry.map_err(|error| error.to_string())?; + let path = entry.path(); + if path.is_dir() { + // Fleet's desired-state layout never descends into .git. + if path.file_name().is_some_and(|name| name == ".git") { + continue; + } + collect_yaml(&path, sources)?; + } else if path + .extension() + .is_some_and(|extension| extension == "yaml") + { + sources.push(path); + } + } + Ok(()) +} + +/// Refuses a repository carrying symlinks: a symlink escapes the isolated +/// candidate and makes the digest dependent on controller filesystem +/// contents. +fn reject_symlinks(base: &Path) -> Result<(), String> { + let entries = std::fs::read_dir(base).map_err(|error| error.to_string())?; + for entry in entries { + let entry = entry.map_err(|error| error.to_string())?; + let path = entry.path(); + let metadata = std::fs::symlink_metadata(&path) + .map_err(|error| format!("the candidate file is unreadable: {error}"))?; + if metadata.file_type().is_symlink() { + return Err(format!( + "the candidate carries a symlink at {}; symlinks are refused so the digest stays confined to the clone", + path.display() + )); + } + if metadata.is_dir() && path.file_name().is_some_and(|name| name != ".git") { + reject_symlinks(&path)?; + } + } + Ok(()) +} diff --git a/crates/providers/fleet-provider-github/Cargo.toml b/crates/providers/fleet-provider-github/Cargo.toml index 3d4f264..bcdf4c9 100644 --- a/crates/providers/fleet-provider-github/Cargo.toml +++ b/crates/providers/fleet-provider-github/Cargo.toml @@ -7,8 +7,14 @@ repository.workspace = true publish.workspace = true [dependencies] +async-trait = "0.1.92" fleet-application = { path = "../../fleet-application" } fleet-core = { path = "../../fleet-core" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" + +[dev-dependencies] +tokio = { version = "1", features = ["rt", "macros"] } [lints] workspace = true diff --git a/crates/providers/fleet-provider-github/src/lib.rs b/crates/providers/fleet-provider-github/src/lib.rs index b8e80c0..11f624e 100644 --- a/crates/providers/fleet-provider-github/src/lib.rs +++ b/crates/providers/fleet-provider-github/src/lib.rs @@ -1,6 +1,178 @@ -//! github provider adapter boundary. - +//! The GitHub provider (FM-403): the App web flow for one-click private +//! repository creation with least permissions and expiring tokens. +//! +//! The flow: the installation access token is requested through GitHub's +//! documented App API with an installation id scoped to the target +//! repository, carries `contents:read/write` on that repository only, +//! and expires (the response carries its own expiry). The token is +//! returned to the caller for storage in Fleet's encrypted secret store; +//! it is never logged, never audited with its value, and never persisted +//! here. +//! +//! Fleet's machine identity is never derived from GitHub: the App +//! installation is evidence for bootstrap, nothing more. #![warn(missing_docs)] -/// Skeleton marker proving that the provider crate is loadable. -pub const SKELETON: &str = "fleet-provider-github"; +use std::time::Duration; + +use serde::Deserialize; + +/// The contents permission the App requests: `contents:write` grants +/// read as well, per GitHub's documented permission values. Least +/// permissions: nothing else is requested. +pub const CONTENTS_PERMISSION: &str = "contents:write"; + +/// The transport contract: one HTTPS call, bounded. +#[async_trait::async_trait] +pub trait GithubTransport: std::fmt::Debug + Send + Sync { + /// Performs one POST to GitHub's installation-token endpoint. + /// + /// # Errors + /// + /// Fails on transport errors; an API refusal is an outcome. + async fn post_installation_token( + &self, + installation_id: &str, + body: &str, + deadline: Duration, + ) -> Result; +} + +/// One HTTP response, bounded. +#[derive(Clone, Debug)] +pub struct HttpResponse { + /// The status code. + pub status: u16, + /// The bounded body. + pub body: String, +} + +/// The installation token GitHub answers with. +#[derive(Debug, Deserialize)] +struct TokenDocument { + #[serde(default, alias = "token")] + value: Option, + #[serde(default, alias = "expiresAt")] + expires_at: Option, +} + +/// The outcome of one bootstrap-token request: the token and its expiry, +/// or an honest refusal. +#[derive(Clone, Eq, PartialEq)] +pub enum BootstrapOutcome { + /// The token was issued; it expires at the documented time. + Issued { + /// The token value, for the caller's encrypted store only. The + /// Debug impl never formats it. + token: String, + /// When the token expires, as GitHub documented it. + expires_at: String, + /// The permissions the token carries. + permissions: Vec, + }, + /// GitHub refused the request; the detail is bounded and redacted. + Refused { + /// The bounded, redacted detail. + detail: String, + }, +} + +impl std::fmt::Debug for BootstrapOutcome { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Issued { + expires_at, + permissions, + .. + } => formatter + .debug_struct("BootstrapOutcome::Issued") + .field("token", &"[redacted]") + .field("expires_at", expires_at) + .field("permissions", permissions) + .finish(), + Self::Refused { detail } => formatter + .debug_struct("BootstrapOutcome::Refused") + .field("detail", detail) + .finish(), + } + } +} + +/// Requests one bootstrap token for an App installation. The token +/// carries `contents:read_and_write` on the installation's repository +/// only, and expires — the caller stores it in Fleet's encrypted secret +/// store, never in logs or audit metadata. +/// +/// # Errors +/// +/// Fails on transport errors; a GitHub refusal is an outcome. +pub async fn request_bootstrap_token( + transport: &dyn GithubTransport, + installation_id: &str, + repository: &str, + deadline: Duration, +) -> Result { + // The repository is scoped in the request: an installation with + // access to more than the target must not yield a token for all of + // them. + let body = serde_json::json!({ + "permissions": { "contents": "write" }, + "repository": repository, + }) + .to_string(); + let outcome = transport + .post_installation_token(installation_id, &body, deadline) + .await?; + // The size bound runs BEFORE the status branch, so refusal details + // stay bounded as documented. + if outcome.body.len() > 64 * 1024 { + return Ok(BootstrapOutcome::Refused { + detail: "the token document exceeds its bound".to_owned(), + }); + } + if outcome.status != 201 { + return Ok(BootstrapOutcome::Refused { + detail: redact(&outcome.body), + }); + } + // An unparseable document is a protocol-level refusal, not a + // transport failure: the honest outcome is `Refused`. + let Ok(document) = serde_json::from_str::(outcome.body.trim()) else { + return Ok(BootstrapOutcome::Refused { + detail: "the token document is not in the documented shape".to_owned(), + }); + }; + let Some(token) = document.value.filter(|token| !token.is_empty()) else { + return Ok(BootstrapOutcome::Refused { + detail: "the token document carries no token".to_owned(), + }); + }; + // The documented expiry is required: a token without one cannot be + // stored as expiring, which is the whole security contract. + let Some(expires_at) = document.expires_at.filter(|expiry| !expiry.is_empty()) else { + return Ok(BootstrapOutcome::Refused { + detail: + "the token document carries no expiry; a token without an expiry cannot be stored" + .to_owned(), + }); + }; + Ok(BootstrapOutcome::Issued { + // The token itself is returned for the encrypted store only; the + // expiry and permissions are safe to carry. + token, + expires_at, + permissions: vec![CONTENTS_PERMISSION.to_owned()], + }) +} + +/// Scrubs credential-shaped material from GitHub's error output before +/// it becomes a detail. +#[must_use] +pub fn redact(text: &str) -> String { + let cleaned: String = text + .chars() + .map(|c| if c.is_control() && c != '\n' { ' ' } else { c }) + .collect(); + let with_urls = fleet_core::redact_url_credentials(&cleaned); + fleet_core::redact_schemeless_credentials(&with_urls) +} diff --git a/crates/providers/fleet-provider-github/tests/fixtures.rs b/crates/providers/fleet-provider-github/tests/fixtures.rs new file mode 100644 index 0000000..839a259 --- /dev/null +++ b/crates/providers/fleet-provider-github/tests/fixtures.rs @@ -0,0 +1,155 @@ +//! Contract fixtures for the GitHub App flow (FM-403): least +//! permissions, expiring tokens, and redacted refusals. + +use fleet_provider_github::{ + BootstrapOutcome, GithubTransport, HttpResponse, request_bootstrap_token, +}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +#[derive(Debug, Clone)] +struct FakeTransport { + status: u16, + body: String, + seen: Arc>>, +} + +#[async_trait::async_trait] +impl GithubTransport for FakeTransport { + async fn post_installation_token( + &self, + installation_id: &str, + body: &str, + _deadline: Duration, + ) -> Result { + self.seen + .lock() + .unwrap() + .push((installation_id.to_owned(), body.to_owned())); + Ok(HttpResponse { + status: self.status, + body: self.body.clone(), + }) + } +} + +#[tokio::test] +async fn the_token_request_carries_least_permissions() { + let seen = Arc::new(Mutex::new(Vec::new())); + let transport = FakeTransport { + status: 201, + body: r#"{"token":"ghs_example","expiresAt":"2026-09-19T00:00:00Z"}"#.to_owned(), + seen: seen.clone(), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Issued { + token, + expires_at, + permissions, + } = outcome + else { + panic!("the token is issued"); + }; + assert_eq!(token, "ghs_example"); + assert_eq!(expires_at, "2026-09-19T00:00:00Z"); + assert_eq!(permissions, ["contents:write"]); + let (installation, body) = &seen.lock().unwrap()[0]; + assert_eq!(installation, "12345"); + assert!( + body.contains(r#""contents":"write""#), + "the request carries contents write only: {body}" + ); + assert!( + !body.contains("admin"), + "no admin permissions are requested" + ); +} + +#[tokio::test] +async fn a_refusal_is_an_honest_outcome_with_redacted_detail() { + let transport = FakeTransport { + status: 404, + body: r#"{"message":"Not Found: https://user:secret@host.invalid"}"#.to_owned(), + seen: Arc::new(Mutex::new(Vec::new())), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Refused { detail } = outcome else { + panic!("the refusal is an outcome"); + }; + assert!(!detail.contains("secret"), "{detail}"); + assert!(detail.contains("***@host.invalid"), "{detail}"); +} + +#[tokio::test] +async fn a_document_without_a_token_is_refused() { + let transport = FakeTransport { + status: 201, + body: r#"{"expiresAt":"2026-09-19T00:00:00Z"}"#.to_owned(), + seen: Arc::new(Mutex::new(Vec::new())), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Refused { detail } = outcome else { + panic!("the refusal is an outcome"); + }; + assert!(detail.contains("no token"), "{detail}"); +} + +#[tokio::test] +async fn a_document_without_an_expiry_is_refused() { + let transport = FakeTransport { + status: 201, + body: r#"{"token":"ghs_example"}"#.to_owned(), + seen: Arc::new(Mutex::new(Vec::new())), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Refused { detail } = outcome else { + panic!("the refusal is an outcome"); + }; + assert!(detail.contains("no expiry"), "{detail}"); +} + +#[tokio::test] +async fn an_unparseable_document_is_a_refusal_not_a_transport_failure() { + let transport = FakeTransport { + status: 201, + body: "not json".to_owned(), + seen: Arc::new(Mutex::new(Vec::new())), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Refused { detail } = outcome else { + panic!("the refusal is an outcome"); + }; + assert!(detail.contains("documented shape"), "{detail}"); +} + +#[tokio::test] +async fn an_oversized_document_is_refused() { + let transport = FakeTransport { + status: 201, + body: "x".repeat(128 * 1024), + seen: Arc::new(Mutex::new(Vec::new())), + }; + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); + let BootstrapOutcome::Refused { detail } = outcome else { + panic!("the refusal is an outcome"); + }; + assert!(detail.contains("exceeds its bound"), "{detail}"); +} diff --git a/crates/providers/fleet-provider-ssh/tests/trust.rs b/crates/providers/fleet-provider-ssh/tests/trust.rs index d0f6f57..dba3f63 100644 --- a/crates/providers/fleet-provider-ssh/tests/trust.rs +++ b/crates/providers/fleet-provider-ssh/tests/trust.rs @@ -7,6 +7,10 @@ use std::net::TcpListener; use std::process::{Child, Command}; use std::time::Duration; +/// The port-allocation lock: the free-port window between allocation and +/// sshd's bind is racy across the parallel test threads of one binary. +static STARTUP_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + /// One running sshd bound to an ephemeral port with its own host key. struct TestSshd { child: Child, @@ -38,6 +42,9 @@ fn free_port() -> u16 { /// current user, authenticating via the agent or an unencrypted key we also /// generate here. fn start_sshd() -> TestSshd { + let _guard = STARTUP_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let dir = tempfile::tempdir().unwrap(); let host_key = dir.path().join("host_ed25519"); let user_key = dir.path().join("user_ed25519");