diff --git a/crates/fractal-api-core/src/api.rs b/crates/fractal-api-core/src/api.rs index c5c96548a5a..4e0d80a1013 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/crates/fractal-api-core/src/api.rs @@ -843,6 +843,10 @@ pub fn router_for_transport(state: AppState, transport: ClientTransportBoundary, "/v1/internal/fleet/endpoints", post(fleet::fleet_publish_endpoints), ) + .route( + "/v1/internal/fleet/fabric-rekey", + post(fleet::fleet_fabric_rekey), + ) .route("/v1/internal/replication-wake", post(replication::replication_wake)) .route( crate::peer::CLIENT_READ_FORWARD_PATH, @@ -8608,6 +8612,7 @@ async fn replication_worker_status( } pub use crate::model::FleetEndpointsRequest; +pub use smallclaims::store::fabric_rekey::{FabricRekeyAction, FabricRekeyRequest, FabricRekeyState}; /// The peers a node reports on: its config peers and the current listening members it dials. /// A dial-out member is never dialed, and an ended name is history, so neither is reported. diff --git a/crates/fractal-api-core/src/api/fleet.rs b/crates/fractal-api-core/src/api/fleet.rs index f2984178b85..0b21de3315e 100644 --- a/crates/fractal-api-core/src/api/fleet.rs +++ b/crates/fractal-api-core/src/api/fleet.rs @@ -6,7 +6,13 @@ pub async fn fleet_publish_endpoints( ) -> Result, ApiError> { let store = state.store.clone(); let written = blocking_store(move || { - store.publish_fleet_endpoints(&request.mode, &request.endpoints, env!("CARGO_PKG_VERSION")) + if request.clear_fabric_rekey { + store.publish_fleet_endpoints_clearing_fabric_rekey( + &request.mode, &request.endpoints, env!("CARGO_PKG_VERSION"), + ) + } else { + store.publish_fleet_endpoints(&request.mode, &request.endpoints, env!("CARGO_PKG_VERSION")) + } }) .await?; if written { @@ -15,6 +21,20 @@ pub async fn fleet_publish_endpoints( Ok(Json(json!({ "published": written }))) } +pub async fn fleet_fabric_rekey( + State(state): State, + Json(request): Json, +) -> Result, ApiError> { + let store = state.store.clone(); + let written = blocking_store(move || { + store.fabric_rekey(&request, env!("CARGO_PKG_VERSION")) + }).await?; + if written { + signal_changed(&state); + } + Ok(Json(json!({ "published": written }))) +} + pub async fn fleet_status(State(state): State) -> Result, ApiError> { let peers = replication_peer_names(&state); let store = state.store.clone(); @@ -339,3 +359,58 @@ pub async fn fleet_admission_residue( .await .map(Json) } + +#[cfg(test)] +mod rekey_tests { + use super::*; + use crate::api::tests_shared::state; + use crate::api::tests_shared_client_v0_tests::json_request; + + #[tokio::test] + async fn signed_rekey_route_commits_preserves_refresh_and_rolls_back() { + let root = tempfile::tempdir().unwrap(); + let state = state(root.path()); + let (key, _) = smallclaims::fleet::MemberKey::generate().unwrap(); + let public = key.public().to_owned(); + state.store.bind_fleet("rekey-api-test").unwrap(); + state.store.pin_fleet_anchor(&public).unwrap(); + state.store.set_member_key(Some(Arc::new(key))).unwrap(); + state.store.admit_fleet_anchor("rekey-api-test", &public, "listening").unwrap(); + state.store.publish_fleet_endpoints("listening", &[json!({ + "transport":"fabric","node":"a".repeat(64),"protocol":"st3/fleet/rekey-api-test" + })], "test").unwrap(); + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() + 500; + let mut request = json!({"action":"prepare","old_node":"a".repeat(64), + "next_node":"b".repeat(64),"deadline":deadline}); + let path = "/v1/internal/fleet/fabric-rekey"; + let (status, body) = json_request(router(state.clone()), path, request.clone()).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["published"], true); + let (status, body) = json_request(router(state.clone()), path, request.clone()).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["published"], false); + request["action"] = json!("commit"); + let (status, body) = json_request(router(state.clone()), path, request.clone()).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, body) = json_request(router(state.clone()), "/v1/internal/fleet/endpoints", json!({ + "mode":"listening", "endpoints":[{"transport":"fabric","node":"a".repeat(64), + "protocol":"st3/fleet/rekey-api-test"}] + })).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let view = state.store.fleet_view_sealed().unwrap(); + assert!(view.members[0].endpoints.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == "b".repeat(64))); + request["action"] = json!("abort"); + let (status, body) = json_request(router(state.clone()), path, request.clone()).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let view = state.store.fleet_view_sealed().unwrap(); + assert!(view.members[0].endpoints.iter().any(|entry| entry["state"] == "aborted")); + let (status, body) = json_request(router(state.clone()), "/v1/internal/fleet/endpoints", json!({ + "mode":"listening", "endpoints":[], "clear_fabric_rekey":true + })).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!(state.store.fleet_view_sealed().unwrap().members[0].endpoints.is_empty()); + request["action"] = json!("commit"); + let (status, _) = json_request(router(state), path, request).await; + assert!(!status.is_success()); + } +} diff --git a/crates/fractal-fabric-adapter/src/main.rs b/crates/fractal-fabric-adapter/src/main.rs index a3e7439c84e..c579c3b300b 100644 --- a/crates/fractal-fabric-adapter/src/main.rs +++ b/crates/fractal-fabric-adapter/src/main.rs @@ -239,6 +239,9 @@ async fn poll( &retained(&snapshot, now_ms()), ) .await?; + if last.state == "applied" { + reconcile::acknowledge(client, socket, fleet, own, &snapshot.membership).await?; + } if last.state == "applied" && snapshot .members diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index e63761922e2..d4525458d11 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -79,16 +79,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()), } } @@ -271,7 +274,66 @@ fn desired( type DesiredPolicy = BTreeMap>; -fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> Vec { +#[derive(Debug, Deserialize, Serialize)] +struct Rotation { + old_node: String, + next_node: String, + deadline: u64, + state: String, +} + +type Expiries = BTreeMap; + +fn rotations( + view: &FleetView, + own_key: &str, + desired: &mut DesiredPolicy, + now: u64, +) -> Result<(Expiries, Vec<(String, Rotation)>)> { + let mut expiries = Expiries::new(); + let mut receipts = Vec::new(); + if !view.members.iter().any(|member| member.member_key == own_key && member.state == "current") { + return Ok((expiries, receipts)); + } + for member in view.members.iter().filter(|member| member.member_key != own_key && member.state == "current") { + let entries = member.endpoints.iter().filter(|entry| entry["transport"] == "fabric-rekey").collect::>(); + ensure!(entries.len() <= 1, "member has multiple Fabric rekey transitions"); + let Some(entry) = entries.first() else { continue }; + let rotation: Rotation = serde_json::from_value((*entry).clone()).context("invalid signed Fabric rekey transition")?; + ensure!(!rotation.old_node.is_empty() && !rotation.next_node.is_empty() && rotation.old_node != rotation.next_node, "invalid Fabric rekey identities"); + ensure!(rotation.deadline <= i64::MAX as u64 && rotation.deadline > 0, "invalid Fabric rekey deadline"); + ensure!(matches!(rotation.state.as_str(), "prepared" | "committed" | "aborted"), "invalid Fabric rekey state"); + let advertised = if rotation.state == "committed" { &rotation.next_node } else { &rotation.old_node }; + let services = desired.get(advertised).cloned().context("Fabric rekey does not match signed routing endpoint")?; + desired.remove(&rotation.old_node); + desired.remove(&rotation.next_node); + let live = rotation.deadline > now; + match rotation.state.as_str() { + "prepared" => { + desired.insert(rotation.old_node.clone(), services.clone()); + if live { + desired.insert(rotation.next_node.clone(), services); + expiries.insert(rotation.next_node.clone(), rotation.deadline); + } + } + "committed" => { + desired.insert(rotation.next_node.clone(), services.clone()); + if live { + desired.insert(rotation.old_node.clone(), services); + expiries.insert(rotation.old_node.clone(), rotation.deadline); + } + } + "aborted" => { desired.insert(rotation.old_node.clone(), services); } + _ => unreachable!(), + } + if live && rotation.state != "aborted" { + receipts.push((member.member_key.clone(), rotation)); + } + } + Ok((expiries, receipts)) +} + +fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy, expiries: &Expiries) -> Vec { let owned = snapshot .peers .iter() @@ -288,21 +350,30 @@ 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; - } - } else { - // No label: avoid conflicting with a static source's user-selected name. + 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(); + // 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 } @@ -390,7 +461,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 { pending.request["operations"] .as_array() .is_some_and(|operations| { @@ -400,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() @@ -464,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; @@ -516,19 +590,20 @@ async fn apply( socket: &Path, fleet_id: &str, desired: DesiredPolicy, + expiries: Expiries, ended: BTreeSet, withdrawing: bool, ) -> Result { let owner = format!("st/fleet/{fleet_id}"); let grants = grants(fleet_id); let mut journal = load_journal(state_dir).await?; - 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?; - let operations = plan(¤t, &owner, &desired); + let operations = plan(¤t, &owner, &desired, &expiries); ensure!( operations.len() <= 128, "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", @@ -560,7 +635,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, &desired) + let mutation = resolve_pending(state_dir, socket, &mut journal, &desired, &expiries) .await? .expect("pending transaction"); journal_changed = false; @@ -570,6 +645,10 @@ async fn apply( "Fabric policy revision regressed after mutation" ); } + 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) @@ -619,7 +698,7 @@ pub async fn reconcile( .members .iter() .any(|member| member.member_key == own_key && member.state == "current"); - let desired = desired + let mut desired: DesiredPolicy = desired .into_iter() .map(|node| { let peer_current = view.members.iter().any(|member| { @@ -636,6 +715,7 @@ pub async fn reconcile( (node, services) }) .collect(); + let (expiries, _) = rotations(view, own_key, &mut desired, (now_ms() / 1000) as u64)?; let ended = view .members .iter() @@ -644,7 +724,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" @@ -658,6 +738,35 @@ pub async fn reconcile( Ok(status) } +/// Receipts are authored under the adapter's own signed membership by the local +/// daemon, and bind every transition selector, including the installation phase. +pub async fn acknowledge( + client: &reqwest::Client, + socket: &Path, + fleet_id: &str, + own_key: &str, + view: &FleetView, +) -> Result<()> { + let nodes = desired(view, own_key, fleet_id, &BTreeSet::new())?; + let mut policy = nodes.into_iter().map(|node| (node, grants(fleet_id))).collect(); + let (expiries, receipts) = rotations(view, own_key, &mut policy, (now_ms() / 1000) as u64)?; + if receipts.is_empty() { return Ok(()); } + let current = snapshot(socket).await?; + ensure!(plan(¤t, &format!("st/fleet/{fleet_id}"), &policy, &expiries).is_empty(), + "Fabric receipt verification failed: exact installed grants/expiry differ"); + for (member_key, rotation) in receipts { + ensure!(u128::from(rotation.deadline) * 1000 > now_ms(), "Fabric rekey expired before receipt publication"); + client.post("http://localhost/v1/internal/fleet/fabric-rekey") + .json(&json!({ + "action":"receipt", "member_key":member_key, + "old_node":rotation.old_node, "next_node":rotation.next_node, + "deadline":rotation.deadline, "state":rotation.state, + })) + .send().await?.error_for_status()?; + } + Ok(()) +} + /// Revoke this fleet's runtime records after it disappears from local fleet settings. /// Call with the same adapter state-directory lock used by reconciliation. pub async fn withdraw(state_dir: &Path, socket: &Path, fleet_id: &str) -> Result { @@ -666,6 +775,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 4154801e861..4cd7b4e2266 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -135,7 +135,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( @@ -168,12 +168,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( @@ -181,19 +185,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] @@ -210,7 +225,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( @@ -273,7 +288,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( @@ -598,7 +613,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!( @@ -898,3 +913,186 @@ async fn leave_drain_never_replays_full_service_grants() { ]) ); } + +fn rotation_view(state: &str, deadline: u64) -> FleetView { + let mut view = view(); + view.members[1].endpoints[0]["node"] = json!(if state == "committed" { "next-node" } else { "peer-node" }); + view.members[1].endpoints.push(json!({ + "transport":"fabric-rekey", "old_node":"peer-node", "next_node":"next-node", + "deadline":deadline, "state":state, + })); + view +} + +fn expiring_peer(node: &str, deadline: u64) -> Value { + let mut value = peer(node, OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]); + value["trust_sources"][0]["expires_at"] = json!(deadline); + for sources in value["services"].as_object_mut().unwrap().values_mut() { + sources[0]["expires_at"] = json!(deadline); + } + value +} + +#[tokio::test] +async fn prepared_committed_and_aborted_reconcile_exact_immutable_expiry() { + let deadline = (now_ms() / 1000) as u64 + 300; + let permanent = |node| peer(node, OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]); + let staged = vec![permanent("peer-node"), expiring_peer("next-node", deadline)]; + let committed = vec![expiring_peer("peer-node", deadline), permanent("next-node")]; + for (state, before, after, expected) in [ + ("prepared", vec![permanent("peer-node")], staged.clone(), json!([ + {"op":"add_peer","node_id":"next-node"}, + {"op":"set_grants","node_id":"next-node","grants":["st3-client-v0","st3/fleet/test-fleet"]}, + {"op":"set_expiry","node_id":"next-node","expires_at":deadline} + ])), + ("committed", staged, committed.clone(), json!([ + {"op":"remove_peer","node_id":"next-node"}, + {"op":"add_peer","node_id":"next-node"}, + {"op":"set_grants","node_id":"next-node","grants":["st3-client-v0","st3/fleet/test-fleet"]}, + {"op":"set_expiry","node_id":"peer-node","expires_at":deadline} + ])), + ("aborted", committed, vec![permanent("peer-node")], json!([ + {"op":"remove_peer","node_id":"next-node"}, + {"op":"remove_peer","node_id":"peer-node"}, + {"op":"add_peer","node_id":"peer-node"}, + {"op":"set_grants","node_id":"peer-node","grants":["st3-client-v0","st3/fleet/test-fleet"]} + ])), + ] { + let root = tempfile::tempdir().unwrap(); + let (socket, server) = serve(root.path(), vec![ + Some(list(before, 7)), Some(mutation(vec![], "complete")), Some(list(after, 8)), + ]); + let status = reconcile(root.path(), &socket, FLEET, "own-key", &rotation_view(state, deadline), &BTreeSet::new()).await.unwrap(); + assert_eq!(status.state, "applied"); + assert_eq!(server.await.unwrap()[1]["request"]["operations"], expected); + } +} + +#[test] +fn expired_transitions_never_renew_staging_and_never_expire_prepared_old() { + for state in ["prepared", "committed", "aborted"] { + let view = rotation_view(state, 1000); + let mut policy = desired(&view, "own-key", FLEET, &BTreeSet::new()).unwrap() + .into_iter().map(|node| (node, grants(FLEET))).collect(); + let (expiries, receipts) = rotations(&view, "own-key", &mut policy, 1000).unwrap(); + assert!(expiries.is_empty()); + assert!(receipts.is_empty()); + assert_eq!(policy.keys().cloned().collect::>(), + vec![if state == "committed" { "next-node" } else { "peer-node" }.to_owned()]); + } +} + +#[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(); +} + +#[tokio::test] +async fn receipts_bind_exact_phase_and_require_installed_grants_even_on_noop() { + for state in ["prepared", "committed"] { + let deadline = (now_ms() / 1000) as u64 + 300; + let root = tempfile::tempdir().unwrap(); + let http_socket = root.path().join("st.sock"); + let listener = UnixListener::bind(&http_socket).unwrap(); + let http = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut reader = BufReader::new(&mut stream); + let mut content_length = 0; + loop { + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + if line == "\r\n" { break; } + if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") { + content_length = value.trim().parse::().unwrap(); + } + } + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).await.unwrap(); + drop(reader); + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}").await.unwrap(); + serde_json::from_slice::(&body).unwrap() + }); + let client = reqwest::Client::builder().unix_socket(http_socket).build().unwrap(); + let permanent = |node| peer(node, OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]); + let installed = if state == "prepared" { + vec![permanent("peer-node"), expiring_peer("next-node", deadline)] + } else { + vec![expiring_peer("peer-node", deadline), permanent("next-node")] + }; + let (socket, server) = serve(root.path(), vec![Some(list(installed, 7))]); + acknowledge(&client, &socket, FLEET, "own-key", &rotation_view(state, deadline)).await.unwrap(); + assert_eq!(http.await.unwrap(), json!({ + "action":"receipt", "member_key":"peer-key", "old_node":"peer-node", + "next_node":"next-node", "deadline":deadline, "state":state, + })); + assert_eq!(server.await.unwrap().len(), 1); + + let failed = tempfile::tempdir().unwrap(); + let (socket, server) = serve(failed.path(), vec![Some(list(vec![permanent("peer-node")], 7))]); + assert!(acknowledge(&client, &socket, FLEET, "own-key", &rotation_view(state, deadline)).await.is_err()); + assert_eq!(server.await.unwrap().len(), 1); + } +} + +#[tokio::test] +async fn elapsed_preparation_removes_next_without_renewal_or_old_expiry() { + let root = tempfile::tempdir().unwrap(); + let deadline = (now_ms() / 1000) as u64 - 1; + let old = peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]); + let (socket, server) = serve(root.path(), vec![ + Some(list(vec![old.clone(), expiring_peer("next-node", deadline)], 7)), + Some(mutation(vec![], "complete")), + Some(list(vec![old], 8)), + ]); + reconcile(root.path(), &socket, FLEET, "own-key", &rotation_view("prepared", deadline), &BTreeSet::new()) + .await.unwrap(); + assert_eq!(server.await.unwrap()[1]["request"]["operations"], + json!([{"op":"remove_peer","node_id":"next-node"}])); +} + +#[test] +fn phase_changes_and_elapsed_preparation_cannot_replay_staging_transaction() { + let deadline = 1300; + let pending = Pending { + created_unix_ms: now_ms(), + request: json!({"operations":[ + {"op":"add_peer","node_id":"next-node"}, + {"op":"set_grants","node_id":"next-node","grants":["st3-client-v0","st3/fleet/test-fleet"]}, + {"op":"set_expiry","node_id":"next-node","expires_at":deadline} + ]}), + }; + for (state, now, expected) in [ + ("prepared", 1000, true), + ("prepared", 1300, false), + ("committed", 1000, false), + ("aborted", 1000, false), + ] { + let view = rotation_view(state, deadline); + let mut policy = desired(&view, "own-key", FLEET, &BTreeSet::new()).unwrap() + .into_iter().map(|node| (node, grants(FLEET))).collect(); + let (expiries, _) = rotations(&view, "own-key", &mut policy, now).unwrap(); + assert_eq!(pending_matches_desired(&pending, &policy, &expiries), expected); + } +} + +#[test] +fn own_metadata_and_noncurrent_members_never_authorize_overlap_or_receipts() { + let mut view = rotation_view("prepared", 1300); + let rotation = view.members[1].endpoints.pop().unwrap(); + view.members[0].endpoints.push(rotation); + let mut policy = DesiredPolicy::from([("peer-node".into(), grants(FLEET))]); + let (expiries, receipts) = rotations(&view, "own-key", &mut policy, 1000).unwrap(); + assert!(expiries.is_empty()); + assert!(receipts.is_empty()); + let mut view = rotation_view("prepared", 1300); + view.members[1].state = "ended".into(); + let (expiries, receipts) = rotations(&view, "own-key", &mut policy, 1000).unwrap(); + assert!(expiries.is_empty()); + assert!(receipts.is_empty()); +} diff --git a/crates/fractal-kernel/src/model.rs b/crates/fractal-kernel/src/model.rs index 8715ea5f3a8..c62671a91bf 100644 --- a/crates/fractal-kernel/src/model.rs +++ b/crates/fractal-kernel/src/model.rs @@ -93,6 +93,9 @@ pub const FIRST_READINESS_FAULT_SCOPE: &str = "scheduler/first-readiness"; pub struct FleetEndpointsRequest { pub mode: String, pub endpoints: Vec, + /// Explicit settings cutover; ordinary worker refreshes preserve signed rekey state. + #[serde(default)] + pub clear_fabric_rekey: bool, } /// The error code of a send or reply addressed to a person. diff --git a/crates/fractal-peer/src/peer.rs b/crates/fractal-peer/src/peer.rs index 0ff6681f5dc..fb887693b71 100644 --- a/crates/fractal-peer/src/peer.rs +++ b/crates/fractal-peer/src/peer.rs @@ -1900,6 +1900,7 @@ impl Backend for MainBackend { &crate::model::FleetEndpointsRequest { mode: mode.into(), endpoints: endpoints.to_vec(), + clear_fabric_rekey: false, }, ) .await?; diff --git a/crates/fractal/src/fleet_cli.rs b/crates/fractal/src/fleet_cli.rs index 5f2983233dd..e4541ea67aa 100644 --- a/crates/fractal/src/fleet_cli.rs +++ b/crates/fractal/src/fleet_cli.rs @@ -1,6 +1,7 @@ use super::*; // The same print macros as main.rs: CLI output goes through st3's writer. use fractal_daemon::{cli_print as print, cli_println as println}; +mod rekey; // Some fleet commands historically print prose even with --json. Keep those bytes // stable for callers; only the human rendering uses the installed fractal command. @@ -534,6 +535,7 @@ pub(super) async fn run_fleet(endpoint: &Endpoint, command: FleetCommand, json_o FleetTransportsCommand::Add { no_service, .. } | FleetTransportsCommand::Remove { no_service, .. } | FleetTransportsCommand::Prefer { no_service, .. } => !*no_service, + FleetTransportsCommand::Rekey { .. } => false, }, _ => false, }; @@ -1107,6 +1109,8 @@ async fn save_and_publish_fleet_settings( &fractal_daemon::api::FleetEndpointsRequest { mode: file.mode.as_str().into(), endpoints, + clear_fabric_rekey: file.mode == fractal_daemon::config::FleetMode::DialOut + || !file.transports.iter().any(|transport| transport == "fabric"), }, ) .await @@ -1130,6 +1134,9 @@ async fn run_fleet_transports( json_output: bool, ) -> Result<()> { use fractal_daemon::fleet::transport::{Fabric, default_fabric_protocol, resolve_tool}; + if matches!(command, FleetTransportsCommand::Rekey { .. }) { + return rekey::run(client, config, json_output).await; + } let lock = lock_fleet_settings(&config.state_dir).await?; let mut file = fractal_daemon::config::FleetFile::load(&config.state_dir)? .context("this machine is not a fleet member")?; @@ -1140,7 +1147,8 @@ async fn run_fleet_transports( let transport = match &command { FleetTransportsCommand::Add { transport, .. } | FleetTransportsCommand::Remove { transport, .. } - | FleetTransportsCommand::Prefer { transport, .. } => transport, + | FleetTransportsCommand::Prefer { transport, .. } + | FleetTransportsCommand::Rekey { transport } => transport, }; let enabled = file .transports @@ -1188,6 +1196,7 @@ async fn run_fleet_transports( .clone() .unwrap_or_else(|| default_fabric_protocol(&file.fleet_id)); let no_service = match command { + FleetTransportsCommand::Rekey { .. } => unreachable!("rekey dispatched before settings edits"), FleetTransportsCommand::Prefer { transport, no_service, diff --git a/crates/fractal/src/fleet_cli/rekey.rs b/crates/fractal/src/fleet_cli/rekey.rs new file mode 100644 index 00000000000..2e416bbeb35 --- /dev/null +++ b/crates/fractal/src/fleet_cli/rekey.rs @@ -0,0 +1,368 @@ +//! Prepare and publish through the old route before guarded Fabric activation. +use super::*; +use fractal_daemon::cli_println as println; +use serde::{Deserialize, Serialize}; +use std::path::Path; + +const OVERLAP_SECONDS: u64 = 600; +const COMMAND_TIMEOUT: Duration = Duration::from_secs(90); + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum Phase { + Prepared, + Committed, + Activating, + Complete, + Aborted, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Progress { + fleet_id: String, + member_key: String, + old_node: String, + next_node: String, + deadline: u64, + phase: Phase, +} + +fn now() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +fn save(path: &Path, progress: &Progress) -> Result<()> { + use std::{io::Write, os::unix::fs::OpenOptionsExt}; + let parent = path.parent().context("rekey journal has no parent")?; + let temporary = path.with_extension("json.tmp"); + let mut file = std::fs::OpenOptions::new() + .write(true).create(true).truncate(true).mode(0o600).open(&temporary)?; + file.write_all(&serde_json::to_vec(progress)?)?; + file.sync_all()?; + std::fs::rename(temporary, path)?; + std::fs::File::open(parent)?.sync_all()?; + Ok(()) +} + +fn load(path: &Path) -> Result> { + match std::fs::read(path) { + Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes).context("read Fabric rekey journal")?)), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error.into()), + } +} + +async fn fabric_json(program: &Path, args: &[&str]) -> Result { + // kill_on_drop bounds a hung subprocess without touching the supervised daemon. + let output = tokio::time::timeout(COMMAND_TIMEOUT, + tokio::process::Command::new(program).args(args).kill_on_drop(true).output()) + .await.context("Fabric identity command timed out; retry the same rekey command")??; + anyhow::ensure!(output.status.success(), "Fabric identity command failed: {}{}", + String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + serde_json::from_slice(&output.stdout).context("Fabric identity command returned malformed JSON") +} + +fn identity(value: &Value, field: &str) -> Result { + let node = value[field].as_str().context("Fabric identity response has no NodeID")?; + anyhow::ensure!(node.len() == 64 && node.bytes().all(|byte| byte.is_ascii_hexdigit()), + "Fabric identity response has an invalid NodeID"); + Ok(node.to_owned()) +} + +async fn publish(client: &Client, progress: &Progress, action: &str) -> Result<()> { + let _: Value = client.post("/v1/internal/fleet/fabric-rekey", &json!({ + "action":action, "old_node":progress.old_node, "next_node":progress.next_node, + "deadline":progress.deadline, + })).await?; + Ok(()) +} + +fn confirmed(view: &fractal_daemon::fleet::FleetView, progress: &Progress, state: &str) -> bool { + view.members.iter().any(|member| { + member.state == "current" && member.member_key != progress.member_key + && member.endpoints.iter().any(|endpoint| { + endpoint["transport"] == "fabric-rekey-receipt" + && endpoint["member_key"] == progress.member_key + && endpoint["old_node"] == progress.old_node + && endpoint["next_node"] == progress.next_node + && endpoint["deadline"] == progress.deadline + && endpoint["state"] == state + }) + }) +} + +async fn wait_for_receipt(client: &Client, progress: &Progress, state: &str) -> Result { + loop { + let remaining = progress.deadline.saturating_sub(now()); + if remaining == 0 { return Ok(false); } + // A slow local API must not extend the authorization window. + let view = tokio::time::timeout(Duration::from_secs(remaining), + client.get::("/v1/internal/fleet/membership")).await; + match view { + Ok(Ok(view)) if confirmed(&view, progress, state) && now() < progress.deadline => return Ok(true), + Ok(Ok(_)) => {}, + Ok(Err(error)) => return Err(error.context("read signed adapter receipts; retry rekey to resume")), + Err(_) => return Ok(false), + } + tokio::time::sleep(Duration::from_secs(1).min(Duration::from_secs(progress.deadline.saturating_sub(now())))).await; + } +} + +async fn abort(client: &Client, path: &Path, progress: &mut Progress) -> Result<()> { + // Publish before marking terminal: a lost reply is retried, never reported as a rollback. + publish(client, progress, "abort").await.context("could not publish rekey rollback; old Fabric identity is unchanged; retry rekey")?; + progress.phase = Phase::Aborted; + save(path, progress)?; + anyhow::bail!("no peer confirmed Fabric rekey before the ten-minute deadline; aborted, old identity remains active") +} + +pub(super) async fn run(client: &Client, config: &Config, json_output: bool) -> Result<()> { + use fractal_daemon::fleet::transport::resolve_tool; + let _lock = lock_fleet_settings(&config.state_dir).await?; + let file = fractal_daemon::config::FleetFile::load(&config.state_dir)? + .context("this machine is not a fleet member")?; + anyhow::ensure!(file.transports.iter().any(|transport| transport == "fabric"), + "this machine is not a Fabric fleet member"); + let view: fractal_daemon::fleet::FleetView = client.get("/v1/internal/fleet/membership").await?; + let member = view.members.iter().find(|member| member.name == config.node && member.state == "current") + .context("this machine is not a current fleet member")?; + anyhow::ensure!(member.endpoints.iter().any(|endpoint| endpoint["transport"] == "fabric"), + "this member has no published Fabric endpoint"); + let program = resolve_tool(file.fabric.as_deref(), "fabric").context("fabric is not installed")?; + let status = fabric_json(&program, &["identity", "status", "--json"]).await + .context("Fabric is too old or unavailable: install the Fabric v0.9.4 identity lifecycle API")?; + let current = identity(&status, "current")?; + let path = config.state_dir.join("fleet/fabric-rekey.json"); + let saved = load(&path)?; + let mut progress = match saved { + Some(progress) if !matches!(progress.phase, Phase::Complete | Phase::Aborted) => { + anyhow::ensure!(progress.fleet_id == file.fleet_id && progress.member_key == member.member_key, + "Fabric rekey journal belongs to another fleet member"); + anyhow::ensure!(current == progress.old_node || current == progress.next_node, + "Fabric identity changed outside this rekey; refusing to overwrite durable progress"); + progress + }, + _ => { + anyhow::ensure!(member.endpoints.iter().any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == current), + "published Fabric identity does not match the local identity"); + let prepared = fabric_json(&program, &["identity", "prepare", "--json"]).await?; + let old_node = identity(&prepared, "current")?; + let next_node = identity(&prepared, "next")?; + anyhow::ensure!(old_node == current && old_node != next_node, "Fabric prepared an unexpected identity"); + let progress = Progress { fleet_id:file.fleet_id.clone(), member_key:member.member_key.clone(), + old_node, next_node, deadline:now() + OVERLAP_SECONDS, phase:Phase::Prepared }; + save(&path, &progress)?; + progress + }, + }; + if current == progress.next_node { + anyhow::ensure!(matches!(progress.phase, Phase::Activating), "new identity active without durable activation decision"); + } else if progress.phase == Phase::Prepared { + if now() >= progress.deadline { return abort(client, &path, &mut progress).await; } + publish(client, &progress, "prepare").await?; + if !wait_for_receipt(client, &progress, "prepared").await? { + return abort(client, &path, &mut progress).await; + } + // Save intent before publication; retrying commit after a lost reply is safe. + progress.phase = Phase::Committed; + save(&path, &progress)?; + } + if progress.phase == Phase::Committed { + if now() >= progress.deadline { return abort(client, &path, &mut progress).await; } + publish(client, &progress, "commit").await?; + if !wait_for_receipt(client, &progress, "committed").await? { + return abort(client, &path, &mut progress).await; + } + // From here the next identity is permanently authorized by a peer. Never + // roll back on an uncertain activate reply: Fabric's guards reconcile it. + progress.phase = Phase::Activating; + save(&path, &progress)?; + } + let activated = fabric_json(&program, &["identity", "activate", "--expect-current", &progress.old_node, + "--expect-next", &progress.next_node, "--json"]).await + .context("Fabric activation incomplete; rerun fleet transports rekey fabric to resume the guarded activation")?; + anyhow::ensure!(identity(&activated, "active")? == progress.next_node, "Fabric activated an unexpected identity"); + progress.phase = Phase::Complete; + save(&path, &progress)?; + if json_output { + println!("{}", json!({"state":"complete","previous":progress.old_node,"active":progress.next_node,"deadline":progress.deadline})); + } else { + println!("Fabric re-keyed in place: {} -> {}. Offline peers recover when signed membership converges.", progress.old_node, progress.next_node); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{Json, Router, routing::{get, post}}; + use std::{os::unix::fs::PermissionsExt, sync::{Arc, Mutex}}; + + const OLD: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + const NEXT: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + const FLEET: &str = "68d50a41-99c6-443a-8791-b09fa2b8e468"; + + fn progress(phase: Phase, deadline: u64) -> Progress { + Progress { fleet_id:FLEET.into(), member_key:"own-key".into(), + old_node:OLD.into(), next_node:NEXT.into(), deadline, phase } + } + + fn membership(rotation: Option<&Value>) -> Value { + let node = if rotation.is_some_and(|entry| entry["action"] == "commit") { NEXT } else { OLD }; + let receipts = rotation.filter(|entry| entry["action"] != "abort").map(|entry| json!([{ + "transport":"fabric-rekey-receipt", "member_key":"own-key", "old_node":OLD, + "next_node":NEXT, "deadline":entry["deadline"], + "state":if entry["action"] == "commit" {"committed"} else {"prepared"}, + }])).unwrap_or_else(|| json!([])); + json!({"anchor":"anchor","members":[ + {"name":"local","member_key":"own-key","state":"current","mode":"listening","start":1, + "endpoints":[{"transport":"fabric","node":node,"protocol":"st3/fleet/fleet"}]}, + {"name":"remote","member_key":"remote-key","state":"current","mode":"listening","start":1,"endpoints":receipts} + ]}) + } + + struct Fixture { + root: tempfile::TempDir, + config: Config, + client: Client, + requests: Arc>>, + server: tokio::task::JoinHandle<()>, + } + impl Drop for Fixture { + fn drop(&mut self) { self.server.abort(); } + } + impl Fixture { + async fn new(fail_activation: bool) -> Self { + let root = tempfile::tempdir().unwrap(); + let program = root.path().join("fabric"); + // A real subprocess boundary with durable active identity and a lost + // activation reply; no process-global environment changes. + std::fs::write(&program, format!(r#"#!/bin/sh +echo "$*" >> '{dir}/calls' +case "$2" in +status) + if test -f '{dir}/active'; then id='{next}'; else id='{old}'; fi + printf '{{"current":"%s","state":"active"}}\n' "$id";; +prepare) printf '{{"current":"{old}","next":"{next}"}}\n';; +activate) + if test '{fail}' = true && ! test -f '{dir}/active'; then + touch '{dir}/active' + exit 1 + fi + touch '{dir}/active' + printf '{{"previous":"{old}","active":"{next}","outcome":"already_active"}}\n';; +*) exit 2;; +esac +"#, dir=root.path().display(), old=OLD, next=NEXT, fail=fail_activation)).unwrap(); + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o700)).unwrap(); + let config = Config { node:"local".into(), state_dir:root.path().to_path_buf(), ..Config::default() }; + fractal_daemon::config::FleetFile { + fleet_id:FLEET.into(), node:Some("local".into()), + transports:vec!["fabric".into()], fabric:Some(program), ..Default::default() + }.save(root.path()).unwrap(); + let requests = Arc::new(Mutex::new(Vec::::new())); + let reads = requests.clone(); + let writes = requests.clone(); + let app = Router::new() + .route("/v1/internal/fleet/membership", get(move || { + let reads = reads.clone(); + async move { Json(json!({"api_version":"st3.v1","value":membership(reads.lock().unwrap().last())})) } + })) + .route("/v1/internal/fleet/fabric-rekey", post(move |Json(request):Json| { + let writes = writes.clone(); + async move { writes.lock().unwrap().push(request); Json(json!({"api_version":"st3.v1","value":{"published":true}})) } + })); + let socket = root.path().join("api.sock"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); + let client = Client::unix(socket); + Self { root, config, client, requests, server } + } + } + + #[tokio::test] + async fn rekey_flow_prepares_publishes_both_phases_then_activates() { + let fixture = Fixture::new(false).await; + run(&fixture.client, &fixture.config, true).await.unwrap(); + let requests = fixture.requests.lock().unwrap(); + assert_eq!(requests.iter().map(|request| request["action"].as_str().unwrap()).collect::>(), ["prepare","commit"]); + assert_eq!(requests[0]["deadline"], requests[1]["deadline"]); + let saved = load(&fixture.root.path().join("fleet/fabric-rekey.json")).unwrap().unwrap(); + assert_eq!(saved.phase, Phase::Complete); + let calls = std::fs::read_to_string(fixture.root.path().join("calls")).unwrap(); + assert!(calls.contains("identity prepare --json")); + assert!(calls.contains(&format!("identity activate --expect-current {OLD} --expect-next {NEXT} --json"))); + } + + #[tokio::test] + async fn rekey_lost_activation_reply_resumes_without_preparing_or_extending_deadline() { + let fixture = Fixture::new(true).await; + assert!(run(&fixture.client, &fixture.config, true).await.is_err()); + let path = fixture.root.path().join("fleet/fabric-rekey.json"); + let saved = load(&path).unwrap().unwrap(); + assert_eq!(saved.phase, Phase::Activating); + run(&fixture.client, &fixture.config, true).await.unwrap(); + let resumed = load(&path).unwrap().unwrap(); + assert_eq!(resumed.phase, Phase::Complete); + assert_eq!(resumed.deadline, saved.deadline); + assert_eq!(fixture.requests.lock().unwrap().len(), 2); + let calls = std::fs::read_to_string(fixture.root.path().join("calls")).unwrap(); + assert_eq!(calls.matches("identity prepare").count(), 1); + assert_eq!(calls.matches("identity activate").count(), 2); + } + + #[tokio::test] + async fn rekey_elapsed_unpublished_journal_aborts_without_activation() { + let fixture = Fixture::new(false).await; + let path = fixture.root.path().join("fleet/fabric-rekey.json"); + save(&path, &progress(Phase::Prepared, now() - 1)).unwrap(); + let error = run(&fixture.client, &fixture.config, true).await.unwrap_err(); + assert!(error.to_string().contains("old identity remains active")); + assert_eq!(load(&path).unwrap().unwrap().phase, Phase::Aborted); + assert_eq!(fixture.requests.lock().unwrap()[0]["action"], "abort"); + assert!(!std::fs::read_to_string(fixture.root.path().join("calls")).unwrap().contains("activate")); + } + + #[tokio::test] + async fn rekey_refuses_old_fabric_before_preparing() { + let fixture = Fixture::new(false).await; + let program = fixture.root.path().join("fabric"); + std::fs::write(&program, "#!/bin/sh\nexit 2\n").unwrap(); + let error = run(&fixture.client, &fixture.config, true).await.unwrap_err(); + assert!(error.to_string().contains("Fabric is too old")); + assert!(fixture.requests.lock().unwrap().is_empty()); + assert!(!fixture.root.path().join("fleet/fabric-rekey.json").exists()); + } + + #[tokio::test] + async fn rekey_refuses_non_fabric_member_without_running_fabric() { + let fixture = Fixture::new(false).await; + let mut file = fractal_daemon::config::FleetFile::load(fixture.root.path()).unwrap().unwrap(); + file.transports = vec!["tailscale".into()]; + file.save(fixture.root.path()).unwrap(); + let error = run(&fixture.client, &fixture.config, true).await.unwrap_err(); + assert!(error.to_string().contains("not a Fabric fleet member")); + assert!(!fixture.root.path().join("calls").exists()); + assert!(fixture.requests.lock().unwrap().is_empty()); + } + + #[test] + fn rekey_receipts_require_remote_current_exact_phase_and_identity() { + let progress = progress(Phase::Prepared, now() + 100); + let request = json!({"action":"prepare","deadline":progress.deadline}); + let mut view:fractal_daemon::fleet::FleetView = serde_json::from_value(membership(Some(&request))).unwrap(); + assert!(confirmed(&view, &progress, "prepared")); + assert!(!confirmed(&view, &progress, "committed")); + view.members[1].endpoints[0]["deadline"] = json!(progress.deadline + 1); + assert!(!confirmed(&view, &progress, "prepared")); + view.members[1].endpoints[0]["deadline"] = json!(progress.deadline); + view.members[1].state = "ended".into(); + assert!(!confirmed(&view, &progress, "prepared")); + view.members[1].state = "current".into(); + view.members[1].member_key = progress.member_key.clone(); + assert!(!confirmed(&view, &progress, "prepared")); + } +} diff --git a/crates/fractal/src/main.rs b/crates/fractal/src/main.rs index f3e3c15ca30..1ced8bbf14e 100644 --- a/crates/fractal/src/main.rs +++ b/crates/fractal/src/main.rs @@ -1057,6 +1057,11 @@ struct FleetModeArgs { #[derive(Subcommand)] enum FleetTransportsCommand { + /// Re-key this member in place after peers install its prepared Fabric identity. + Rekey { + #[arg(value_parser = ["fabric"])] + transport: String, + }, /// Expose Fabric, then announce it. Requires a listening member with a port. Add { #[arg(value_parser = ["fabric"])] diff --git a/crates/smallclaims/src/store.rs b/crates/smallclaims/src/store.rs index b53f1d7eef8..9979e8de9b7 100644 --- a/crates/smallclaims/src/store.rs +++ b/crates/smallclaims/src/store.rs @@ -70,6 +70,7 @@ mod schema_install_tests; #[cfg(any(test, feature = "test-support"))] pub mod contents; pub mod prepared_seal; +pub mod fabric_rekey; #[cfg(test)] mod prepared_seal_tests; mod binary_payloads; @@ -837,6 +838,8 @@ pub struct Store { /// Held while one thread builds the next replication snapshot, so concurrent callers reuse /// it instead of building their own. pub replication_snapshot_build: Mutex<()>, + /// Serializes endpoint refreshes and signed Fabric rekey transitions. + fleet_endpoints_publication: Mutex<()>, pub replication_sync: Mutex>, /// Whether exchanges compare projections with a peer at all, keeping what a reproduction /// needs when they differ. Off unless the daemon's divergence check is on: a default @@ -1253,6 +1256,7 @@ impl Store { replica_generation: AtomicU64::new(0), replication_snapshot: Mutex::new(None), replication_snapshot_build: Mutex::new(()), + fleet_endpoints_publication: Mutex::new(()), replication_sync: Mutex::new(BTreeMap::new()), divergence_capture: std::sync::atomic::AtomicBool::new(false), trim_local: AtomicBool::new(false), @@ -2502,6 +2506,30 @@ impl Store { mode: &str, endpoints: &[Value], build: &str, + ) -> Result { + let _publication = self.fleet_endpoints_publication.lock().unwrap_or_else(PoisonError::into_inner); + let endpoints = self.preserve_fabric_rekey(mode, endpoints, mode != "listening")?; + self.publish_fleet_endpoints_locked(mode, &endpoints, build) + } + + /// An explicit settings withdrawal may clear completed rekey state, but cannot + /// discard a live prepared/committed transition. Remote receipts remain signed. + pub fn publish_fleet_endpoints_clearing_fabric_rekey( + &self, + mode: &str, + endpoints: &[Value], + build: &str, + ) -> Result { + let _publication = self.fleet_endpoints_publication.lock().unwrap_or_else(PoisonError::into_inner); + let endpoints = self.preserve_fabric_rekey(mode, endpoints, true)?; + self.publish_fleet_endpoints_locked(mode, &endpoints, build) + } + + fn publish_fleet_endpoints_locked( + &self, + mode: &str, + endpoints: &[Value], + build: &str, ) -> Result { let Some(member_key) = self.member_public_key() else { return Ok(false); diff --git a/crates/smallclaims/src/store/fabric_rekey.rs b/crates/smallclaims/src/store/fabric_rekey.rs new file mode 100644 index 00000000000..83299a56f27 --- /dev/null +++ b/crates/smallclaims/src/store/fabric_rekey.rs @@ -0,0 +1,392 @@ +//! Signed, non-routing rekey metadata shares the member endpoint announcement. +//! The publication mutex prevents ordinary worker refreshes from erasing a transition. +use super::*; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct FabricRekeyRequest { + pub action: FabricRekeyAction, + pub old_node: String, + pub next_node: String, + pub deadline: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub member_key: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum FabricRekeyAction { Prepare, Commit, Abort, Receipt } + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum FabricRekeyState { Prepared, Committed } + +fn metadata(endpoint: &Value) -> bool { + matches!(endpoint["transport"].as_str(), Some("fabric-rekey" | "fabric-rekey-receipt")) +} + +fn selectors(endpoint: &Value, request: &FabricRekeyRequest) -> bool { + endpoint["old_node"] == request.old_node + && endpoint["next_node"] == request.next_node + && endpoint["deadline"].as_u64() == Some(request.deadline) +} + +fn node_id(node: &str) -> bool { + node.len() == 64 && node.bytes().all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) +} + +impl Store { + fn own_endpoint_fields(&self) -> Result { + let key = self.member_public_key().context("this node has no fleet member key")?; + Ok(self.latest_claim(&format!("host/{}", self.origin), Some("fleet.member-endpoints"))? + .map(|claim| claim.body["fields"].clone()) + .filter(|fields| fields["member_key"] == key) + .unwrap_or_else(|| json!({"endpoints":[]}))) + } + + pub(super) fn preserve_fabric_rekey(&self, mode: &str, endpoints: &[Value], clear: bool) -> Result> { + // Metadata is writable only through the validated transition operation, not the + // worker's generic endpoint publication endpoint. + let mut result: Vec<_> = endpoints.iter() + .filter(|entry| !metadata(entry) && (mode == "listening" || entry["transport"] != "fabric")) + .cloned().collect(); + if self.member_public_key().is_none() { return Ok(result); } + let fields = self.own_endpoint_fields()?; + if let Some(previous) = fields["endpoints"].as_array() { + if clear { + // An explicit transport withdrawal wins over rotation metadata. + // The settings CLI serializes it with rekey using its fleet lock. + result.extend(previous.iter().filter(|entry| entry["transport"] == "fabric-rekey-receipt").cloned()); + return Ok(result); + } + result.extend(previous.iter().filter(|entry| metadata(entry)).cloned()); + if let Some(rotation) = previous.iter().find(|entry| entry["transport"] == "fabric-rekey") { + let node = if rotation["state"] == "committed" { &rotation["next_node"] } else { &rotation["old_node"] }; + for endpoint in &mut result { + if endpoint["transport"] == "fabric" { endpoint["node"] = node.clone(); } + } + // A refresh before Fabric is ready must not remove the transition's route. + if !result.iter().any(|entry| entry["transport"] == "fabric") { + result.extend(previous.iter().filter(|entry| entry["transport"] == "fabric").cloned()); + } + } + } + Ok(result) + } + + /// Publish and seal before replying. `receipt` is called by the local adapter only + /// after its verified runtime snapshot proves grants for the selected state installed. + pub fn fabric_rekey(&self, request: &FabricRekeyRequest, build: &str) -> Result { + let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(); + self.fabric_rekey_at(request, build, now) + } + + fn fabric_rekey_at(&self, request: &FabricRekeyRequest, build: &str, now: u64) -> Result { + anyhow::ensure!(node_id(&request.old_node) && node_id(&request.next_node) + && request.old_node != request.next_node, "rekey requires distinct canonical Fabric node ids"); + let _publication = self.fleet_endpoints_publication.lock().unwrap_or_else(PoisonError::into_inner); + let membership = self.fleet_membership()?; + let crate::fleet::MemberState::Current(own) = membership.state(&self.origin) else { + anyhow::bail!("rekey requires a current local fleet member"); + }; + anyhow::ensure!(Some(own.member_key.as_str()) == self.member_public_key().as_deref(), "local member key mismatch"); + let fields = self.own_endpoint_fields()?; + let mut endpoints = fields["endpoints"].as_array().cloned().unwrap_or_default(); + if request.action == FabricRekeyAction::Receipt { + let target_key = request.member_key.as_deref().context("receipt requires member_key")?; + anyhow::ensure!(target_key != own.member_key, "own receipts do not acknowledge a rekey"); + let state = request.state.context("receipt requires state")?; + let expected_state = serde_json::to_value(state)?; + let target = membership.incarnations().find(|member| member.member_key == target_key + && matches!(membership.state(&member.name), crate::fleet::MemberState::Current(current) if current.member_key == target_key)) + .context("receipt target is not a current member")?; + anyhow::ensure!(request.deadline > now && target.endpoints.iter().any(|entry| + entry["transport"] == "fabric-rekey" && selectors(entry, request) + && entry["state"] == expected_state), "receipt must match an exact live target transition"); + let receipt = json!({"transport":"fabric-rekey-receipt", "member_key":target_key, + "old_node":request.old_node,"next_node":request.next_node, + "deadline":request.deadline,"state":state}); + endpoints.retain(|entry| !(entry["transport"] == "fabric-rekey-receipt" && entry["member_key"] == target_key)); + endpoints.push(receipt); + } else { + anyhow::ensure!(request.member_key.is_none() && request.state.is_none(), "member_key and state are receipt-only selectors"); + let previous = endpoints.iter().find(|entry| entry["transport"] == "fabric-rekey"); + let next_state = match request.action { + FabricRekeyAction::Prepare => { + anyhow::ensure!(request.deadline > now && request.deadline <= now.saturating_add(600), "prepare deadline must be within 600 seconds"); + if let Some(previous) = previous { + if selectors(previous, request) { + anyhow::ensure!(previous["state"] == "prepared", "transition already ended preparation"); + } else { + anyhow::ensure!(previous["state"] == "aborted" || previous["deadline"].as_u64().is_some_and(|deadline| deadline <= now), "conflicting active rekey"); + // A pending Fabric key may be reused by a new operation only + // after an aborted operation's original window has elapsed. + let ended = previous["state"] == "aborted" + && previous["deadline"].as_u64().is_some_and(|deadline| deadline <= now); + anyhow::ensure!(previous["old_node"] != request.old_node + || previous["next_node"] != request.next_node || ended, + "retries cannot extend a rekey deadline"); + } + } + anyhow::ensure!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == request.old_node), "prepare old_node is not the current Fabric member endpoint"); + "prepared" + } + FabricRekeyAction::Commit | FabricRekeyAction::Abort => { + if request.action == FabricRekeyAction::Abort && previous.is_none() { + anyhow::ensure!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric" + && entry["node"] == request.old_node), "abort without preparation requires the current old endpoint"); + return Ok(false); + } + let previous = previous.context("no prepared rekey")?; + anyhow::ensure!(selectors(previous, request), "transition selectors do not match"); + if request.action == FabricRekeyAction::Commit { + anyhow::ensure!(request.deadline > now && matches!(previous["state"].as_str(), Some("prepared" | "committed")), "commit requires a live prepared transition"); + "committed" + } else { + // Roll back even after a committed acknowledgement timeout: the CLI + // has not activated the next local identity yet. + anyhow::ensure!(matches!(previous["state"].as_str(), Some("prepared" | "committed" | "aborted")), "invalid abort transition"); + "aborted" + } + } + FabricRekeyAction::Receipt => unreachable!(), + }; + endpoints.retain(|entry| entry["transport"] != "fabric-rekey"); + endpoints.push(json!({"transport":"fabric-rekey","old_node":request.old_node, + "next_node":request.next_node,"deadline":request.deadline,"state":next_state})); + for endpoint in &mut endpoints { + if endpoint["transport"] == "fabric" { + endpoint["node"] = json!(if next_state == "committed" { &request.next_node } else { &request.old_node }); + } + } + } + self.publish_fleet_endpoints_locked(fields["mode"].as_str().unwrap_or(&own.mode), &endpoints, build) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::fleet::MemberKey; + use crate::store::runtime::Plain; + + fn configure(store: &Store) -> Arc { + store.bind_fleet("rekey-test").unwrap(); + let (key, _) = MemberKey::generate().unwrap(); + let key = Arc::new(key); + store.pin_fleet_anchor(key.public()).unwrap(); + store.set_member_key(Some(key.clone())).unwrap(); + store.admit_fleet_anchor("rekey-test", key.public(), "listening").unwrap(); + store.publish_fleet_endpoints("listening", &[json!({ + "transport":"fabric","node":"a".repeat(64),"protocol":"st3/fleet/rekey-test" + })], "test").unwrap(); + key + } + + fn request(action: FabricRekeyAction) -> FabricRekeyRequest { + FabricRekeyRequest { + action, old_node: "a".repeat(64), next_node: "b".repeat(64), + deadline: 1600, member_key: None, state: None, + } + } + + fn current_endpoints(store: &Store) -> Vec { + let view = store.fleet_view_sealed().unwrap(); + view.members.iter().find(|member| member.name == "anchor").unwrap().endpoints.clone() + } + + #[test] + fn signed_transition_survives_worker_refresh_and_abort_from_committed() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + configure(&store); + let mut request = request(FabricRekeyAction::Prepare); + assert!(store.fabric_rekey_at(&request, "test", 1000).unwrap()); + assert!(!store.fabric_rekey_at(&request, "test", 1001).unwrap()); + let prepared = current_endpoints(&store); + assert!(prepared.iter().any(|entry| entry["state"] == "prepared")); + let routes = crate::fleet::transport::routes_from_endpoints(&prepared, + crate::fleet::transport::LocalTransports { fabric: true, ..Default::default() }); + assert_eq!(routes.len(), 1, "signed rekey metadata must never become a route"); + store.publish_fleet_endpoints("listening", &[json!({ + "transport":"fabric","node":"c".repeat(64),"protocol":"st3/fleet/rekey-test" + })], "test").unwrap(); + assert_eq!(current_endpoints(&store), prepared); + request.action = FabricRekeyAction::Commit; + assert!(store.fabric_rekey_at(&request, "test", 1002).unwrap()); + assert!(!store.fabric_rekey_at(&request, "test", 1003).unwrap()); + store.publish_fleet_endpoints("listening", &[], "test").unwrap(); + let committed = current_endpoints(&store); + assert!(committed.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == request.next_node)); + request.action = FabricRekeyAction::Abort; + assert!(store.fabric_rekey_at(&request, "test", 1601).unwrap()); + assert!(!store.fabric_rekey_at(&request, "test", 1602).unwrap()); + let aborted = current_endpoints(&store); + assert!(aborted.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == request.old_node)); + assert!(aborted.iter().any(|entry| entry["state"] == "aborted")); + } + + #[test] + fn prepare_refuses_wrong_identity_conflicts_extended_retries_and_expired_commit() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + configure(&store); + let mut invalid = request(FabricRekeyAction::Prepare); + invalid.old_node = "c".repeat(64); + assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err()); + invalid = request(FabricRekeyAction::Prepare); + invalid.deadline = 1601; + assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err()); + let prepared = request(FabricRekeyAction::Prepare); + store.fabric_rekey_at(&prepared, "test", 1000).unwrap(); + invalid.deadline = 1600; + invalid.next_node = "c".repeat(64); + assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err()); + invalid = prepared.clone(); + invalid.deadline = 1601; + assert!(store.fabric_rekey_at(&invalid, "test", 1001).is_err()); + assert!(store.fabric_rekey_at(&invalid, "test", 1600).is_err()); + invalid = prepared; + invalid.action = FabricRekeyAction::Commit; + assert!(store.fabric_rekey_at(&invalid, "test", 1600).is_err()); + } + + #[test] + fn crashed_prepare_resumes_original_signed_deadline() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("claims.db"); + let store = Store::open(&path, "anchor", Arc::new(Plain)).unwrap(); + let key = configure(&store); + let prepared = request(FabricRekeyAction::Prepare); + store.fabric_rekey_at(&prepared, "test", 1000).unwrap(); + drop(store); + let store = Store::open(&path, "anchor", Arc::new(Plain)).unwrap(); + store.set_member_key(Some(key)).unwrap(); + assert!(!store.fabric_rekey_at(&prepared, "test", 1100).unwrap()); + let mut extended = prepared.clone(); + extended.deadline += 1; + assert!(store.fabric_rekey_at(&extended, "test", 1100).is_err()); + assert!(current_endpoints(&store).iter().any(|entry| entry["deadline"] == 1600)); + } + + #[test] + fn abort_before_first_prepare_is_safe_and_receipts_refuse_own_or_missing_target() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + let key = configure(&store); + assert!(!store.fabric_rekey_at(&request(FabricRekeyAction::Abort), "test", 1601).unwrap()); + let mut receipt = request(FabricRekeyAction::Receipt); + receipt.member_key = Some(key.public().to_owned()); + receipt.state = Some(FabricRekeyState::Prepared); + assert!(store.fabric_rekey_at(&receipt, "test", 1000).is_err()); + receipt.member_key = Some("missing".into()); + assert!(store.fabric_rekey_at(&receipt, "test", 1000).is_err()); + } + + #[test] + fn receipt_is_signed_and_bound_to_remote_live_state_and_all_selectors() { + let anchor = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + let anchor_key = configure(&anchor); + let peer = Store::open_memory("peer", Arc::new(Plain)).unwrap(); + let (peer_key, _) = MemberKey::generate().unwrap(); + let peer_key = Arc::new(peer_key); + peer.bind_fleet("rekey-test").unwrap(); + peer.pin_fleet_anchor(anchor_key.public()).unwrap(); + peer.set_member_key(Some(peer_key.clone())).unwrap(); + anchor.append_claim(&ClaimInput { + subject: "host/peer".into(), kind: "fleet.member-admitted".into(), actor: None, + fields: BTreeMap::from([ + ("fleet_id".into(), json!("rekey-test")), + ("member_key".into(), json!(peer_key.public())), + ("via".into(), json!("invite")), ("mode".into(), json!("listening")), + ]), + evidence: Vec::new(), expected_subject: None, idempotency_key: None, + }).unwrap(); + let mut rotation = request(FabricRekeyAction::Prepare); + anchor.fabric_rekey_at(&rotation, "test", 1000).unwrap(); + let exchange = anchor.export_replication_exchange("rekey-test", &ReplicationInventory::default()).unwrap(); + peer.receive_replication_exchange("anchor", "rekey-test", &exchange).unwrap(); + peer.validate_replication_backlog().unwrap(); + peer.apply_replication_repairs().unwrap(); + peer.project_replication_backlog().unwrap(); + let mut receipt = request(FabricRekeyAction::Receipt); + receipt.member_key = Some(anchor_key.public().to_owned()); + receipt.state = Some(FabricRekeyState::Committed); + assert!(peer.fabric_rekey_at(&receipt, "test", 1001).is_err()); + receipt.state = Some(FabricRekeyState::Prepared); + receipt.deadline -= 1; + assert!(peer.fabric_rekey_at(&receipt, "test", 1001).is_err()); + receipt.deadline += 1; + assert!(peer.fabric_rekey_at(&receipt, "test", 1001).unwrap()); + assert!(!peer.fabric_rekey_at(&receipt, "test", 1002).unwrap()); + peer.publish_fleet_endpoints("listening", &[], "test").unwrap(); + let view = peer.fleet_view_sealed().unwrap(); + let own = view.members.iter().find(|member| member.name == "peer").unwrap(); + assert!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric-rekey-receipt" + && entry["state"] == "prepared" && entry["member_key"] == anchor_key.public())); + rotation.action = FabricRekeyAction::Commit; + anchor.fabric_rekey_at(&rotation, "test", 1003).unwrap(); + let exchange = anchor.export_replication_exchange("rekey-test", &ReplicationInventory::default()).unwrap(); + peer.receive_replication_exchange("anchor", "rekey-test", &exchange).unwrap(); + peer.validate_replication_backlog().unwrap(); + peer.apply_replication_repairs().unwrap(); + peer.project_replication_backlog().unwrap(); + assert!(peer.fabric_rekey_at(&receipt, "test", 1004).is_err()); + receipt.state = Some(FabricRekeyState::Committed); + assert!(peer.fabric_rekey_at(&receipt, "test", 1004).unwrap()); + assert!(peer.fabric_rekey_at(&receipt, "test", 1600).is_err()); + peer.publish_fleet_endpoints_clearing_fabric_rekey("listening", &[], "test").unwrap(); + let view = peer.fleet_view_sealed().unwrap(); + let own = view.members.iter().find(|member| member.name == "peer").unwrap(); + assert!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric-rekey-receipt" + && entry["state"] == "committed"), "settings clear must preserve remote receipts"); + } + + #[test] + fn explicit_clear_withdraws_committed_route_without_resurrection() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + configure(&store); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(); + let mut rotation = request(FabricRekeyAction::Prepare); + rotation.deadline = now + 500; + store.fabric_rekey_at(&rotation, "test", now).unwrap(); + rotation.action = FabricRekeyAction::Commit; + store.fabric_rekey_at(&rotation, "test", now).unwrap(); + assert!(store.publish_fleet_endpoints_clearing_fabric_rekey("listening", &[], "test").unwrap()); + assert!(current_endpoints(&store).is_empty()); + assert!(!store.publish_fleet_endpoints("listening", &[], "test").unwrap()); + assert!(current_endpoints(&store).is_empty()); + } + + #[test] + fn dial_out_publication_clears_completed_rotation_and_never_advertises_fabric() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + configure(&store); + let rotation = request(FabricRekeyAction::Prepare); + store.fabric_rekey_at(&rotation, "test", 1000).unwrap(); + // Its historical deadline is expired: changing mode can withdraw the route. + store.publish_fleet_endpoints("dial-out", &[json!({ + "transport":"fabric","node":rotation.old_node,"protocol":"st3/fleet/rekey-test" + })], "test").unwrap(); + assert!(current_endpoints(&store).is_empty()); + assert_eq!(store.fleet_view_sealed().unwrap().members[0].mode, "dial-out"); + } + + #[test] + fn expired_aborted_operation_can_start_again_with_same_pending_key() { + let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap(); + configure(&store); + let mut rotation = request(FabricRekeyAction::Prepare); + store.fabric_rekey_at(&rotation, "test", 1000).unwrap(); + rotation.action = FabricRekeyAction::Abort; + store.fabric_rekey_at(&rotation, "test", 1001).unwrap(); + rotation.action = FabricRekeyAction::Prepare; + rotation.deadline = 1700; + assert!(store.fabric_rekey_at(&rotation, "test", 1100).is_err()); + assert!(store.fabric_rekey_at(&rotation, "test", 1600).unwrap()); + assert!(current_endpoints(&store).iter().any(|entry| + entry["transport"] == "fabric-rekey" && entry["state"] == "prepared" + && entry["deadline"] == 1700)); + } +} diff --git a/docs/fleet-join.md b/docs/fleet-join.md index 6d30a0ac260..371aa555846 100644 --- a/docs/fleet-join.md +++ b/docs/fleet-join.md @@ -284,6 +284,46 @@ The worker exposes its loopback listener with the exposure. The adapter maintains trust and grants; Fractal does not edit `peers.toml`. +#### Re-key an existing Fabric member + +Run this on the machine whose Fabric identity must change: + +```sh +fractal fleet transports rekey fabric +``` + +Use Fabric's identity lifecycle and runtime expiry APIs (v0.9.4), a running +Fractal daemon and replication worker, and a running `st-fabric-adapter` on +the other members. At least one other current member must confirm both policy +phases before the original ten-minute deadline. A single-member or disconnected +fleet cannot complete a re-key. + +The command asks Fabric to prepare a key without changing its live identity. +Fractal publishes a signed rotation announcement under the same member key over +the old route. Peers temporarily grant the prepared NodeID until the immutable +deadline, keeping the old NodeID permanent. Their adapters publish signed receipts +only after checking the installed Fabric policy. After one receipt, Fractal +publishes the committed endpoint; peers make the new NodeID permanent and expire +the old NodeID at the original deadline. Fractal waits for a committed receipt +before asking Fabric to activate the prepared identity. Fabric never serves two +local identities at once. Adapter policy changes are revision-checked atomic +batches and do not grant terminal access. + +The command persists progress in `fleet/fabric-rekey.json` under the Fractal state +directory. After an interruption, run the same command: it resumes the original +deadline, or retries the same guarded activation after the activation decision. +Do not delete the journal, edit Fabric's private keys, or run `fabric identity +activate` separately. An uncertain activation response is not a rollback. + +If no peer confirms before the deadline, Fractal publishes an abort and leaves +the old local identity active. A stopped adapter cannot extend staged access: +Fabric enforces the installed expiry even while the adapter is down. Abort +withdrawal and restoration still require claim delivery and reconciliation. +An offline peer may miss the cutover and must learn current signed membership +through another reachable fleet member before it can authorize the new NodeID. +Static and other-owner rights are not revoked: remove them at their source if +complete retirement of the old NodeID is required. + #### Adapter helper boundary The CLI discovers `st-fabric-adapter` beside its executable or on `PATH`; diff --git a/docs/site/guides/two-machines.md b/docs/site/guides/two-machines.md index c5b701a8530..4ec962dda2e 100644 --- a/docs/site/guides/two-machines.md +++ b/docs/site/guides/two-machines.md @@ -78,4 +78,26 @@ mailbox, internal, and administration routes still require their appropriate loc transport. Fabric routing does not bypass client API pairing, bearer credentials, or device signing: the remote daemon applies the same client authorization rules. +### Re-key a Fabric member + +With Fabric v0.9.4 and the fleet adapters running, execute one command on the +machine whose identity changes: + +```sh +fractal fleet transports rekey fabric +``` + +Fractal prepares a new Fabric NodeID, announces it under the existing signed +member identity through the old route, and waits for a peer adapter's installed +policy receipt. It commits the new endpoint and waits for a second receipt before +activating the new local identity. Both waits share a ten-minute deadline. +Fabric enforces expiring overlap grants; it never serves two local identities. +If no peer confirms in time, the command aborts without activating the new key. + +Run the same command after a crash to resume its durable journal and guarded +activation. Do not rotate keys by hand. Offline members may need a reachable +member to deliver the signed cutover before their Fabric route works again. +See the [re-key procedure](https://github.com/compoundingtech/fractal/blob/main/docs/fleet-join.md#re-key-an-existing-fabric-member) +for rollback, static grants, and recovery boundaries. + See [Fleet membership](fleet-join.md) for trust and invitation behavior and [replication](https://github.com/compoundingtech/fractal/blob/main/docs/st3/replication.md) for digests and recovery. Keep raw state and keys when investigating inconsistent history; do not reset a replica merely to make its status look healthy. diff --git a/flake.lock b/flake.lock index 85b859c27fb..1dd1da56cd5 100644 --- a/flake.lock +++ b/flake.lock @@ -52,17 +52,17 @@ ] }, "locked": { - "lastModified": 1792912755, - "narHash": "sha256-Zs+3vv/BbQTFVbuMpC3+ENdnjVIDiOwlrPjyhP23RUk=", + "lastModified": 1793345926, + "narHash": "sha256-hhy7n28KPOmsdR5DXvnMVlV1i5LcBW7GZX8ctaPw5ZY=", "owner": "compoundingtech", "repo": "fabric", - "rev": "1e6e93e9a662a40749f91d63eeecc4e98de114bd", + "rev": "bd45ec9cac98c217266855a696f772d364fe66a5", "type": "github" }, "original": { "owner": "compoundingtech", "repo": "fabric", - "rev": "1e6e93e9a662a40749f91d63eeecc4e98de114bd", + "rev": "bd45ec9cac98c217266855a696f772d364fe66a5", "type": "github" } }, diff --git a/flake.nix b/flake.nix index bda897f88b4..e3797b9b2cf 100644 --- a/flake.nix +++ b/flake.nix @@ -9,8 +9,8 @@ # The runtime, Rust crates and native terminal library share one producer revision. pty.url = "github:compoundingtech/pty/79d526d1be4f77bb3764dc1d880492806f8a07fc"; pty.inputs.nixpkgs.follows = "nixpkgs"; - # Merged pairing stack, including the daemon-identity restricted client. - fabric.url = "github:compoundingtech/fabric/1e6e93e9a662a40749f91d63eeecc4e98de114bd"; + # Identity prepare/activate recovery fixes; move to v0.9.4 before readiness. + fabric.url = "github:compoundingtech/fabric/bd45ec9cac98c217266855a696f772d364fe66a5"; fabric.inputs.nixpkgs.follows = "nixpkgs"; # Shared CI generators and the `otelite` collector used by release-integration. # Keep the build toolchain aligned with the source member in megarepo.kdl. diff --git a/release-notes/fabric-rekey-command.json b/release-notes/fabric-rekey-command.json new file mode 100644 index 00000000000..d61f9cc2286 --- /dev/null +++ b/release-notes/fabric-rekey-command.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Re-key a Fabric fleet member in one command with signed policy receipts and bounded overlap", + "replay": {"status": "none", "detail": "Existing signed member-endpoint claims carry rotation and receipt metadata; no claim replay is required."}, + "schema": {"status": "none", "detail": "No database schema change; fleet member endpoints include non-routing Fabric rotation and receipt metadata."}, + "rules": {"status": "changed", "detail": "Adapters stage an expiring next NodeID, confirm installed policy, and commit the new endpoint before guarded local activation."}, + "compatibility": {"status": "changed", "detail": "Re-key requires Fabric v0.9.4 identity and runtime expiry APIs and updated Fractal adapters on confirming peers."}, + "downtime": {"status": "changed", "detail": "Fabric activation restarts its local endpoint; existing Fabric tunnels reconnect. Offline peers may need membership recovery."}, + "manual": {"status": "changed", "detail": "Run fractal fleet transports rekey fabric on the rotating member. At least one remote member must confirm both phases within ten minutes."}, + "recovery": {"status": "changed", "detail": "Rerun the same command after interruption to resume its durable journal without extending the deadline. No confirmation aborts with the old identity active; uncertain activation retries guarded reconciliation. Static and other-owner grants remain operator-owned."} +}