Fabric re-key R3: CLI

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,

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<Option<Progress>> {
+    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<Value> {
+    // 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<String> {
+    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<bool> {
+    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::<fractal_daemon::fleet::FleetView>("/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(())
+}
+

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"])]

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`,

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