diff --git a/crates/fractal-fabric-adapter/src/main.rs b/crates/fractal-fabric-adapter/src/main.rs index 5d90e19fa94..cf59746d7d1 100644 --- a/crates/fractal-fabric-adapter/src/main.rs +++ b/crates/fractal-fabric-adapter/src/main.rs @@ -266,6 +266,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 8de48ffff32..1112a57d2e2 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -862,6 +862,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 { diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index 7bf310fef82..e0d476301df 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -1076,6 +1076,53 @@ async fn successful_mutation_without_verified_install_is_an_error() { 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(); @@ -1133,6 +1180,80 @@ fn own_metadata_and_noncurrent_members_never_authorize_overlap_or_receipts() { assert!(receipts.is_empty()); } +#[tokio::test] +async fn rekey_crash_recovery_replays_only_live_writer_batches() { + let deadline = (now_ms() / 1000) as u64 + 300; + for owner in [OWNER.to_owned(), format!("{RENAMED_OWNER_PREFIX}{FLEET}")] { + for (state, live_writer) in [("prepared", true), ("committed", false), ("aborted", false)] { + let root = tempfile::tempdir().unwrap(); + let mut policy = DesiredPolicy::from([("peer-node".into(), grants(FLEET))]); + if state == "committed" { + policy = DesiredPolicy::from([("next-node".into(), grants(FLEET))]); + } + let (expiries, _) = rotations(&rotation_view(state, deadline), "own-key", + &mut policy, (now_ms() / 1000) as u64).unwrap(); + let request = json!({ + "operation":"batch", "owner":owner, "request_id":"st-rekey-crash", + "expected_revision":7, "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} + ] + }); + save_journal(root.path(), &Journal { + pending: Some(Pending { request: request.clone(), created_unix_ms: now_ms() }), + ..Journal::default() + }).await.unwrap(); + let mut journal = load_journal(root.path()).await.unwrap(); + let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]); + resolve_pending(root.path(), &socket, &mut journal, &policy, &expiries, &BTreeSet::new()).await.unwrap(); + let requests = server.await.unwrap(); + if owner == OWNER && live_writer { + assert_eq!(requests[0]["request"], request); + } else { + assert_eq!(requests[0]["request"], + json!({"operation":"cancellation","request_id":"st-rekey-crash"})); + } + assert!(load_journal(root.path()).await.unwrap().pending.is_none()); + } + } +} + +#[test] +fn renamed_rekey_expiry_recovery_only_retains_or_removes_existing_grants() { + let owner = format!("{RENAMED_OWNER_PREFIX}{FLEET}"); + let services = [format!("{RENAMED_OWNER_PREFIX}{FLEET}"), "fractal-client-v0".into()]; + for state in ["prepared", "committed", "aborted"] { + let mut policy = DesiredPolicy::from([( + if state == "committed" { "next-node" } else { "peer-node" }.into(), + grants(FLEET), + )]); + let (expiries, _) = rotations(&rotation_view(state, 1300), "own-key", &mut policy, 1000).unwrap(); + let mut staged = peer("next-node", &owner, &[&services[0], &services[1]]); + staged["trust_sources"][0]["expires_at"] = json!(1300); + let snapshot: Snapshot = serde_json::from_value(list(vec![ + peer("peer-node", &owner, &[&services[0], &services[1]]), staged, + ], 7)["reply"]["snapshot"].clone()).unwrap(); + let retained = retained_policy(&snapshot, &owner, &policy); + let operations = plan(&snapshot, &owner, &retained, &expiries); + let expected = match state { + "prepared" => json!([]), + "committed" => json!([ + {"op":"remove_peer","node_id":"next-node"}, + {"op":"remove_peer","node_id":"peer-node"} + ]), + "aborted" => json!([{"op":"remove_peer","node_id":"next-node"}]), + _ => unreachable!(), + }; + assert_eq!(json!(operations), expected); + assert!(operations.iter().all(|operation| operation["op"] == "remove_peer")); + let pending = Pending { + created_unix_ms: now_ms(), + request: json!({"owner":owner,"operations":operations}), + }; + assert!(pending_matches_desired(&pending, &policy, &expiries)); + } +} fn serve_capability(root: &Path, capability: &'static str, replies: Vec) -> (PathBuf, tokio::task::JoinHandle>) { diff --git a/release-notes/fabric-rekey-adapter-receipts.json b/release-notes/fabric-rekey-adapter-receipts.json new file mode 100644 index 00000000000..9a60b3ca666 --- /dev/null +++ b/release-notes/fabric-rekey-adapter-receipts.json @@ -0,0 +1 @@ +{"version": 1,"summary": "Publish exact verified Fabric rekey receipts and preserve deny-only renamed-owner crash recovery","replay": {"status": "none","detail": "No graph replay."},"schema": {"status": "none","detail": "Reuses fleet.member-endpoints claims; no database migration."},"rules": {"status": "none","detail": "No checkpoint rule change."},"compatibility": {"status": "changed","detail": "Non-routing rekey metadata is preserved during worker refresh and removed by explicit transport withdrawal."},"downtime": {"status": "none","detail": "No service restart required."},"manual": {"status": "none","detail": "No manual step."},"recovery": {"status": "changed","detail": "Durable recovery replays live writer batches only; renamed-owner recovery cannot add grants or promote an expiring record."}}