Fabric re-key R6: adapter overlap (merge together with R6b) (hosted #4995, re-key v2)

crates/fractal-fabric-adapter/src/reconcile.rs

@@ -190,6 +190,24 @@ async fn send(socket: &Path, request: &Value) -> Result<Reply> {
     })
 }
 
+async fn supports_expiry(socket: &Path) -> Result<bool> {
+    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::<Value>(&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::<u64>().ok()),
+        version.next().and_then(|v| v.parse::<u64>().ok()),
+        version.next().and_then(|v| v.parse::<u64>().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<Snapshot> {
     let mut first: Option<Snapshot> = None;
     let mut cursor = None;
@@ -319,8 +337,65 @@ fn desired(
 
 type DesiredPolicy = BTreeMap<String, BTreeSet<String>>;
 
+#[derive(Debug, Deserialize, Serialize)]
+struct Rotation {
+    old_node: String,
+    next_node: String,
+    deadline: u64,
+    state: String,
+}
+
 type Expiries = BTreeMap<String, u64>;
 
+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::<Vec<_>>();
+        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<Value> {
     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"

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<_>>(),
+            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"));
+    }
+}

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."}}