From eaebea1eab9fc62ca765821be0b5359e23354505 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:01:57 +0100 Subject: [PATCH 1/2] Publish bounded Fabric rekeys through the fleet API agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- crates/fractal-api-core/src/api.rs | 5 + crates/fractal-api-core/src/api/fleet.rs | 77 ++++++++- crates/fractal-kernel/src/model.rs | 3 + crates/fractal-peer/src/peer.rs | 1 + crates/fractal/src/fleet_cli.rs | 2 + crates/smallclaims/src/store/fabric_rekey.rs | 160 +++++++++++++++++++ release-notes/fabric-rekey-internal-api.json | 32 ++++ 7 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 release-notes/fabric-rekey-internal-api.json diff --git a/crates/fractal-api-core/src/api.rs b/crates/fractal-api-core/src/api.rs index ea483c67db0..db9d465e3c7 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/crates/fractal-api-core/src/api.rs @@ -899,6 +899,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( "/v1/internal/replication/shard/{shard}/peers", @@ -9406,6 +9410,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 3f7dd49c14e..2002f06fa06 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 }))) +} + /// Publish the configured trust root, not an exporter's proposed replacement key. pub async fn fleet_anchor(State(state): State) -> Result, ApiError> { let store = state.store.clone(); @@ -380,3 +400,58 @@ pub async fn admission_override( } Ok(Json(claim)) } + +#[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-kernel/src/model.rs b/crates/fractal-kernel/src/model.rs index 39b55d42343..e07dd04cd8c 100644 --- a/crates/fractal-kernel/src/model.rs +++ b/crates/fractal-kernel/src/model.rs @@ -96,6 +96,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 6ed7dd9e02d..df494e4122e 100644 --- a/crates/fractal-peer/src/peer.rs +++ b/crates/fractal-peer/src/peer.rs @@ -1977,6 +1977,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 d9adaf24486..8bbf63fa3ff 100644 --- a/crates/fractal/src/fleet_cli.rs +++ b/crates/fractal/src/fleet_cli.rs @@ -1126,6 +1126,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 diff --git a/crates/smallclaims/src/store/fabric_rekey.rs b/crates/smallclaims/src/store/fabric_rekey.rs index 07b35ccf2c1..9c5daba8e80 100644 --- a/crates/smallclaims/src/store/fabric_rekey.rs +++ b/crates/smallclaims/src/store/fabric_rekey.rs @@ -228,4 +228,164 @@ mod tests { 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":"fractal/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/release-notes/fabric-rekey-internal-api.json b/release-notes/fabric-rekey-internal-api.json new file mode 100644 index 00000000000..03116b6d824 --- /dev/null +++ b/release-notes/fabric-rekey-internal-api.json @@ -0,0 +1,32 @@ +{ + "version": 1, + "summary": "Expose signed Fabric rekey publication and exact peer receipts through the internal fleet API", + "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": "Reject changed deadlines and mismatched receipts; resume signed state after a restart and preserve explicit transport withdrawal." + } +} From 8cddeed511e301695321f15e752ddf5dccec53ca Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 4 Nov 2026 08:40:34 +0100 Subject: [PATCH 2/2] test: account for signed Fabric rekey route in daemon costs agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- crates/fractal/tests/daemon_cost.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/fractal/tests/daemon_cost.rs b/crates/fractal/tests/daemon_cost.rs index b7f5f87858d..674f8facef8 100644 --- a/crates/fractal/tests/daemon_cost.rs +++ b/crates/fractal/tests/daemon_cost.rs @@ -166,6 +166,10 @@ const NOT_MEASURED: &[(&str, &str)] = &[ "GET /v1/internal/fleet/leave-state", "proxies a live replication worker's in-memory membership/receipt snapshot over its private socket; no claim-store scan; worker snapshot and API incarnation controls are unit-covered", ), + ( + "POST /v1/internal/fleet/fabric-rekey", + "requires a genuinely signed local fleet member and exact live transition metadata, absent from generated cost stores; an operator transition or adapter receipt folds current fleet membership and publishes one signed endpoint announcement; membership-history growth is not measured here; api::fleet::tests::signed_rekey_route_commits_preserves_refresh_and_rolls_back covers the real route", + ), ( "POST /v1/internal/fleet/pairing-admission", "explicit one-use fleet admission proxies the bounded signed request to the live worker; uses the existing invite redemption and verified membership fold, not a polling read path",