diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index 369a88d9439..f23c45a0dc5 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -18,6 +18,8 @@ use smallclaims::fleet::{FleetView, MemberView}; const REQUEST_LIMIT: usize = 1024 * 1024; pub(super) const RENAMED_OWNER_PREFIX: &str = concat!("fractal", "/fleet/"); +#[cfg(test)] +const RENAMED_PTY_PREFIX: &str = concat!("fractal", "/pty/"); const RESPONSE_LIMIT: u64 = 4 * 1024 * 1024; const CONTROL_TIMEOUT: Duration = Duration::from_secs(5); @@ -80,16 +82,19 @@ pub fn control_socket() -> Result { #[derive(Debug, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] enum Source { - Runtime { owner: String }, + Runtime { + owner: String, + expires_at: Option, + }, Static { path: PathBuf }, } impl Source { fn owned_by(&self, owner: &str) -> bool { - matches!(self, Self::Runtime { owner: source_owner } if source_owner == owner) + matches!(self, Self::Runtime { owner: source_owner, .. } if source_owner == owner) } fn describe(&self) -> String { match self { - Self::Runtime { owner } => format!("runtime owner {owner}"), + Self::Runtime { owner, .. } => format!("runtime owner {owner}"), Self::Static { path } => format!("static {}", path.display()), } } @@ -314,7 +319,9 @@ fn desired( type DesiredPolicy = BTreeMap>; -fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> Vec { +type Expiries = BTreeMap; + +fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy, expiries: &Expiries) -> Vec { let owned = snapshot .peers .iter() @@ -331,21 +338,38 @@ fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> Vec operations.push(json!({"op":"remove_peer", "node_id":node})); } for (node, grants) in desired { - if let Some(peer) = owned.get(node.as_str()) { - let actual = peer - .services - .iter() - .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner))) - .map(|(service, _)| service.clone()) - .collect::>(); - if actual == *grants { - continue; + let peer = owned.get(node.as_str()); + let actual_expiry = peer.and_then(|peer| peer.trust_sources.iter().find_map(|source| match source { + Source::Runtime { owner: source_owner, expires_at } if source_owner == owner => Some(*expires_at), + _ => None, + })).flatten(); + let expected_expiry = expiries.get(node).copied(); + if owner.starts_with(RENAMED_OWNER_PREFIX) { + // Recovery is a reader: expiry changes may revoke the old record, + // never promote it by removing and re-adding grants. + if actual_expiry != expected_expiry { + operations.push(json!({"op":"remove_peer", "node_id":node})); } - } else { - // No label: avoid conflicting with a static source's user-selected name. + continue; + } + // Fabric expiry is immutable. Promotion/abort replaces the record in + // this same atomic batch, never renews a staged record by add_peer alone. + let replace = actual_expiry.is_some() && actual_expiry != expected_expiry; + if replace { + operations.push(json!({"op":"remove_peer", "node_id":node})); + } + let actual = peer.map(|peer| peer.services.iter() + .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner))) + .map(|(service, _)| service.clone()).collect::>()); + if peer.is_none() || replace { operations.push(json!({"op":"add_peer", "node_id":node})); } - operations.push(json!({"op":"set_grants", "node_id":node, "grants":grants})); + if replace || actual.as_ref() != Some(grants) { + operations.push(json!({"op":"set_grants", "node_id":node, "grants":grants})); + } + if expected_expiry.is_some() && (replace || actual_expiry != expected_expiry) { + operations.push(json!({"op":"set_expiry", "node_id":node, "expires_at":expected_expiry})); + } } operations } @@ -433,7 +457,7 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> { .await? } -fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { +fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy, expiries: &Expiries) -> bool { if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with(RENAMED_OWNER_PREFIX)) { return pending.request["operations"].as_array().is_some_and(|operations| operations.iter().all(|operation| operation["op"] == "remove_peer")); @@ -447,7 +471,9 @@ fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { }; match operation["op"].as_str() { Some("add_peer") => desired.contains_key(node), - Some("remove_peer") => !desired.contains_key(node), + Some("remove_peer") => !desired.contains_key(node) + || pending.request["operations"].as_array().is_some_and(|ops| ops.iter().any(|op| op["op"] == "add_peer" && op["node_id"] == node)), + Some("set_expiry") => expiries.get(node).is_some_and(|deadline| operation["expires_at"].as_u64() == Some(*deadline)), Some("set_grants") => desired.get(node).is_some_and(|grants| { operation["grants"].as_array().is_some_and(|services| { services.len() == grants.len() @@ -511,11 +537,12 @@ async fn resolve_pending( socket: &Path, journal: &mut Journal, desired: &DesiredPolicy, + expiries: &Expiries, ) -> Result> { let Some(pending) = &journal.pending else { return Ok(None); }; - let replay_allowed = pending_matches_desired(pending, desired); + let replay_allowed = pending_matches_desired(pending, desired, expiries); match complete_pending(socket, pending, replay_allowed).await { Ok(mutation) => { journal.pending = None; @@ -563,6 +590,7 @@ async fn apply( socket: &Path, fleet_id: &str, desired: DesiredPolicy, + expiries: Expiries, ended: BTreeSet, withdrawing: bool, ) -> Result { @@ -576,13 +604,13 @@ async fn apply( "pending Fabric batch belongs to another fleet" ); } - resolve_pending(state_dir, socket, &mut journal, &desired).await?; + resolve_pending(state_dir, socket, &mut journal, &desired, &expiries).await?; let mut journal_changed = false; for node in ended { journal_changed |= journal.retired.insert(node); } let mut current = snapshot(socket).await?; - for owner in [owner, format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] { + for owner in [owner.clone(), format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] { let retained; let policy = if owner.starts_with("st/") { &desired @@ -590,7 +618,7 @@ async fn apply( retained = retained_policy(¤t, &owner, &desired); &retained }; - let operations = plan(¤t, &owner, policy); + let operations = plan(¤t, &owner, policy, &expiries); ensure!( operations.len() <= 128, "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", @@ -622,7 +650,7 @@ async fn apply( // Durable before sending: restarting st must not forget an uncertain commit // or replace a pending/timed-out cancellation with an empty-diff success. save_journal(state_dir, &journal).await?; - let mutation = resolve_pending(state_dir, socket, &mut journal, policy) + let mutation = resolve_pending(state_dir, socket, &mut journal, policy, &expiries) .await? .expect("pending transaction"); journal_changed = false; @@ -633,6 +661,10 @@ async fn apply( ); } } + ensure!( + plan(¤t, &owner, &desired, &expiries).is_empty(), + "Fabric final snapshot does not match the desired exact grants and expiry; no rekey receipt is safe" + ); let retired_count = journal.retired.len(); journal.retired.retain(|node| { !desired.contains_key(node) @@ -682,7 +714,7 @@ pub async fn reconcile( .members .iter() .any(|member| member.member_key == own_key && member.state == "current"); - let desired = desired + let desired: DesiredPolicy = desired .into_iter() .map(|node| { let peer_current = view.members.iter().any(|member| { @@ -699,6 +731,7 @@ pub async fn reconcile( (node, services) }) .collect(); + let expiries = Expiries::new(); let ended = view .members .iter() @@ -707,7 +740,7 @@ pub async fn reconcile( .filter(|endpoint| endpoint["transport"] == "fabric") .filter_map(|endpoint| endpoint["node"].as_str().map(str::to_owned)) .collect(); - let mut status = apply(state_dir, socket, fleet_id, desired, ended, false).await?; + let mut status = apply(state_dir, socket, fleet_id, desired, expiries, ended, false).await?; if status.state == "applied" && view.members.iter().any(|member| { member.state == "ended" @@ -729,6 +762,7 @@ pub async fn withdraw(state_dir: &Path, socket: &Path, fleet_id: &str) -> Result socket, fleet_id, DesiredPolicy::new(), + Expiries::new(), BTreeSet::new(), true, ) diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index 212197ac365..a6574fb84ca 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -38,7 +38,7 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { BTreeSet::from(["st3/fleet/test-fleet".into()]), )]); let retained = retained_policy(&snapshot, renamed, &desired); - let operations = plan(&snapshot, renamed, &retained); + let operations = plan(&snapshot, renamed, &retained, &Expiries::new()); assert_eq!( operations[0], json!({"op":"remove_peer","node_id":"old-node"}) @@ -47,9 +47,9 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { let mut pending = uncertain_add(); pending.request["owner"] = json!(renamed); pending.request["operations"][1]["grants"] = json!(["fractal/fleet/test-fleet"]); - assert!(!pending_matches_desired(&pending, &desired)); - pending.request["operations"][1]["grants"] = json!(["fractal/pty/test-fleet"]); - assert!(!pending_matches_desired(&pending, &desired)); + assert!(!pending_matches_desired(&pending, &desired, &Expiries::new())); + pending.request["operations"][1]["grants"] = json!([format!("{RENAMED_PTY_PREFIX}{FLEET}")]); + assert!(!pending_matches_desired(&pending, &desired, &Expiries::new())); } #[tokio::test] @@ -59,7 +59,7 @@ async fn renamed_pending_addition_is_observed_never_sent() { pending.request["owner"] = json!(format!("{RENAMED_OWNER_PREFIX}{FLEET}")); let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]); complete_pending(&socket, &pending, pending_matches_desired(&pending, - &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]))).await.unwrap(); + &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]), &Expiries::new())).await.unwrap(); assert_eq!(server.await.unwrap()[0]["request"]["operation"], "cancellation"); } @@ -214,7 +214,7 @@ async fn adds_peer_and_replaces_exact_grants_in_one_atomic_batch() { vec![ Some(list(vec![], 7)), Some(mutation(vec![], "complete")), - Some(list(vec![], 8)), + Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)), ], ); let status = reconcile( @@ -247,12 +247,16 @@ async fn adds_peer_and_replaces_exact_grants_in_one_atomic_batch() { #[tokio::test] async fn rekey_removes_old_identity_and_adds_new_in_same_transaction() { let root = tempfile::tempdir().unwrap(); + let services = ["st3-client-v0", "st3/fleet/test-fleet"]; + let mut membership = view(); + let member_key = membership.members[1].member_key.clone(); + membership.members[1].endpoints[0]["node"] = json!("new-node"); let (socket, server) = serve( root.path(), vec![ - Some(list(vec![peer("old-node", OWNER, &["exec", "echo"])], 7)), + Some(list(vec![peer("old-node", OWNER, &services)], 7)), Some(mutation(vec![], "complete")), - Some(list(vec![], 8)), + Some(list(vec![peer("new-node", OWNER, &services)], 8)), ], ); reconcile( @@ -260,19 +264,30 @@ async fn rekey_removes_old_identity_and_adds_new_in_same_transaction() { &socket, FLEET, "own-key", - &view(), + &membership, &BTreeSet::new(), ) .await .unwrap(); let requests = server.await.unwrap(); - let operations = requests[1]["request"]["operations"].as_array().unwrap(); + assert_eq!(membership.members[1].member_key, member_key); + // Only list -> one mutation -> list crosses the control boundary. There is + // no separate remove/add request that could expose an intermediate policy. + assert_eq!(requests.len(), 3); + assert_eq!(requests[0]["request"]["operation"], "list"); + assert_eq!(requests[2]["request"]["operation"], "list"); + let batch = &requests[1]["request"]; + assert_eq!(batch["operation"], "batch"); + assert_eq!(batch["owner"], OWNER); + assert_eq!(batch["expected_revision"], 7); assert_eq!( - operations[0], - json!({"op":"remove_peer", "node_id":"old-node"}) + batch["operations"], + json!([ + {"op":"remove_peer", "node_id":"old-node"}, + {"op":"add_peer", "node_id":"new-node"}, + {"op":"set_grants", "node_id":"new-node", "grants":services} + ]) ); - assert_eq!(operations[1]["node_id"], "peer-node"); - assert_eq!(operations.len(), 3); } #[tokio::test] @@ -289,7 +304,7 @@ async fn existing_owner_grants_are_replaced_without_touching_other_owners() { 7, )), Some(mutation(vec![], "complete")), - Some(list(vec![], 8)), + Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)), ], ); reconcile( @@ -352,7 +367,7 @@ async fn lost_reply_retries_identical_request_id_revision_and_contents() { Some(list(vec![], 7)), None, Some(mutation(vec![], "complete")), - Some(list(vec![], 8)), + Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)), ], ); reconcile( @@ -677,7 +692,7 @@ async fn real_fault_class_revision_conflict_allows_fresh_reconciliation() { Some(conflict), Some(list(vec![], 8)), Some(mutation(vec![], "complete")), - Some(list(vec![], 9)), + Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 9)), ], ); assert!( @@ -977,3 +992,15 @@ async fn leave_drain_never_replays_full_service_grants() { ]) ); } + +#[tokio::test] +async fn successful_mutation_without_verified_install_is_an_error() { + let root = tempfile::tempdir().unwrap(); + let (socket, server) = serve(root.path(), vec![ + Some(list(vec![], 7)), Some(mutation(vec![], "complete")), Some(list(vec![], 8)), + ]); + assert!(reconcile(root.path(), &socket, FLEET, "own-key", &view(), &BTreeSet::new()) + .await.unwrap_err().to_string().contains("final snapshot")); + server.await.unwrap(); +} + diff --git a/release-notes/fabric-rekey-runtime-expiry.json b/release-notes/fabric-rekey-runtime-expiry.json new file mode 100644 index 00000000000..c9d6b84d834 --- /dev/null +++ b/release-notes/fabric-rekey-runtime-expiry.json @@ -0,0 +1 @@ +{"version": 1,"summary": "Verify exact Fabric runtime grants and immutable expiry before accepting policy installation","replay": {"status": "none","detail": "No graph replay."},"schema": {"status": "none","detail": "Reuses fleet.member-endpoints claims; no database migration."},"rules": {"status": "none","detail": "No checkpoint rule change."},"compatibility": {"status": "changed","detail": "Non-routing rekey metadata is preserved during worker refresh and removed by explicit transport withdrawal."},"downtime": {"status": "none","detail": "No service restart required."},"manual": {"status": "none","detail": "No manual step."},"recovery": {"status": "changed","detail": "Renamed-owner recovery can retain or revoke existing access only; immutable expiry changes replace writer-owned records atomically."}}