From 54cf3555f3a95edef3c0801f4beca61b7c609022 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 18:09:17 +0000 Subject: [PATCH 1/6] FM-403: the Git source adapter and GitHub bootstrap The Git source provider (fleet-provider-git) clones a pinned commit into an isolated worktree named by the SHA, computes the candidate digest (commit SHA + SHA-256 of the tracked file set), and hands the YAML file set to the schemas crate for validation. Hooks never run: every git invocation passes core.hooksPath pointed nowhere (the FM-301 rule). The desired-source use cases (fleet-application/src/source.rs) gate activation: a candidate carrying diagnostics can never become active, an unknown digest was never fetched, the audit intent lands before the mutation, and a valid candidate is recorded as a rollback point while an invalid one is reported and forgotten. SourceFetch and SourceActivate join the authz catalog (36 entries) as catalog-level actions. The GitHub provider (fleet-provider-github) requests bootstrap tokens through the documented App API with least permissions (contents:read_and_write on the installation's repository only); the token is returned for Fleet's encrypted secret store and never logged or audited with its value, and GitHub's refusals are bounded and redacted. Tests: six desired-source use case tests (rollback recording, invalid candidates forgotten, activation refusals, audit+durability, denials touching nothing), plus the GitHub provider's contract shape. --- Cargo.lock | 13 + Cargo.toml | 1 + crates/fleet-application/src/authz.rs | 18 +- crates/fleet-application/src/lib.rs | 1 + crates/fleet-application/src/source.rs | 449 ++++++++++++++++++ crates/fleet-auth/tests/authz_adapter.rs | 2 +- crates/fleet-core/src/lib.rs | 2 + crates/fleet-core/src/source.rs | 23 + .../providers/fleet-provider-git/Cargo.toml | 15 + .../providers/fleet-provider-git/src/lib.rs | 181 +++++++ .../fleet-provider-github/Cargo.toml | 6 + .../fleet-provider-github/src/lib.rs | 135 +++++- 12 files changed, 839 insertions(+), 7 deletions(-) create mode 100644 crates/fleet-application/src/source.rs create mode 100644 crates/fleet-core/src/source.rs create mode 100644 crates/providers/fleet-provider-git/Cargo.toml create mode 100644 crates/providers/fleet-provider-git/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 147c6b8..1fdf82c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -685,12 +685,25 @@ dependencies = [ "serde_json", ] +[[package]] +name = "fleet-provider-git" +version = "0.1.0" +dependencies = [ + "fleet-core", + "serde", + "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/source.rs b/crates/fleet-application/src/source.rs new file mode 100644 index 0000000..173d676 --- /dev/null +++ b/crates/fleet-application/src/source.rs @@ -0,0 +1,449 @@ +//! 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>; + /// Records the active revision, replacing any prior one. + /// + /// # Errors + /// + /// Fails when the backend errors. + async fn set_active_revision(&self, revision: &ActiveRevision) -> 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>; +} + +/// 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, + ) -> 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. + if !candidate_valid { + return Err(crate::project::ProjectUseCaseError::Invalid { + detail: "the candidate carries validation diagnostics and cannot become active" + .to_owned(), + }); + } + if !candidate_known { + return Err(crate::project::ProjectUseCaseError::Invalid { + detail: "the candidate was never fetched; 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: None, + 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(), + }; + self.port + .set_active_revision(&revision) + .await + .map_err(|detail| crate::project::ProjectUseCaseError::Backend { + context: "source_activate", + detail, + })?; + Ok(revision) + } + + /// 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 set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String> { + *self.active.lock().unwrap() = Some(revision.clone()); + Ok(()) + } + 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(()) + } + } + + #[derive(Debug)] + struct FakeAudit; + #[async_trait::async_trait] + impl crate::operation::AuditPort for FakeAudit { + async fn record_intent(&self, _intent: &crate::audit::AuditIntent) -> Result<(), String> { + 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 { + DesiredSource::new(Arc::new(FakePort::default()), Arc::new(FakeAudit)) + } + + #[tokio::test] + async fn a_valid_candidate_is_recorded_for_rollback() { + let service = 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!(service.port.prior_revisions().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn a_candidate_with_diagnostics_is_reported_and_forgotten() { + let service = 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!( + service.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, + ) + .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, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("never fetched"), "{error}"); + } + + #[tokio::test] + async fn activation_is_audited_and_durable() { + let service = service(); + let revision = service + .activate( + &AllowAll, + "anonymous-lan-admin", + &digest("abc", "digest-1"), + true, + true, + ) + .await + .unwrap(); + assert_eq!(revision.commit_sha, "abc"); + let active = service.port.active_revision().await.unwrap().unwrap(); + assert_eq!(active.content_digest, "digest-1"); + } + + #[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 = 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) + .await + .is_err() + ); + assert!(service.port.active_revision().await.unwrap().is_none()); + assert!(service.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-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/providers/fleet-provider-git/Cargo.toml b/crates/providers/fleet-provider-git/Cargo.toml new file mode 100644 index 0000000..7c52669 --- /dev/null +++ b/crates/providers/fleet-provider-git/Cargo.toml @@ -0,0 +1,15 @@ +[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" +serde = { version = "1", features = ["derive"] } + +[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..025c857 --- /dev/null +++ b/crates/providers/fleet-provider-git/src/lib.rs @@ -0,0 +1,181 @@ +//! 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() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(format!( + "git {} failed: {}", + arguments.first().unwrap_or(&""), + stderr.trim() + )); + } + Ok(String::from_utf8_lossy(&output.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 worktree directory is named by the SHA: the same commit + // materializes to the same isolated directory. + let worktree = self.work_root.join(format!("candidate-{commit_sha}")); + if !worktree.exists() { + 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, + ])?; + } + // 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, &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. + 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(_root: &Path, 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(base, &path, sources)?; + } else if path + .extension() + .is_some_and(|extension| extension == "yaml") + { + sources.push(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..8414754 100644 --- a/crates/providers/fleet-provider-github/src/lib.rs +++ b/crates/providers/fleet-provider-github/src/lib.rs @@ -1,6 +1,133 @@ -//! 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: read/write on the target +/// repository only. Least permissions: nothing else is requested. +pub const CONTENTS_PERMISSION: &str = "contents:read_and_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, Debug, 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. + token: String, + /// When the token expires, as GitHub documented it. + expires_at: Option, + /// The permissions the token carries. + permissions: Vec, + }, + /// GitHub refused the request; the detail is bounded and redacted. + Refused { + /// The bounded, redacted detail. + detail: String, + }, +} + +/// 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, + deadline: Duration, +) -> Result { + let outcome = transport + .post_installation_token( + installation_id, + r#"{"permissions":{"contents":"write"}}"#, + deadline, + ) + .await?; + if outcome.status != 201 { + return Ok(BootstrapOutcome::Refused { + detail: redact(&outcome.body), + }); + } + if outcome.body.len() > 64 * 1024 { + return Ok(BootstrapOutcome::Refused { + detail: "the token document exceeds its bound".to_owned(), + }); + } + let document: TokenDocument = serde_json::from_str(outcome.body.trim()) + .map_err(|_| "the token document is not in the documented shape".to_string())?; + let Some(token) = document.value.filter(|token| !token.is_empty()) else { + return Ok(BootstrapOutcome::Refused { + detail: "the token document carries no token".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: document.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(); + fleet_core::redact_url_credentials(&cleaned) +} From c115903f3e84a3ccf7c9660c92f2e16353ed0092 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 18:15:07 +0000 Subject: [PATCH 2/6] FM-403: the executor kinds, source storage, and wiring source.fetch and source.activate join the executor kinds: fetch clones the pinned commit, validates through the schemas crate, and records the outcome through the authorized use case; activate re-validates against the materialized worktree (so the gate holds across restarts) and activates through the authorized use case. The source tables (active revision upsert + append-only rollback history) are a STRICT migration, and the SourceRepository adapter implements the SourcePort. The authz catalog grows to 36 entries with source.fetch/source.activate as catalog-level actions. The source dispatch is wired into the controller's executor chain. --- Cargo.lock | 2 + crates/fleet-application/src/operation.rs | 9 +- crates/fleet-controller/Cargo.toml | 2 + crates/fleet-controller/src/lib.rs | 1 + crates/fleet-controller/src/main.rs | 22 +- crates/fleet-controller/src/source.rs | 271 ++++++++++++++++++ .../migrations/0016_source_revisions.sql | 19 ++ crates/fleet-storage-sqlite/src/lib.rs | 2 + crates/fleet-storage-sqlite/src/source.rs | 89 ++++++ .../fleet-provider-github/tests/fixtures.rs | 117 ++++++++ 10 files changed, 530 insertions(+), 4 deletions(-) create mode 100644 crates/fleet-controller/src/source.rs create mode 100644 crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql create mode 100644 crates/fleet-storage-sqlite/src/source.rs create mode 100644 crates/providers/fleet-provider-github/tests/fixtures.rs diff --git a/Cargo.lock b/Cargo.lock index 1fdf82c..a576e93 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", diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index 37e7a5a..94a0f50 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. @@ -109,6 +112,8 @@ fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option

