diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index f23c45a0dc5..605f89ae157 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -190,6 +190,24 @@ async fn send(socket: &Path, request: &Value) -> Result { }) } +async fn supports_expiry(socket: &Path) -> Result { + let reply: Value = tokio::time::timeout(CONTROL_TIMEOUT, async { + let mut stream = tokio::net::UnixStream::connect(socket).await?; + stream.write_all(b"{\"type\":\"reachability_status\"}\n").await?; + let mut bytes = Vec::new(); + BufReader::new(stream).take(RESPONSE_LIMIT + 1).read_until(b'\n', &mut bytes).await?; + ensure!(bytes.len() as u64 <= RESPONSE_LIMIT && bytes.last() == Some(&b'\n'), + "Fabric version probe returned an incomplete or oversized response"); + serde_json::from_slice::(&bytes).context("invalid Fabric version probe response") + }).await??; + let mut version = reply["version"].as_str().unwrap_or_default().split(['.', '+', '-']); + let parsed = (version.next().and_then(|v| v.parse::().ok()), + version.next().and_then(|v| v.parse::().ok()), + version.next().and_then(|v| v.parse::().ok())); + Ok(reply["type"] == "reachability_status" + && matches!(parsed, (Some(major), Some(minor), Some(patch)) if (major, minor, patch) >= (0, 9, 4))) +} + async fn snapshot(socket: &Path) -> Result { let mut first: Option = None; let mut cursor = None; @@ -319,8 +337,65 @@ fn desired( type DesiredPolicy = BTreeMap>; +#[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 @@ -714,7 +789,7 @@ pub async fn reconcile( .members .iter() .any(|member| member.member_key == own_key && member.state == "current"); - let desired: DesiredPolicy = desired + let mut desired: DesiredPolicy = desired .into_iter() .map(|node| { let peer_current = view.members.iter().any(|member| { @@ -731,7 +806,20 @@ pub async fn reconcile( (node, services) }) .collect(); - let expiries = Expiries::new(); + let (mut expiries, rotations) = rotations(view, own_key, &mut desired, (now_ms() / 1000) as u64)?; + let blocked_reason = if expiries.is_empty() { None } else { + match supports_expiry(socket).await { + Ok(true) => None, + Ok(false) => Some("fabric_without_expiry: upgrade every participating Fabric daemon to v0.9.4 before rekey".to_owned()), + Err(error) => Some(format!("fabric_capability_unknown: {error:#}")), + } + }; + if blocked_reason.is_some() { + // Never turn expiring overlap into permanent authorization on an older daemon. + // Keep the old route; ordinary peer additions and revocations still reconcile. + for (_, rotation) in rotations { desired.remove(&rotation.next_node); } + expiries.clear(); + } let ended = view .members .iter() @@ -741,6 +829,10 @@ pub async fn reconcile( .filter_map(|endpoint| endpoint["node"].as_str().map(str::to_owned)) .collect(); let mut status = apply(state_dir, socket, fleet_id, desired, expiries, ended, false).await?; + if let Some(reason) = blocked_reason.filter(|_| status.state == "applied") { + status.state = "rekey_blocked".into(); + status.message = Some(reason); + } if status.state == "applied" && view.members.iter().any(|member| { member.state == "ended" diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index a6574fb84ca..3b083fb894d 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -993,6 +993,78 @@ 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 mut replies = vec![ + Some(list(before, 7)), Some(mutation(vec![], "complete")), Some(list(after, 8)), + ]; + if state != "aborted" { + replies.insert(0, Some(json!({"type":"reachability_status","version":"0.9.4+19e7f89"}))); + } + let (socket, server) = serve(root.path(), replies); + 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 + usize::from(state != "aborted")]["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(); @@ -1004,3 +1076,117 @@ async fn successful_mutation_without_verified_install_is_an_error() { server.await.unwrap(); } +#[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()); +} + + +#[tokio::test] +async fn fabric_without_expiry_blocks_rekey_but_applies_unrelated_changes() { + for capability in ["old", "timeout", "malformed", "truncated", "closed"] { + let root = tempfile::tempdir().unwrap(); + let socket = root.path().join("control.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let permanent = |node| peer(node, OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]); + let replies = vec![ + json!({"type":"reachability_status","version":"0.9.3+old"}), + list(vec![permanent("peer-node"), permanent("gone")], 7), + mutation(vec![], "complete"), + list(vec![permanent("peer-node"), permanent("new-node")], 8), + ]; + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + for mut reply in replies { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut line = String::new(); + BufReader::new(&mut stream).read_line(&mut line).await.unwrap(); + let request: Value = serde_json::from_str(&line).unwrap(); + if request["request"]["operations"].as_array().is_some_and(|ops| + ops.iter().any(|op| op["op"] == "set_expiry")) { + reply = refusal("unsupported_operation"); + } + let probing = request["type"] == "reachability_status"; + requests.push(request); + if probing && capability != "old" { + match capability { + "timeout" => tokio::time::sleep(CONTROL_TIMEOUT + Duration::from_millis(20)).await, + "malformed" => stream.write_all(b"not-json\n").await.unwrap(), + "truncated" => stream.write_all(b"{\"type\":\"reachability_status\"").await.unwrap(), + "closed" => (), + _ => unreachable!(), + } + continue; + } + let mut bytes = serde_json::to_vec(&reply).unwrap(); + bytes.push(b'\n'); + stream.write_all(&bytes).await.unwrap(); + } + requests + }); + let mut view = rotation_view("prepared", (now_ms() / 1000) as u64 + 300); + view.members.push(member("new", "new-key", "new-node", "current")); + let status = reconcile(root.path(), &socket, FLEET, "own-key", &view, &BTreeSet::new()).await.unwrap(); + assert_eq!(status.state, "rekey_blocked"); + assert!(status.message.unwrap().contains(if capability == "old" { + "fabric_without_expiry" + } else { "fabric_capability_unknown" })); + let requests = server.await.unwrap(); + let operations = requests[2]["request"]["operations"].as_array().unwrap(); + assert!(operations.iter().any(|op| op["op"] == "remove_peer" && op["node_id"] == "gone")); + assert!(operations.iter().any(|op| op["op"] == "add_peer" && op["node_id"] == "new-node")); + assert!(operations.iter().all(|op| op["op"] != "set_expiry" && op["node_id"] != "next-node")); + } +} diff --git a/release-notes/fabric-rekey-bounded-overlap.json b/release-notes/fabric-rekey-bounded-overlap.json new file mode 100644 index 00000000000..b5606c21257 --- /dev/null +++ b/release-notes/fabric-rekey-bounded-overlap.json @@ -0,0 +1 @@ +{"version": 1,"summary": "Reconcile signed Fabric identity rotations with bounded overlap; older Fabric blocks rekey without blocking unrelated policy changes","replay": {"status": "none","detail": "No graph replay."},"schema": {"status": "none","detail": "Reuses fleet.member-endpoints claims; Fabric expiry writes change durable runtime state."},"rules": {"status": "none","detail": "No checkpoint rule change."},"compatibility": {"status": "changed","detail": "The control-socket daemon version must support expiry before overlap is installed. Older daemons report rekey_blocked: fabric_without_expiry; failed probes report fabric_capability_unknown with the diagnostic. Ordinary expiry-free policy changes continue in both cases."},"downtime": {"status": "none","detail": "No Fractal downtime required."},"manual": {"status": "changed","detail": "Hard deploy order: upgrade Fabric to v0.9.4 on every participating member before any rekey, then deploy Fractal with R1-R7."},"recovery": {"status": "changed","detail": "After the first expiry write, no in-place Fabric v0.9.3 rollback is safe without an explicit state migration. Elapsed preparation cannot renew staging or replay an obsolete transaction."}}