diff --git a/crates/fractal-fabric-adapter/src/pairing.rs b/crates/fractal-fabric-adapter/src/pairing.rs index 1db6b305c9b..1ecb8bb43e3 100644 --- a/crates/fractal-fabric-adapter/src/pairing.rs +++ b/crates/fractal-fabric-adapter/src/pairing.rs @@ -25,6 +25,8 @@ const LIMIT: u64 = 1024 * 1024; #[derive(Serialize, Deserialize)] struct Journal { fleet: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + owner: Option, creation: String, deadline: u64, attempt: Option, @@ -38,6 +40,24 @@ struct Journal { fn owner(fleet: &str) -> String { format!("st/fleet/{fleet}") } +fn accepted_owner(value: &Value, fleet: &str) -> bool { + value.as_str().is_some_and(|value| { + value.strip_prefix("st/fleet/") == Some(fleet) + || value.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(fleet) + }) +} +fn journal_owner(journal: &Journal) -> Result> { + let selected = journal.owner.as_deref().map_or_else( + || std::borrow::Cow::Owned(owner(&journal.fleet)), + std::borrow::Cow::Borrowed, + ); + ensure!( + selected.strip_prefix("st/fleet/") == Some(journal.fleet.as_str()) + || selected.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(journal.fleet.as_str()), + "pairing journal belongs to another fleet" + ); + Ok(selected) +} // Both the immutable permit and reciprocal policy are client-only. Static operator // terminal grants belong to Fabric configuration, never to this adapter owner. fn grants(fleet: &str) -> Value { @@ -177,7 +197,8 @@ async fn snapshot(client: &reqwest::Client) -> Result { bytes.extend_from_slice(&chunk); } let snapshot = - serde_json::from_slice::>(&bytes)?.value; + serde_json::from_slice::>(&bytes)? + .value; ensure!( snapshot.persistence_error.is_none() && snapshot.membership.anchor.is_some(), "st membership is not durable and verified" @@ -261,8 +282,9 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re .iter() .any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == gateway - && endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet))), + && endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet) + })), "sponsor lacks a unique current signed endpoint for the Fabric gateway" ); let deadline = code.expires_at.min((super::now_ms() / 1000) as u64 + 600); @@ -275,6 +297,7 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re ); let mut journal = Journal { fleet: fleet.clone(), + owner: None, creation: creation.clone(), deadline, attempt: None, @@ -351,7 +374,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu } let reply = control( socket, - &json!({"type":"pairing","request":{"operation":"query","owner":owner(&journal.fleet),"creation_id":journal.creation}}), + &json!({"type":"pairing","request":{"operation":"query","owner":journal_owner(&journal)?,"creation_id":journal.creation}}), ) .await?; // Fabric prunes terminal receipts after 24 hours and reports this exact @@ -385,7 +408,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu journal.confirmed = state_name == "confirmed"; super::save(&path, &journal)?; } else if journal.attempt.is_none() { - local(socket, json!({"operation":"cancel","owner":owner(&journal.fleet),"creation_id":journal.creation})).await?; + local(socket, json!({"operation":"cancel","owner":journal_owner(&journal)?,"creation_id":journal.creation})).await?; } else if state_name == "pending" && process(client, socket, &path, &mut journal, status) .await @@ -428,7 +451,7 @@ async fn process( if journal.response.is_none() { let reply = local( socket, - json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt}), + json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt}), ) .await?; let message = &reply["exchange"]["message"]; @@ -467,8 +490,8 @@ async fn process( super::save(path, journal)?; } let payload = journal.response.as_ref().context("missing answer")?; - local(socket, json!({"operation":"pair_send","owner":owner(&journal.fleet),"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?; - let reply = local(socket, json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt,"ack":1})).await?; + local(socket, json!({"operation":"pair_send","owner":journal_owner(journal)?,"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?; + let reply = local(socket, json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt,"ack":1})).await?; let message = &reply["exchange"]["message"]; if message.is_null() { return Ok(()); @@ -499,7 +522,7 @@ async fn process( &journal.fleet, )?; let policy_revision = revision(socket).await?; - let confirmation = json!({"operation":"confirm","owner":owner(&journal.fleet),"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")}); + let confirmation = json!({"operation":"confirm","owner":journal_owner(journal)?,"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")}); journal.confirmation = Some(confirmation.clone()); super::save(path, journal)?; finish_confirmation(socket, path, journal, confirmation).await?; @@ -512,6 +535,7 @@ async fn finish_confirmation( journal: &mut Journal, confirmation: Value, ) -> Result<()> { + ensure!(journal_owner(journal)? == owner(&journal.fleet), "renamed-owner permit remains pending; reader phase cannot confirm new authority"); let reply = control(socket, &json!({"type":"pairing","request":confirmation})).await?; if reply["type"] == "pairing" && reply["reply"]["error"]["code"] == "policy_revision_conflict" { // This explicit refusal cannot have committed. Only this case may discard @@ -728,8 +752,12 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> { .iter() .any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == pairing.gateway - && endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &sealed.fleet_id))), + && endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol( + protocol, + &sealed.fleet_id, + ) + })), "authenticated sponsor lacks signed exact Fabric endpoint" ); if let Some(resume) = &start.resume { @@ -748,10 +776,12 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> { let mutation = if path.exists() { let retained: Value = serde_json::from_slice(&std::fs::read(&path)?)?; ensure!( - retained["request"]["owner"] == owner(&sealed.fleet_id) + accepted_owner(&retained["request"]["owner"], &sealed.fleet_id) && retained["request"]["operations"][0]["node_id"] == pairing.gateway, "retained reciprocal mutation belongs to another identity" ); + ensure!(retained["request"]["owner"] == owner(&sealed.fleet_id), + "renamed-owner reciprocal addition remains pending; reader phase cannot replay new authority"); retained } else { json!({"type":"runtime_peers","request":{"operation":"batch","owner":owner(&sealed.fleet_id),"request_id":format!("pair-{}", pairing.attempt),"expected_revision":revision(socket).await?,"operations":[{"op":"add_peer","node_id":pairing.gateway,"direction":"normal"},{"op":"set_grants","node_id":pairing.gateway,"grants":grants(&sealed.fleet_id)}]}}) diff --git a/crates/fractal-fabric-adapter/src/pairing_tests.rs b/crates/fractal-fabric-adapter/src/pairing_tests.rs index 6872eefcc08..37f3d713041 100644 --- a/crates/fractal-fabric-adapter/src/pairing_tests.rs +++ b/crates/fractal-fabric-adapter/src/pairing_tests.rs @@ -20,6 +20,7 @@ fn request(node: &str) -> JoinRequest { fn journal() -> Journal { Journal { fleet: "fleet".into(), + owner: None, creation: "creation".into(), deadline: 600, attempt: Some("attempt".into()), @@ -66,6 +67,29 @@ fn permit_and_reciprocal_policy_grants_are_client_only() { assert_eq!(grants("fleet"), json!(["st3/fleet/fleet", "st3-client-v0"])); } +#[test] +fn recovered_pairing_owner_is_exact_and_legacy_journals_keep_their_owner() { + let old = serde_json::to_value(journal()).unwrap(); + assert!(old.get("owner").is_none()); + let mut recovered: Journal = serde_json::from_value(old).unwrap(); + assert_eq!(journal_owner(&recovered).unwrap(), "st/fleet/fleet"); + for owner in ["st/fleet/fleet", "fractal/fleet/fleet"] { + recovered.owner = Some(owner.into()); + assert_eq!(journal_owner(&recovered).unwrap(), owner); + } + recovered.owner = Some("fractal/fleet/other".into()); + assert!(journal_owner(&recovered).is_err()); +} + +#[tokio::test] +async fn renamed_permit_cannot_confirm_or_install_grants() { + let mut journal = journal(); + journal.owner = Some(format!("{}fleet", super::super::reconcile::RENAMED_OWNER_PREFIX)); + assert!(finish_confirmation(Path::new("absent.sock"), Path::new("absent.json"), + &mut journal, json!({"operation":"confirm"})).await.unwrap_err() + .to_string().contains("remains pending")); +} + #[test] fn quarantined_pairing_writes_fail_closed_without_accepting_a_status() { let error = pairing_reply(json!({ diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index 8ffa9634491..369a88d9439 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -17,6 +17,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use smallclaims::fleet::{FleetView, MemberView}; const REQUEST_LIMIT: usize = 1024 * 1024; +pub(super) const RENAMED_OWNER_PREFIX: &str = concat!("fractal", "/fleet/"); const RESPONSE_LIMIT: u64 = 4 * 1024 * 1024; const CONTROL_TIMEOUT: Duration = Duration::from_secs(5); @@ -220,6 +221,46 @@ fn grants(fleet_id: &str) -> BTreeSet { [format!("st3/fleet/{fleet_id}"), "st3-client-v0".into()].into() } +fn accepted_grants(fleet: &str) -> BTreeSet { + grants(fleet) + .into_iter() + .chain([format!("{RENAMED_OWNER_PREFIX}{fleet}"), "fractal-client-v0".into()]) + .collect() +} + +fn service_allowed(expected: &BTreeSet, service: &str) -> bool { + expected.contains(service) + || (service == "fractal-client-v0" && expected.contains("st3-client-v0")) + || service.strip_prefix(RENAMED_OWNER_PREFIX).is_some_and(|fleet| { + expected + .iter() + .any(|grant| grant.strip_prefix("st3/fleet/") == Some(fleet)) + }) +} + +// A reader may reduce an existing renamed owner's authority, never publish new authority. +fn retained_policy(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> DesiredPolicy { + snapshot + .peers + .iter() + .filter_map(|peer| { + let expected = desired.get(&peer.node_id)?; + if !peer + .trust_sources + .iter() + .any(|source| source.owned_by(owner)) + { + return None; + } + let actual: BTreeSet<_> = peer.services.iter() + .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner))) + .map(|(service, _)| service.clone()).collect(); + actual.iter().all(|service| service_allowed(expected, service)) + .then(|| (peer.node_id.clone(), actual)) + }) + .collect() +} + fn desired( view: &FleetView, own_key: &str, @@ -255,8 +296,9 @@ fn desired( } // Do not turn arbitrary signed endpoint names into Fabric grants. ensure!( - endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id)), + endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id) + }), "member {} advertises a nonstandard Fabric replication protocol; expected st3/fleet/{fleet_id} or fractal/fleet/{fleet_id}", member.name ); @@ -318,7 +360,7 @@ fn residual(peers: &[Peer], desired: &DesiredPolicy, grants: &BTreeSet) grants.contains(*service) && !desired .get(&peer.node_id) - .is_some_and(|expected| expected.contains(*service)) + .is_some_and(|expected| service_allowed(expected, service)) }) .map(|(service, sources)| { format!( @@ -392,6 +434,10 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> { } fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { + if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with(RENAMED_OWNER_PREFIX)) { + return pending.request["operations"].as_array().is_some_and(|operations| + operations.iter().all(|operation| operation["op"] == "remove_peer")); + } pending.request["operations"] .as_array() .is_some_and(|operations| { @@ -521,55 +567,71 @@ async fn apply( withdrawing: bool, ) -> Result { let owner = format!("st/fleet/{fleet_id}"); - let grants = grants(fleet_id); + let grants = accepted_grants(fleet_id); let mut journal = load_journal(state_dir).await?; + if let Some(pending) = &journal.pending { + ensure!( + pending.request["owner"] == owner + || pending.request["owner"] == format!("{RENAMED_OWNER_PREFIX}{fleet_id}"), + "pending Fabric batch belongs to another fleet" + ); + } resolve_pending(state_dir, socket, &mut journal, &desired).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); - ensure!( - operations.len() <= 128, - "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", - operations.len() - ); - if !operations.is_empty() { - for operation in &operations { - if operation["op"] == "remove_peer" { - journal.retired.insert( - operation["node_id"] - .as_str() - .context("removal has no NodeID")? - .to_owned(), - ); - } - } - let revision = current.policy_revision; - let digest = Sha256::digest(serde_json::to_vec(&json!([ - &owner, - ¤t.instance, - revision, - &operations - ]))?); - journal.pending = Some(Pending { - request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)), - "expected_revision":revision, "operations":operations}), - created_unix_ms: now_ms(), - }); - // 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) - .await? - .expect("pending transaction"); - journal_changed = false; - current = snapshot(socket).await?; + for owner in [owner, format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] { + let retained; + let policy = if owner.starts_with("st/") { + &desired + } else { + retained = retained_policy(¤t, &owner, &desired); + &retained + }; + let operations = plan(¤t, &owner, policy); ensure!( - current.policy_revision >= mutation.current_revision, - "Fabric policy revision regressed after mutation" + operations.len() <= 128, + "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", + operations.len() ); + if !operations.is_empty() { + for operation in &operations { + if operation["op"] == "remove_peer" { + journal.retired.insert( + operation["node_id"] + .as_str() + .context("removal has no NodeID")? + .to_owned(), + ); + } + } + let revision = current.policy_revision; + let digest = Sha256::digest(serde_json::to_vec(&json!([ + &owner, + ¤t.instance, + revision, + &operations + ]))?); + journal.pending = Some(Pending { + request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)), + "expected_revision":revision, "operations":operations}), + created_unix_ms: now_ms(), + }); + // 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, policy) + .await? + .expect("pending transaction"); + journal_changed = false; + current = snapshot(socket).await?; + ensure!( + current.policy_revision >= mutation.current_revision, + "Fabric policy revision regressed after mutation" + ); + } } let retired_count = journal.retired.len(); journal.retired.retain(|node| { @@ -590,7 +652,7 @@ async fn apply( || journal.retired.contains(&peer.node_id) || desired .get(&peer.node_id) - .is_some_and(|expected| expected.len() < grants.len()) + .is_some_and(|expected| expected.len() < 2) }) .collect::>(); let message = residual(&relevant, &desired, &grants); diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index f62969fd746..212197ac365 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -4,6 +4,65 @@ use tokio::net::UnixListener; const FLEET: &str = "test-fleet"; const OWNER: &str = "st/fleet/test-fleet"; +#[tokio::test] +async fn renamed_owner_revocation_and_residuals_never_grant_pty() { + let root = tempfile::tempdir().unwrap(); + let renamed = "fractal/fleet/test-fleet"; + let removed = vec![peer("gone", OWNER, &["st3-client-v0"]), peer("gone", renamed, &["fractal-client-v0"])]; + let (socket, server) = serve(root.path(), vec![ + Some(list(removed, 7)), + Some(mutation(vec![], "complete")), + Some(list(vec![peer("gone", renamed, &["fractal-client-v0"])], 8)), + Some(mutation(vec![], "complete")), + Some(list(vec![peer("gone", "static", &["fractal-client-v0"])], 9)), + ]); + assert_eq!(withdraw(root.path(), &socket, FLEET).await.unwrap().state, "residual_access"); + let requests = server.await.unwrap(); + for (index, owner) in [(1, OWNER), (3, renamed)] { + assert_eq!(requests[index]["request"]["owner"], owner); + assert_eq!( + requests[index]["request"]["operations"], + json!([{"op":"remove_peer","node_id":"gone"}]) + ); + } +} + +#[test] +fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { + let renamed = "fractal/fleet/test-fleet"; + let services = ["fractal/fleet/test-fleet", "fractal-client-v0", "fractal/pty/test-fleet"]; + let records = vec![peer("peer-node", renamed, &services), peer("old-node", renamed, &["st3-client-v0"])]; + let snapshot: Snapshot = serde_json::from_value(list(records, 7)["reply"]["snapshot"].clone()).unwrap(); + let desired = DesiredPolicy::from([( + "peer-node".into(), + BTreeSet::from(["st3/fleet/test-fleet".into()]), + )]); + let retained = retained_policy(&snapshot, renamed, &desired); + let operations = plan(&snapshot, renamed, &retained); + assert_eq!( + operations[0], + json!({"op":"remove_peer","node_id":"old-node"}) + ); + assert_eq!(operations[1], json!({"op":"remove_peer","node_id":"peer-node"})); + let mut pending = uncertain_add(); + pending.request["owner"] = json!(renamed); + pending.request["operations"][1]["grants"] = json!(["fractal/fleet/test-fleet"]); + assert!(!pending_matches_desired(&pending, &desired)); + pending.request["operations"][1]["grants"] = json!(["fractal/pty/test-fleet"]); + assert!(!pending_matches_desired(&pending, &desired)); +} + +#[tokio::test] +async fn renamed_pending_addition_is_observed_never_sent() { + let root = tempfile::tempdir().unwrap(); + let mut pending = uncertain_add(); + pending.request["owner"] = json!(format!("{RENAMED_OWNER_PREFIX}{FLEET}")); + let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]); + complete_pending(&socket, &pending, pending_matches_desired(&pending, + &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]))).await.unwrap(); + assert_eq!(server.await.unwrap()[0]["request"]["operation"], "cancellation"); +} + fn member(name: &str, key: &str, node: &str, state: &str) -> MemberView { MemberView { name: name.into(), diff --git a/release-notes/fabric-adapter-dual-owner-readers.json b/release-notes/fabric-adapter-dual-owner-readers.json new file mode 100644 index 00000000000..0635969ef17 --- /dev/null +++ b/release-notes/fabric-adapter-dual-owner-readers.json @@ -0,0 +1,10 @@ +{ + "version": 1, "summary": "Recognize both exact fleet adapter owners during recovery and revocation without switching policy producers.", + "replay": {"status": "none", "detail": "Pending requests retain their exact owner and request identity."}, + "schema": {"status": "none", "detail": "No database schema change."}, + "rules": {"status": "none", "detail": "No claim or checkpoint change."}, + "compatibility": {"status": "changed", "detail": "Adapter readers recognize st/fleet and fractal/fleet owners for this fleet, including retained pairing journals. Renamed-owner authority can only be retained or reduced; new policy and permits still use the legacy owner. Both protocol families participate in residual-access detection. PTY access remains operator-only."}, + "downtime": {"status": "none", "detail": "No host or service changes."}, + "manual": {"status": "none", "detail": "No migration or state rewrite required."}, + "recovery": {"status": "none", "detail": "Missing journal owner decodes as the original legacy owner."} +}