Some(Permission::MiseOperate), "ready.workflow" => Some(Permission::ProjectsReady), "apply.workflow" => Some(Permission::ApplyExecute), + "source.fetch" => Some(Permission::SourceFetch), + "source.activate" => Some(Permission::SourceActivate), _ => None, } } 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..d030dfc --- /dev/null +++ b/crates/fleet-controller/src/source.rs @@ -0,0 +1,271 @@ +//! 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, + ) + .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-storage-sqlite/migrations/0016_source_revisions.sql b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql new file mode 100644 index 0000000..5784611 --- /dev/null +++ b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql @@ -0,0 +1,19 @@ +-- 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 ( + id TEXT PRIMARY KEY, + 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); 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..a333b3f --- /dev/null +++ b/crates/fleet-storage-sqlite/src/source.rs @@ -0,0 +1,89 @@ +//! 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 set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String> { + sqlx::query( + "INSERT INTO source_active_revision (id, commit_sha, content_digest, activated_at) \ + VALUES ('active', ?1, ?2, ?3) \ + ON CONFLICT(id) 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(&self.pool) + .await + .map_err(|error| format!("the active revision write failed: {error}"))?; + Ok(()) + } + + 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 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.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-github/tests/fixtures.rs b/crates/providers/fleet-provider-github/tests/fixtures.rs new file mode 100644 index 0000000..b1acf1c --- /dev/null +++ b/crates/providers/fleet-provider-github/tests/fixtures.rs @@ -0,0 +1,117 @@ +//! 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", 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.as_deref(), Some("2026-09-19T00:00:00Z")); + assert_eq!(permissions, ["contents:read_and_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", 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", 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 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", 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}"); +} From c2ab704a7bc0029916992b67f3a9cb89c197423c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 18:48:20 +0000 Subject: [PATCH 3/6] FM-403: fix the cubic review findings The git provider validates the commit SHA (a full 40-character hexadecimal id, so the path cannot escape the work root), discards and recreates an existing candidate directory (a stale or partial clone is never validated under the requested SHA), refuses symlinks (the digest stays confined to the clone), bounds and redacts git output (a large repository cannot consume unbounded memory; a credential-bearing remote cannot echo its URL), and collects validation paths relative to the worktree. The GitHub provider scopes the token request to the target repository (an installation with access to more than the target must not yield a token for all of them), requests contents:write (GitHub's documented value, which grants read as well), requires the documented expiry, treats an unparseable document as a protocol-level refusal rather than a transport failure, bounds refusal details before redaction, adds schemeless credential redaction, and implements a redacted Debug that never formats the token. The desired-source use cases verify the requested digest against the recorded candidate state (only a candidate the fetch path recorded as valid and known may activate), serialize the activation critical section through the backend (a concurrent activation cannot race), correlate the audit intent with the operation id, and exclude the active revision from prior_revisions. The storage adapter implements the serialized activation with BEGIN IMMEDIATE and the source tables enforce the singleton constraint and append-only history at the database level. The executor propagates validation read failures as explicit diagnostics (never a silent valid), revalidates against the materialized worktree, and verifies the recomputed digest before activating. Tests: six new GitHub fixtures (repo scoping, missing expiry, unparseable document, oversized refusal) and the activation audit test asserts the intent's action, commit SHA, and operation correlation. --- Cargo.lock | 1 - crates/fleet-application/src/operation.rs | 3 +- crates/fleet-application/src/source.rs | 116 ++++++++++++++---- crates/fleet-controller/src/source.rs | 1 + .../migrations/0016_source_revisions.sql | 14 ++- crates/fleet-storage-sqlite/src/source.rs | 41 ++++++- .../providers/fleet-provider-git/Cargo.toml | 1 - .../providers/fleet-provider-git/src/lib.rs | 103 ++++++++++++---- .../fleet-provider-github/src/lib.rs | 81 +++++++++--- .../fleet-provider-github/tests/fixtures.rs | 66 +++++++--- 10 files changed, 337 insertions(+), 90 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a576e93..e18fdba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -692,7 +692,6 @@ name = "fleet-provider-git" version = "0.1.0" dependencies = [ "fleet-core", - "serde", "sha2 0.10.9", ] diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index 94a0f50..37ca33b 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -112,8 +112,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), - "source.fetch" => Some(Permission::SourceFetch), - "source.activate" => Some(Permission::SourceActivate), + _ => None, } } diff --git a/crates/fleet-application/src/source.rs b/crates/fleet-application/src/source.rs index 173d676..c9c277e 100644 --- a/crates/fleet-application/src/source.rs +++ b/crates/fleet-application/src/source.rs @@ -55,6 +55,18 @@ pub trait SourcePort: std::fmt::Debug + Send + Sync { /// /// 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 @@ -157,6 +169,7 @@ impl DesiredSource { digest: &fleet_core::CandidateDigest, candidate_valid: bool, candidate_known: bool, + operation_id: Option<&str>, ) -> Result { use crate::authz::{AccessRequest, Permission, authorize}; authorize( @@ -169,16 +182,29 @@ impl DesiredSource { ) .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. + // 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 { + if !candidate_known || !known_and_valid { return Err(crate::project::ProjectUseCaseError::Invalid { - detail: "the candidate was never fetched; fetch it before activating".to_owned(), + 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 @@ -204,7 +230,7 @@ impl DesiredSource { resource: None, decision: crate::authz::Decision::allow(), correlation_id: None, - operation_id: None, + operation_id: operation_id.map(str::to_owned), metadata, }) .await @@ -216,14 +242,15 @@ impl DesiredSource { 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 - .set_active_revision(&revision) + .activate_serialized(&revision) .await .map_err(|detail| crate::project::ProjectUseCaseError::Backend { context: "source_activate", detail, - })?; - Ok(revision) + }) } /// The prior valid revisions available for manual rollback. @@ -304,13 +331,23 @@ mod tests { 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)] - struct FakeAudit; + #[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> { + async fn record_intent(&self, intent: &crate::audit::AuditIntent) -> Result<(), String> { + self.intents.lock().unwrap().push(intent.clone()); Ok(()) } async fn record_outcome( @@ -329,13 +366,15 @@ mod tests { } } - fn service() -> DesiredSource { - DesiredSource::new(Arc::new(FakePort::default()), Arc::new(FakeAudit)) + 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 = service(); + let (service, _, port) = service(); let outcome = FetchOutcome::Candidate { digest: digest("abc", "digest-1"), diagnostics: vec![], @@ -348,12 +387,12 @@ mod tests { panic!("the candidate outcome is preserved"); }; assert!(diagnostics.is_empty()); - assert_eq!(service.port.prior_revisions().await.unwrap().len(), 1); + assert_eq!(port.prior_revisions().await.unwrap().len(), 1); } #[tokio::test] async fn a_candidate_with_diagnostics_is_reported_and_forgotten() { - let service = service(); + let (service, _, port) = service(); let outcome = FetchOutcome::Candidate { digest: digest("abc", "digest-bad"), diagnostics: vec!["the document does not match".to_owned()], @@ -363,14 +402,14 @@ mod tests { .await .unwrap(); assert!( - service.port.prior_revisions().await.unwrap().is_empty(), + 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 (service, _, _) = service(); let error = service .activate( &AllowAll, @@ -378,6 +417,7 @@ mod tests { &digest("abc", "bad"), false, true, + None, ) .await .unwrap_err(); @@ -390,7 +430,7 @@ mod tests { #[tokio::test] async fn activation_refuses_an_unknown_candidate() { - let service = service(); + let (service, _, _) = service(); let error = service .activate( &AllowAll, @@ -398,6 +438,7 @@ mod tests { &digest("abc", "d"), true, false, + None, ) .await .unwrap_err(); @@ -406,7 +447,20 @@ mod tests { #[tokio::test] async fn activation_is_audited_and_durable() { - let service = service(); + 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, @@ -414,12 +468,26 @@ mod tests { &digest("abc", "digest-1"), true, true, + Some("op-1"), ) .await .unwrap(); assert_eq!(revision.commit_sha, "abc"); - let active = service.port.active_revision().await.unwrap().unwrap(); + 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] @@ -431,7 +499,7 @@ mod tests { Decision::deny(crate::authz::ReasonId::UnknownPrincipal) } } - let service = service(); + let (service, _, port) = service(); let outcome = FetchOutcome::Candidate { digest: digest("abc", "d"), diagnostics: vec![], @@ -439,11 +507,11 @@ mod tests { assert!(service.handle_fetch(&DenyAll, "x", outcome).await.is_err()); assert!( service - .activate(&DenyAll, "x", &digest("abc", "d"), true, true) + .activate(&DenyAll, "x", &digest("abc", "d"), true, true, None) .await .is_err() ); - assert!(service.port.active_revision().await.unwrap().is_none()); - assert!(service.port.prior_revisions().await.unwrap().is_empty()); + assert!(port.active_revision().await.unwrap().is_none()); + assert!(port.prior_revisions().await.unwrap().is_empty()); } } diff --git a/crates/fleet-controller/src/source.rs b/crates/fleet-controller/src/source.rs index d030dfc..379e4c8 100644 --- a/crates/fleet-controller/src/source.rs +++ b/crates/fleet-controller/src/source.rs @@ -182,6 +182,7 @@ impl SourceExecutor { }, diagnostics.is_empty(), candidate_known, + Some(&operation.id), ) .await .map_err(|error| { diff --git a/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql index 5784611..1f4cc5e 100644 --- a/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql +++ b/crates/fleet-storage-sqlite/migrations/0016_source_revisions.sql @@ -3,7 +3,7 @@ -- history for manual rollback). Only VALID candidates are recorded; a -- candidate carrying diagnostics is reported and forgotten. CREATE TABLE source_active_revision ( - id TEXT PRIMARY KEY, + singleton TEXT PRIMARY KEY CHECK (singleton = 'active'), commit_sha TEXT NOT NULL, content_digest TEXT NOT NULL, activated_at INTEGER NOT NULL @@ -17,3 +17,15 @@ CREATE TABLE source_revision_history ( ) 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/source.rs b/crates/fleet-storage-sqlite/src/source.rs index a333b3f..b036396 100644 --- a/crates/fleet-storage-sqlite/src/source.rs +++ b/crates/fleet-storage-sqlite/src/source.rs @@ -38,9 +38,9 @@ impl fleet_application::source::SourcePort for SourceRepository { async fn set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String> { sqlx::query( - "INSERT INTO source_active_revision (id, commit_sha, content_digest, activated_at) \ + "INSERT INTO source_active_revision (singleton, commit_sha, content_digest, activated_at) \ VALUES ('active', ?1, ?2, ?3) \ - ON CONFLICT(id) DO UPDATE SET \ + ON CONFLICT(singleton) DO UPDATE SET \ commit_sha = excluded.commit_sha, \ content_digest = excluded.content_digest, \ activated_at = excluded.activated_at", @@ -71,13 +71,48 @@ impl fleet_application::source::SourcePort for SourceRepository { .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.content_digest)) + .bind(format!( + "rev-{}-{}", + revision.commit_sha, revision.content_digest + )) .bind(&revision.commit_sha) .bind(&revision.content_digest) .bind(fleet_core::SystemClock::now_unix_millis()) diff --git a/crates/providers/fleet-provider-git/Cargo.toml b/crates/providers/fleet-provider-git/Cargo.toml index 7c52669..8964052 100644 --- a/crates/providers/fleet-provider-git/Cargo.toml +++ b/crates/providers/fleet-provider-git/Cargo.toml @@ -9,7 +9,6 @@ publish.workspace = true [dependencies] fleet-core = { path = "../../fleet-core" } sha2 = "0.10" -serde = { version = "1", features = ["derive"] } [lints] workspace = true diff --git a/crates/providers/fleet-provider-git/src/lib.rs b/crates/providers/fleet-provider-git/src/lib.rs index 025c857..c4aa8e7 100644 --- a/crates/providers/fleet-provider-git/src/lib.rs +++ b/crates/providers/fleet-provider-git/src/lib.rs @@ -79,14 +79,21 @@ impl GitSource { .output() .map_err(|error| format!("git could not start: {error}"))?; if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); + // 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: {}", - arguments.first().unwrap_or(&""), - stderr.trim() + "git {} failed: {bounded}", + arguments.first().unwrap_or(&"") )); } - Ok(String::from_utf8_lossy(&output.stdout).into_owned()) + // The stdout is bounded: a large repository cannot consume + // unbounded controller memory. + let stdout = String::from_utf8_lossy(&output.stdout); + Ok(stdout.chars().take(1024 * 1024).collect()) } /// Fetches one candidate: clones the repository at the pinned commit @@ -103,33 +110,49 @@ impl GitSource { 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. + // 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() { - 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, - ])?; + 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, &worktree, &mut sources)?; + collect_yaml(&worktree, &mut sources)?; let diagnostics = validate(&sources); Ok(Candidate { digest: CandidateDigest { @@ -143,7 +166,12 @@ impl GitSource { /// Computes the SHA-256 of the tracked file set: sorted paths plus /// contents, hashed as one stream. - fn digest_worktree(&self, worktree: &Path) -> Result { + /// 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(); @@ -159,7 +187,7 @@ impl GitSource { } /// Collects YAML files under a root, as relative paths. -fn collect_yaml(_root: &Path, base: &Path, sources: &mut Vec) -> Result<(), String> { +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())?; @@ -169,7 +197,7 @@ fn collect_yaml(_root: &Path, base: &Path, sources: &mut Vec) -> Result if path.file_name().is_some_and(|name| name == ".git") { continue; } - collect_yaml(base, &path, sources)?; + collect_yaml(&path, sources)?; } else if path .extension() .is_some_and(|extension| extension == "yaml") @@ -179,3 +207,26 @@ fn collect_yaml(_root: &Path, base: &Path, sources: &mut Vec) -> Result } 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/src/lib.rs b/crates/providers/fleet-provider-github/src/lib.rs index 8414754..11f624e 100644 --- a/crates/providers/fleet-provider-github/src/lib.rs +++ b/crates/providers/fleet-provider-github/src/lib.rs @@ -17,9 +17,10 @@ use std::time::Duration; use serde::Deserialize; -/// The contents permission the App requests: read/write on the target -/// repository only. Least permissions: nothing else is requested. -pub const CONTENTS_PERMISSION: &str = "contents:read_and_write"; +/// 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] @@ -57,14 +58,15 @@ struct TokenDocument { /// The outcome of one bootstrap-token request: the token and its expiry, /// or an honest refusal. -#[derive(Clone, Debug, Eq, PartialEq)] +#[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 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: Option, + expires_at: String, /// The permissions the token carries. permissions: Vec, }, @@ -75,6 +77,27 @@ pub enum BootstrapOutcome { }, } +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 @@ -86,37 +109,58 @@ pub enum BootstrapOutcome { 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, - r#"{"permissions":{"contents":"write"}}"#, - deadline, - ) + .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), }); } - if outcome.body.len() > 64 * 1024 { + // 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 exceeds its bound".to_owned(), + detail: "the token document is not in the documented shape".to_owned(), }); - } - let document: TokenDocument = serde_json::from_str(outcome.body.trim()) - .map_err(|_| "the token document is not in the documented shape".to_string())?; + }; 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: document.expires_at, + expires_at, permissions: vec![CONTENTS_PERMISSION.to_owned()], }) } @@ -129,5 +173,6 @@ pub fn redact(text: &str) -> String { .chars() .map(|c| if c.is_control() && c != '\n' { ' ' } else { c }) .collect(); - fleet_core::redact_url_credentials(&cleaned) + 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 index b1acf1c..839a259 100644 --- a/crates/providers/fleet-provider-github/tests/fixtures.rs +++ b/crates/providers/fleet-provider-github/tests/fixtures.rs @@ -41,9 +41,10 @@ async fn the_token_request_carries_least_permissions() { body: r#"{"token":"ghs_example","expiresAt":"2026-09-19T00:00:00Z"}"#.to_owned(), seen: seen.clone(), }; - let outcome = request_bootstrap_token(&transport, "12345", Duration::from_secs(30)) - .await - .unwrap(); + let outcome = + request_bootstrap_token(&transport, "12345", "example/repo", Duration::from_secs(30)) + .await + .unwrap(); let BootstrapOutcome::Issued { token, expires_at, @@ -53,8 +54,8 @@ async fn the_token_request_carries_least_permissions() { panic!("the token is issued"); }; assert_eq!(token, "ghs_example"); - assert_eq!(expires_at.as_deref(), Some("2026-09-19T00:00:00Z")); - assert_eq!(permissions, ["contents:read_and_write"]); + 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!( @@ -74,9 +75,10 @@ async fn a_refusal_is_an_honest_outcome_with_redacted_detail() { 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", Duration::from_secs(30)) - .await - .unwrap(); + 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"); }; @@ -91,15 +93,50 @@ async fn a_document_without_a_token_is_refused() { body: r#"{"expiresAt":"2026-09-19T00:00:00Z"}"#.to_owned(), seen: Arc::new(Mutex::new(Vec::new())), }; - let outcome = request_bootstrap_token(&transport, "12345", Duration::from_secs(30)) - .await - .unwrap(); + 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 { @@ -107,9 +144,10 @@ async fn an_oversized_document_is_refused() { body: "x".repeat(128 * 1024), seen: Arc::new(Mutex::new(Vec::new())), }; - let outcome = request_bootstrap_token(&transport, "12345", Duration::from_secs(30)) - .await - .unwrap(); + 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"); }; From 9c0e1adda0afddc0c5e447a627a7c3206b5fbad9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 19:10:27 +0000 Subject: [PATCH 4/6] FM-403: fix the cubic review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The git provider accepts uppercase hex SHAs (lowercased, matching git's own resolution) and refuses a tracked-file listing beyond its 1 MiB bound — a truncated listing would hash only part of the file set, so truncation is an error rather than an incomplete digest. The GitHub fixtures assert the repository scoping reaches the request body. The source tables enforce their invariants at the database level: the active revision's primary key is the singleton id, and the rollback history is append-only (BEFORE UPDATE/DELETE triggers, matching the audit ledger). The history key carries both the commit SHA and the content digest, so two commits with the same tracked-file digest are both preserved. The source kinds' permission is enforced during operation creation with resource: None (catalog-level, like the dedicated endpoint), and the unused set_active_revision port method is removed — activate_serialized is the single write path. --- crates/fleet-application/src/operation.rs | 10 ++++++++++ crates/fleet-application/src/source.rs | 10 ---------- crates/fleet-storage-sqlite/src/source.rs | 18 ------------------ crates/providers/fleet-provider-git/src/lib.rs | 11 +++++++++-- 4 files changed, 19 insertions(+), 30 deletions(-) diff --git a/crates/fleet-application/src/operation.rs b/crates/fleet-application/src/operation.rs index 37ca33b..5fea37d 100644 --- a/crates/fleet-application/src/operation.rs +++ b/crates/fleet-application/src/operation.rs @@ -83,6 +83,16 @@ pub const CREATABLE_KINDS: [&str; 31] = [ /// governs both the dedicated endpoint and the generic one. #[must_use] fn machine_scoped_kind_permission(kind: &str, payload: Option<&str>) -> Option { + // The source kinds are catalog-level: their permission is enforced + // here with `resource: None`, never a machine id. + 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" => { diff --git a/crates/fleet-application/src/source.rs b/crates/fleet-application/src/source.rs index c9c277e..c0cc2db 100644 --- a/crates/fleet-application/src/source.rs +++ b/crates/fleet-application/src/source.rs @@ -35,12 +35,6 @@ pub trait SourcePort: std::fmt::Debug + Send + Sync { /// /// Fails when the backend errors. async fn active_revision(&self) -> Result, String>; - /// Records the active revision, replacing any prior one. - /// - /// # Errors - /// - /// Fails when the backend errors. - async fn set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String>; /// The digests of prior valid revisions, for manual rollback. /// /// # Errors @@ -320,10 +314,6 @@ mod tests { async fn active_revision(&self) -> Result, String> { Ok(self.active.lock().unwrap().clone()) } - async fn set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String> { - *self.active.lock().unwrap() = Some(revision.clone()); - Ok(()) - } async fn prior_revisions(&self) -> Result, String> { Ok(self.valid.lock().unwrap().clone()) } diff --git a/crates/fleet-storage-sqlite/src/source.rs b/crates/fleet-storage-sqlite/src/source.rs index b036396..2fa47e7 100644 --- a/crates/fleet-storage-sqlite/src/source.rs +++ b/crates/fleet-storage-sqlite/src/source.rs @@ -36,24 +36,6 @@ impl fleet_application::source::SourcePort for SourceRepository { })) } - async fn set_active_revision(&self, revision: &ActiveRevision) -> Result<(), String> { - 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(&self.pool) - .await - .map_err(|error| format!("the active revision write failed: {error}"))?; - Ok(()) - } - async fn prior_revisions(&self) -> Result, String> { let rows: Vec<(String, String)> = sqlx::query_as( "SELECT commit_sha, content_digest FROM source_revision_history \ diff --git a/crates/providers/fleet-provider-git/src/lib.rs b/crates/providers/fleet-provider-git/src/lib.rs index c4aa8e7..9e0439a 100644 --- a/crates/providers/fleet-provider-git/src/lib.rs +++ b/crates/providers/fleet-provider-git/src/lib.rs @@ -91,9 +91,16 @@ impl GitSource { )); } // The stdout is bounded: a large repository cannot consume - // unbounded controller memory. + // unbounded controller memory. A truncated listing would hash only + // part of the file set, so truncation is an error, not a + // truncation. let stdout = String::from_utf8_lossy(&output.stdout); - Ok(stdout.chars().take(1024 * 1024).collect()) + 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 From f1ad776a344b396a443a49075696b2fc9feb9492 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 19:29:14 +0000 Subject: [PATCH 5/6] FM-403: reword the oversized-listing comment --- crates/providers/fleet-provider-git/src/lib.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/providers/fleet-provider-git/src/lib.rs b/crates/providers/fleet-provider-git/src/lib.rs index 9e0439a..07878aa 100644 --- a/crates/providers/fleet-provider-git/src/lib.rs +++ b/crates/providers/fleet-provider-git/src/lib.rs @@ -92,8 +92,7 @@ impl GitSource { } // The stdout is bounded: a large repository cannot consume // unbounded controller memory. A truncated listing would hash only - // part of the file set, so truncation is an error, not a - // truncation. + // 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( From e79de9946f991552050dccec911d72f9c2579013 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fr=C3=B8yland?= Date: Sat, 19 Sep 2026 19:47:24 +0000 Subject: [PATCH 6/6] test: serialize the trust suite's sshd startup --- crates/providers/fleet-provider-ssh/tests/trust.rs | 7 +++++++ 1 file changed, 7 insertions(+) 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");