diff --git a/crates/fractal/src/fleet_cli.rs b/crates/fractal/src/fleet_cli.rs index 8bbf63fa3ff..11c64307ed3 100644 --- a/crates/fractal/src/fleet_cli.rs +++ b/crates/fractal/src/fleet_cli.rs @@ -1,6 +1,7 @@ use super::*; // The same print macros as main.rs: CLI output goes through st3's writer. use fractal_daemon::{cli_print as print, cli_println as println}; +mod rekey; // Some fleet commands historically print prose even with --json. Keep those bytes // stable for callers; only the human rendering uses the installed fractal command. @@ -534,6 +535,7 @@ pub(super) async fn run_fleet(endpoint: &Endpoint, command: FleetCommand, json_o FleetTransportsCommand::Add { no_service, .. } | FleetTransportsCommand::Remove { no_service, .. } | FleetTransportsCommand::Prefer { no_service, .. } => !*no_service, + FleetTransportsCommand::Rekey { .. } => false, }, _ => false, }; @@ -1151,6 +1153,9 @@ async fn run_fleet_transports( json_output: bool, ) -> Result<()> { use fractal_daemon::fleet::transport::{Fabric, default_fabric_protocol, resolve_tool}; + if matches!(command, FleetTransportsCommand::Rekey { .. }) { + return rekey::run(client, config, json_output).await; + } let lock = lock_fleet_settings(&config.state_dir).await?; let mut file = fractal_daemon::config::FleetFile::load(&config.state_dir)? .context("this machine is not a fleet member")?; @@ -1161,7 +1166,8 @@ async fn run_fleet_transports( let transport = match &command { FleetTransportsCommand::Add { transport, .. } | FleetTransportsCommand::Remove { transport, .. } - | FleetTransportsCommand::Prefer { transport, .. } => transport, + | FleetTransportsCommand::Prefer { transport, .. } + | FleetTransportsCommand::Rekey { transport } => transport, }; let enabled = file .transports @@ -1209,6 +1215,7 @@ async fn run_fleet_transports( .clone() .unwrap_or_else(|| default_fabric_protocol(&file.fleet_id)); let no_service = match command { + FleetTransportsCommand::Rekey { .. } => unreachable!("rekey dispatched before settings edits"), FleetTransportsCommand::Prefer { transport, no_service, diff --git a/crates/fractal/src/fleet_cli/rekey.rs b/crates/fractal/src/fleet_cli/rekey.rs new file mode 100644 index 00000000000..38717e68d76 --- /dev/null +++ b/crates/fractal/src/fleet_cli/rekey.rs @@ -0,0 +1,197 @@ +//! Prepare and publish through the old route before guarded Fabric activation. +use super::*; +use fractal_daemon::cli_println as println; +use serde::{Deserialize, Serialize}; +use std::path::Path; + +const OVERLAP_SECONDS: u64 = 600; +const COMMAND_TIMEOUT: Duration = Duration::from_secs(90); + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum Phase { + Prepared, + Committed, + Activating, + Complete, + Aborted, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Progress { + fleet_id: String, + member_key: String, + old_node: String, + next_node: String, + deadline: u64, + phase: Phase, +} + +fn now() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +fn save(path: &Path, progress: &Progress) -> Result<()> { + use std::{io::Write, os::unix::fs::OpenOptionsExt}; + let parent = path.parent().context("rekey journal has no parent")?; + let temporary = path.with_extension("json.tmp"); + let mut file = std::fs::OpenOptions::new() + .write(true).create(true).truncate(true).mode(0o600).open(&temporary)?; + file.write_all(&serde_json::to_vec(progress)?)?; + file.sync_all()?; + std::fs::rename(temporary, path)?; + std::fs::File::open(parent)?.sync_all()?; + Ok(()) +} + +fn load(path: &Path) -> Result> { + match std::fs::read(path) { + Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes).context("read Fabric rekey journal")?)), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error.into()), + } +} + +async fn fabric_json(program: &Path, args: &[&str]) -> Result { + // kill_on_drop bounds a hung subprocess without touching the supervised daemon. + let output = tokio::time::timeout(COMMAND_TIMEOUT, + tokio::process::Command::new(program).args(args).kill_on_drop(true).output()) + .await.context("Fabric identity command timed out; retry the same rekey command")??; + anyhow::ensure!(output.status.success(), "Fabric identity command failed: {}{}", + String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + serde_json::from_slice(&output.stdout).context("Fabric identity command returned malformed JSON") +} + +fn identity(value: &Value, field: &str) -> Result { + let node = value[field].as_str().context("Fabric identity response has no NodeID")?; + anyhow::ensure!(node.len() == 64 && node.bytes().all(|byte| byte.is_ascii_hexdigit()), + "Fabric identity response has an invalid NodeID"); + Ok(node.to_owned()) +} + +async fn publish(client: &Client, progress: &Progress, action: &str) -> Result<()> { + let _: Value = client.post("/v1/internal/fleet/fabric-rekey", &json!({ + "action":action, "old_node":progress.old_node, "next_node":progress.next_node, + "deadline":progress.deadline, + })).await?; + Ok(()) +} + +fn confirmed(view: &fractal_daemon::fleet::FleetView, progress: &Progress, state: &str) -> bool { + view.members.iter().any(|member| { + member.state == "current" && member.member_key != progress.member_key + && member.endpoints.iter().any(|endpoint| { + endpoint["transport"] == "fabric-rekey-receipt" + && endpoint["member_key"] == progress.member_key + && endpoint["old_node"] == progress.old_node + && endpoint["next_node"] == progress.next_node + && endpoint["deadline"] == progress.deadline + && endpoint["state"] == state + }) + }) +} + +async fn wait_for_receipt(client: &Client, progress: &Progress, state: &str) -> Result { + loop { + let remaining = progress.deadline.saturating_sub(now()); + if remaining == 0 { return Ok(false); } + // A slow local API must not extend the authorization window. + let view = tokio::time::timeout(Duration::from_secs(remaining), + client.get::("/v1/internal/fleet/membership")).await; + match view { + Ok(Ok(view)) if confirmed(&view, progress, state) && now() < progress.deadline => return Ok(true), + Ok(Ok(_)) => {}, + Ok(Err(error)) => return Err(error.context("read signed adapter receipts; retry rekey to resume")), + Err(_) => return Ok(false), + } + tokio::time::sleep(Duration::from_secs(1).min(Duration::from_secs(progress.deadline.saturating_sub(now())))).await; + } +} + +async fn abort(client: &Client, path: &Path, progress: &mut Progress) -> Result<()> { + // Publish before marking terminal: a lost reply is retried, never reported as a rollback. + publish(client, progress, "abort").await.context("could not publish rekey rollback; old Fabric identity is unchanged; retry rekey")?; + progress.phase = Phase::Aborted; + save(path, progress)?; + anyhow::bail!("no peer confirmed Fabric rekey before the ten-minute deadline; aborted, old identity remains active") +} + +pub(super) async fn run(client: &Client, config: &Config, json_output: bool) -> Result<()> { + use fractal_daemon::fleet::transport::resolve_tool; + let _lock = lock_fleet_settings(&config.state_dir).await?; + let file = fractal_daemon::config::FleetFile::load(&config.state_dir)? + .context("this machine is not a fleet member")?; + anyhow::ensure!(file.transports.iter().any(|transport| transport == "fabric"), + "this machine is not a Fabric fleet member"); + let view: fractal_daemon::fleet::FleetView = client.get("/v1/internal/fleet/membership").await?; + let member = view.members.iter().find(|member| member.name == config.node && member.state == "current") + .context("this machine is not a current fleet member")?; + anyhow::ensure!(member.endpoints.iter().any(|endpoint| endpoint["transport"] == "fabric"), + "this member has no published Fabric endpoint"); + let program = resolve_tool(file.fabric.as_deref(), "fabric").context("fabric is not installed")?; + let status = fabric_json(&program, &["identity", "status", "--json"]).await + .context("Fabric is too old or unavailable: install the Fabric v0.9.4 identity lifecycle API")?; + let current = identity(&status, "current")?; + let path = config.state_dir.join("fleet/fabric-rekey.json"); + let saved = load(&path)?; + let mut progress = match saved { + Some(progress) if !matches!(progress.phase, Phase::Complete | Phase::Aborted) => { + anyhow::ensure!(progress.fleet_id == file.fleet_id && progress.member_key == member.member_key, + "Fabric rekey journal belongs to another fleet member"); + anyhow::ensure!(current == progress.old_node || current == progress.next_node, + "Fabric identity changed outside this rekey; refusing to overwrite durable progress"); + progress + }, + _ => { + anyhow::ensure!(member.endpoints.iter().any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == current), + "published Fabric identity does not match the local identity"); + let prepared = fabric_json(&program, &["identity", "prepare", "--json"]).await?; + let old_node = identity(&prepared, "current")?; + let next_node = identity(&prepared, "next")?; + anyhow::ensure!(old_node == current && old_node != next_node, "Fabric prepared an unexpected identity"); + let progress = Progress { fleet_id:file.fleet_id.clone(), member_key:member.member_key.clone(), + old_node, next_node, deadline:now() + OVERLAP_SECONDS, phase:Phase::Prepared }; + save(&path, &progress)?; + progress + }, + }; + if current == progress.next_node { + anyhow::ensure!(matches!(progress.phase, Phase::Activating), "new identity active without durable activation decision"); + } else if progress.phase == Phase::Prepared { + if now() >= progress.deadline { return abort(client, &path, &mut progress).await; } + publish(client, &progress, "prepare").await?; + if !wait_for_receipt(client, &progress, "prepared").await? { + return abort(client, &path, &mut progress).await; + } + // Save intent before publication; retrying commit after a lost reply is safe. + progress.phase = Phase::Committed; + save(&path, &progress)?; + } + if progress.phase == Phase::Committed { + if now() >= progress.deadline { return abort(client, &path, &mut progress).await; } + publish(client, &progress, "commit").await?; + if !wait_for_receipt(client, &progress, "committed").await? { + return abort(client, &path, &mut progress).await; + } + // From here the next identity is permanently authorized by a peer. Never + // roll back on an uncertain activate reply: Fabric's guards reconcile it. + progress.phase = Phase::Activating; + save(&path, &progress)?; + } + let activated = fabric_json(&program, &["identity", "activate", "--expect-current", &progress.old_node, + "--expect-next", &progress.next_node, "--json"]).await + .context("Fabric activation incomplete; rerun fleet transports rekey fabric to resume the guarded activation")?; + anyhow::ensure!(identity(&activated, "active")? == progress.next_node, "Fabric activated an unexpected identity"); + progress.phase = Phase::Complete; + save(&path, &progress)?; + if json_output { + println!("{}", json!({"state":"complete","previous":progress.old_node,"active":progress.next_node,"deadline":progress.deadline})); + } else { + println!("Fabric re-keyed in place: {} -> {}. Offline peers recover when signed membership converges.", progress.old_node, progress.next_node); + } + Ok(()) +} + diff --git a/crates/fractal/src/main.rs b/crates/fractal/src/main.rs index 64f1c6dce4b..7f589988791 100644 --- a/crates/fractal/src/main.rs +++ b/crates/fractal/src/main.rs @@ -1084,6 +1084,11 @@ struct FleetModeArgs { #[derive(Subcommand)] enum FleetTransportsCommand { + /// Re-key this member in place after peers install its prepared Fabric identity. + Rekey { + #[arg(value_parser = ["fabric"])] + transport: String, + }, /// Expose Fabric, then announce it. Requires a listening member with a port. Add { #[arg(value_parser = ["fabric"])] diff --git a/crates/fractal/src/main_tests.rs b/crates/fractal/src/main_tests.rs index fa22af05b31..77705724f82 100644 --- a/crates/fractal/src/main_tests.rs +++ b/crates/fractal/src/main_tests.rs @@ -1,5 +1,17 @@ use super::*; +#[test] +fn fleet_rekey_accepts_only_an_explicit_fabric_transport() { + assert!(Cli::try_parse_from(["fractal", "fleet", "transports", "rekey", "fabric"]).is_ok()); + for transport in ["tcp", "iroh", ""] { + let mut args = vec!["fractal", "fleet", "transports", "rekey"]; + if !transport.is_empty() { + args.push(transport); + } + assert!(Cli::try_parse_from(args).is_err(), "{transport}"); + } +} + #[test] fn fleet_admit_without_binds_mutating_actor_to_harness() { for actor in ["agent/own", "agent/peer", "person/operator"] { diff --git a/docs/fleet-join.md b/docs/fleet-join.md index 79dced68509..6d315c78c85 100644 --- a/docs/fleet-join.md +++ b/docs/fleet-join.md @@ -286,6 +286,46 @@ The worker exposes its loopback listener with the exposure. The adapter maintains trust and grants; Fractal does not edit `peers.toml`. +#### Re-key an existing Fabric member + +Run this on the machine whose Fabric identity must change: + +```sh +fractal fleet transports rekey fabric +``` + +Use Fabric's identity lifecycle and runtime expiry APIs (v0.9.4), a running +Fractal daemon and replication worker, and a running `fractal-fabric-adapter` on +the other members. At least one other current member must confirm both policy +phases before the original ten-minute deadline. A single-member or disconnected +fleet cannot complete a re-key. + +The command asks Fabric to prepare a key without changing its live identity. +Fractal publishes a signed rotation announcement under the same member key over +the old route. Peers temporarily grant the prepared NodeID until the immutable +deadline, keeping the old NodeID permanent. Their adapters publish signed receipts +only after checking the installed Fabric policy. After one receipt, Fractal +publishes the committed endpoint; peers make the new NodeID permanent and expire +the old NodeID at the original deadline. Fractal waits for a committed receipt +before asking Fabric to activate the prepared identity. Fabric never serves two +local identities at once. Adapter policy changes are revision-checked atomic +batches and do not grant terminal access. + +The command persists progress in `fleet/fabric-rekey.json` under the Fractal state +directory. After an interruption, run the same command: it resumes the original +deadline, or retries the same guarded activation after the activation decision. +Do not delete the journal, edit Fabric's private keys, or run `fabric identity +activate` separately. An uncertain activation response is not a rollback. + +If no peer confirms before the deadline, Fractal publishes an abort and leaves +the old local identity active. A stopped adapter cannot extend staged access: +Fabric enforces the installed expiry even while the adapter is down. Abort +withdrawal and restoration still require claim delivery and reconciliation. +An offline peer may miss the cutover and must learn current signed membership +through another reachable fleet member before it can authorize the new NodeID. +Static and other-owner rights are not revoked: remove them at their source if +complete retirement of the old NodeID is required. + #### Adapter helper boundary The CLI discovers `fractal-fabric-adapter` beside its executable or on `PATH`, diff --git a/docs/st3/action-coverage.json b/docs/st3/action-coverage.json index fa636ba1f2e..8f6b734a487 100644 --- a/docs/st3/action-coverage.json +++ b/docs/st3/action-coverage.json @@ -1315,6 +1315,15 @@ ], "verification": "real CLI over isolated Unix daemons with a Fabric shim; both reorder directions, disabled-carrier refusal, exposure preservation, settings-lock contention, atomic save and no-op" }, + { + "command": "st fleet transports rekey", + "description": "Re-key this member in place after peers install its prepared Fabric identity", + "tests": [ + "crates/fractal/src/main_tests.rs::fleet_rekey_accepts_only_an_explicit_fabric_transport", + "crates/fractal-api-core/src/api/fleet.rs::signed_rekey_route_commits_preserves_refresh_and_rolls_back" + ], + "verification": "CLI parser accepts only explicit Fabric; the real API route publishes signed transitions, preserves worker refreshes and rolls back" + }, { "command": "st fleet transports remove", "description": "Withdraw Fabric from this fleet member while keeping a usable carrier", diff --git a/docs/st3/actor-contract.json b/docs/st3/actor-contract.json index 4c55ab71a0e..7591b8dc2a5 100644 --- a/docs/st3/actor-contract.json +++ b/docs/st3/actor-contract.json @@ -1011,6 +1011,15 @@ "actor_environment": null, "ignores_actor": false }, + "st fleet transports rekey": { + "actor": "none", + "accepts_person": false, + "can_carry_person": false, + "as_role": "none", + "alternate_actor_option": null, + "actor_environment": null, + "ignores_actor": false + }, "st fleet transports remove": { "actor": "none", "accepts_person": false, diff --git a/release-notes/fabric-rekey-command.json b/release-notes/fabric-rekey-command.json new file mode 100644 index 00000000000..77b8d8702c9 --- /dev/null +++ b/release-notes/fabric-rekey-command.json @@ -0,0 +1 @@ +{"version": 1,"summary": "Rotate Fabric identity with signed peer confirmation before guarded local activation","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": "changed","detail": "Run fractal fleet transports rekey fabric with Fabric v0.9.4 and a reachable current fleet peer."},"recovery": {"status": "changed","detail": "The local rekey journal resumes lost replies without extending the original ten-minute deadline."}}