diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c3ce836..96f99959 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,20 @@ Deprecated, Fixed or Security. ### Added +- `fabric identity prepare|activate|status [--json]` supports in-place consumer + re-keying without dual-identity serving. Preparation durably stores one private + pending key; guarded activation uses a single rename and waits for the + service-aware restart to report the new NodeID. Lost-reply retries return + `already_active`; `identity_busy`, `identity_mismatch`, and `restart_failed` + distinguish contention, stale guards, and retry-only-restart recovery. + Peer policy, exposures, receipts/revocation history and audit remain local; + old transport sessions, connections and dial declarations are retired. + Sync keeps its history but changes its NodeID-derived author. Retired secret + keys are deleted after the conservative ten-minute overlap upper bound. + Foreground daemons participate in the bounded stop/restart handshake; failed + pre-rename swaps best-effort restore the current daemon and discard stale + handoffs. Live runtime-policy mode overrides stale markers. Cleanup failures + do not block inspection, and preparation replaces corrupt pending keys. - **Provisional:** runtime peer records support immutable `expires_at` deadlines within the next 600 seconds. Existing atomic batches can install a prepared next NodeID's exact grants and schedule the old owner record's retirement. diff --git a/README.md b/README.md index f8af33c0..0bfc5330 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,70 @@ still holds to answer one echo within three seconds and resets only a connection that does not, at most once a minute per peer; the endpoint itself is not rebuilt for that. +### Re-key this machine in place + +Identity rotation is **prepare → publish over the old identity → activate**. +Fabric never serves two local identities at once. + +```sh +fabric identity prepare --json +# {"current":"","next":""} +``` + +Preparation durably stores a private 0600 key inside Fabric's home. Repeating it +returns the same `next`; the running daemon is unchanged. Over the still-working +old connection, arrange for every peer to trust `next` with the exact existing +grants and set the old runtime record's immutable expiry (at most ten minutes) +in one [atomic runtime-peer batch](docs/runtime-peer-api.md). +Only after peers have installed that batch: + +```sh +fabric identity activate --expect-current --expect-next --json +# {"previous":"","active":"","outcome":"activated"} +fabric identity status --json +# {"current":"","next":null,"state":"active"} +``` + +Both foreground and background CLI daemons retain the same per-home lifetime +lease through endpoint and child teardown; a stale socket alone is not liveness. + +Activation fences exposure changes and stops the old daemon before the single +atomic rename, then asks the existing native supervisor to restart or directly +spawns a standalone replacement daemon. Success waits for the live daemon to +report the new NodeID. If activation fails after requesting shutdown but before +rename, it best-effort restarts the current identity and reports whether recovery +succeeded. The identity remains prepared. Lost-reply retries with the same guards +return `already_active`; they never swap again, and `previous` echoes the caller's +`--expect-current`, not a recovered historical value. A mismatched guard returns +`identity_mismatch` without swapping the identity. The identity lock remains held +through restart: concurrent `status`, `prepare`, or `activate` returns +`identity_busy` rather than waiting. + +After a crash, `status` reports either `prepared` (old key plus pending new key) +or `active` (the rename completed). `restart_failed` is non-zero and includes +`previous` and `active` in the JSON error. Repeat the same activate command to +retry only daemon reconciliation; do not restore the old key. The old daemon +has already stopped before a post-rename restart can fail. + +Static and runtime peer policy, exposure declarations, retained transaction and +revocation receipts, and the audit log remain in the same home. Tunnel sessions, +mux connections, queued dials, and persisted dial declarations bound to the old +local NodeID are retired; consumers explicitly establish new dials. Persisted and +ephemeral exposures are carried across this restart without making ephemeral ones +permanent: replacement startup consumes the one-start handoff even if the +activating CLI crashes. An aborted pre-rename activation removes its handoff and +clears a reachable old daemon's exposure fence. If the CLI itself dies after the +snapshot, exposure changes return `identity_busy` until the same guarded +activation is retried or the daemon is restarted. Sync author identity is the +NodeID byte-for-byte: the companion stops its old-author engine and starts a +new-author engine over the existing sync history and files. + +The old private key is retained only in a non-serving retired file. Because this +machine cannot observe other machines' expiry records, it uses the conservative +upper bound: ten minutes after activation. The daemon removes expired retired +keys while running; identity commands also perform cleanup, and an offline home +is cleaned on its next daemon start. Retired keys are never loaded for serving. + ### Presence, partitions, and local ownership Fabric reports connection facts about each trusted peer by canonical NodeID. diff --git a/crates/fabric-config/src/error.rs b/crates/fabric-config/src/error.rs index 108e21be..57d68a3b 100644 --- a/crates/fabric-config/src/error.rs +++ b/crates/fabric-config/src/error.rs @@ -119,6 +119,9 @@ error_codes! { PairingExchangeInactive => "pairing_exchange_inactive", Refusal; RequestIdConflict => "request_id_conflict", Usage; CancellationTimeout => "cancellation_timeout", Fault; + IdentityMismatch => "identity_mismatch", Usage; + IdentityBusy => "identity_busy", Fault; + RestartFailed => "restart_failed", Fault; InvalidArgument => "invalid_argument", Usage; } @@ -166,6 +169,7 @@ pub struct FabricError { message: String, peer: Option, service: Option, + identity: Option<(String, String)>, } impl FabricError { @@ -175,6 +179,7 @@ impl FabricError { message: message.into(), peer: None, service: None, + identity: None, } } @@ -190,6 +195,11 @@ impl FabricError { self } + pub fn identity(mut self, previous: impl Into, active: impl Into) -> Self { + self.identity = Some((previous.into(), active.into())); + self + } + pub fn code(&self) -> ErrorCode { self.code } @@ -355,7 +365,7 @@ pub fn render(error: &anyhow::Error, json: bool) -> Report { let found = find(error); let message = without_tokens(&format!("{error:#}")); let stdout = json.then(|| { - json!({ + let mut object = json!({ "error": { "code": code.as_str(), "class": code.class().as_str(), @@ -363,8 +373,12 @@ pub fn render(error: &anyhow::Error, json: bool) -> Report { "peer": found.and_then(FabricError::peer_name), "service": found.and_then(FabricError::service_name), } - }) - .to_string() + }); + if let Some((previous, active)) = found.and_then(|error| error.identity.as_ref()) { + object["error"]["previous"] = previous.clone().into(); + object["error"]["active"] = active.clone().into(); + } + object.to_string() }); Report { stderr: format!("Error: {}", code.reason(&message)), diff --git a/crates/fabric-sync/tests/sync_process.rs b/crates/fabric-sync/tests/sync_process.rs index 08b60595..cb3b5d9a 100644 --- a/crates/fabric-sync/tests/sync_process.rs +++ b/crates/fabric-sync/tests/sync_process.rs @@ -442,12 +442,29 @@ async fn an_incompatible_companion_is_refused_and_named() -> Result<()> { /// engine that kept running keeps its state. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_running_companion_reattaches_after_the_daemon_restarts() -> Result<()> { + companion_restarts_with_identity(false).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn identity_activation_changes_sync_author_and_preserves_sync_history() -> Result<()> { + companion_restarts_with_identity(true).await +} + +async fn companion_restarts_with_identity(rotate: bool) -> Result<()> { let _guard = SYNC_PROCESS_LOCK.lock().await; let mut a = Node::start(Side::New, "bus").await?; let b = Node::start(Side::New, "bus").await?; pair(&a, &b).await?; std::fs::write(a.folder.join("before.txt"), b"before the restart")?; assert!(wait_for_file(&b.folder.join("before.txt"), b"before the restart").await); + let previous = a.id(); + let prepared = if rotate { Some(fabric::identity::prepare(&a.home)?) } else { None }; + if let Some(prepared) = &prepared { + let mut peers = PeerBook::load(&b.home)?; + peers.add_with_allow(prepared.next, Some("a-next".into()), None, Some(vec!["sync".into()])); + peers.save(&b.home)?; + b.reload_peers().await?; + } // The shape of every update: the daemon goes down and comes back while the // companion keeps running. @@ -465,6 +482,7 @@ async fn a_running_companion_reattaches_after_the_daemon_restarts() -> Result<() .wait_for(Duration::from_secs(30), |phase| !phase.is_active()) .await .context("the companion never noticed the daemon leaving")?; + let activation = prepared.as_ref().map(|prepared| fabric::identity::activate(&a.home, prepared.current, prepared.next)).transpose()?; let placeholder = std::mem::replace( &mut a.daemon, Daemon::New(FabricNode::start(a.home.clone()).await?), @@ -473,8 +491,19 @@ async fn a_running_companion_reattaches_after_the_daemon_restarts() -> Result<() Daemon::New(node) => node.shutdown().await?, Daemon::Old { .. } => unreachable!("the placeholder is this build"), } - // b must re-trust a's new address; the identity is the same. + if rotate { + assert_ne!(a.id(), previous); + assert_ne!( + fabric_sync::companion::author_from_node_id(&previous.to_string())?, + fabric_sync::companion::author_from_node_id(&a.id().to_string())?, + ); + let mut peers = PeerBook::load(&b.home)?; + peers.remove(&previous.to_string()); + peers.save(&b.home)?; + } + // Address hints and the new public key are published before syncing resumes. trust(&b, &a, "a").await?; + drop(activation); companion .wait_for(Duration::from_secs(30), |phase| phase.is_active()) diff --git a/docs/errors.md b/docs/errors.md index 545f2864..98bded0b 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -44,6 +44,9 @@ Each code has a class, and the class says what a caller should do: | `pairing_exchange_inactive` | refusal | 77 | The attempt is not pending and cannot exchange messages. No delivery occurred; reconcile its status rather than assuming an empty success. | | `request_id_conflict` | usage | 2 | A retained request ID was reused with different contents. | | `cancellation_timeout` | fault | 69 | Policy committed, but selected work did not finish within two seconds. The receipt lists remaining admissions and sessions; policy is not rolled back. | +| `identity_mismatch` | usage | 2 | Active or prepared identity does not match the activation guards; nothing changed. | +| `identity_busy` | fault | 69 | An identity operation holds the lock (including through restart), or exposure changes are fenced by an identity handoff. Retry the same guarded activation or restart the daemon to recover an abandoned handoff. | +| `restart_failed` | fault | 69 | Identity changed, but the replacement daemon did not become ready. Retry the same guarded activation; never roll back the key. | | `invalid_argument` | usage | 2 | Fabric rejected an argument's value. | @@ -109,6 +112,17 @@ object, and exits with the status above: - `message` is the text for people. - `peer` is the peer as the caller named it, and `service` the service, or `null` when the failure concerns no single one. Both keys are always present. +- An identity activation's `restart_failed` additionally carries `previous` and + `active` NodeIDs. The rename has happened; retry the same guarded activation + to reconcile the daemon, not to replace or roll back the key again. +- Identity control requests and native service commands have finite timeouts, + reported as `timeout` (or `restart_failed` after the rename). A pre-rename + failure after shutdown triggers a best-effort restart of the current identity; + its error says whether that restart succeeded. The key remains prepared. +- A pre-rename abort removes `identity.exposures` and clears the exposure fence + when the old daemon is reachable. If the CLI crashes after the snapshot or + cannot reach that daemon, repeat the same guarded activation or restart it: + `expose`/`unexpose` otherwise continue returning `identity_busy`. - Nothing else is printed on stdout for a failure. Stderr still carries the human line; do not parse it when you have the JSON. diff --git a/docs/json-output.md b/docs/json-output.md index 02cd62e8..86d4c415 100644 --- a/docs/json-output.md +++ b/docs/json-output.md @@ -17,6 +17,9 @@ a new key may be added, so read objects by key and ignore keys you do not know. | Command | Object | | --- | --- | | `id` | `{"id": ""}` | +| `identity prepare` | `{"current": "", "next": ""}` | +| `identity activate --expect-current OLD --expect-next NEW` | `{"previous": "", "active": "", "outcome": "activated"}`; lost-reply retries return `"already_active"`; `previous` always echoes OLD | +| `identity status` | `{"current": "", "next": "" (or null), "state": "prepared" (or "active")}` | | `addr` | the daemon's `EndpointAddr` object (`{"id": ..., "addrs": [...]}`), with or without `--json`; it was JSON before the flag existed | | `peers` | `{"peers": [{"node", "name" (or null), "allow": [...], "dial_only", "inbound_only", "roaming"}]}` | | `probe PEER PROTOCOL` | `{"peer", "peer_id", "protocol", "outcome", "round_trip_micros", "transport", "error", "code", "class"}`; `outcome` is `supported`, `unsupported`, `unreachable` or `timeout`, and the exit status is unchanged | diff --git a/src/config.rs b/src/config.rs index 0b00ea6e..a78fe290 100644 --- a/src/config.rs +++ b/src/config.rs @@ -70,8 +70,8 @@ pub fn load_or_create_identity(home: &FabricHome) -> Result { if path.exists() { let raw = fs::read_to_string(&path) .with_context(|| format!("failed to read {}", path.display()))?; - let file: IdentityFile = - toml::from_str(&raw).with_context(|| format!("failed to parse {}", path.display()))?; + let file: IdentityFile = toml::from_str(&raw) + .map_err(|_| anyhow::anyhow!("invalid identity key file {}", path.display()))?; return Ok(file.secret_key); } diff --git a/src/control.rs b/src/control.rs index c14155f4..c89c62da 100644 --- a/src/control.rs +++ b/src/control.rs @@ -20,6 +20,13 @@ pub enum ControlRequest { Status, ReachabilityStatus, ReloadPeers, + IdentitySnapshot { + current: iroh::EndpointId, + next: iroh::EndpointId, + }, + IdentityAbort { + current: iroh::EndpointId, + }, RuntimePeers { request: crate::runtime_peers::Request, }, @@ -179,6 +186,9 @@ pub struct DialRestoreFailure { #[derive(Debug, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum ControlResponse { + IdentitySnapshot { + runtime_peers: bool, + }, RuntimePeers { reply: crate::runtime_peers::Reply, }, diff --git a/src/daemon.rs b/src/daemon.rs index ae08c3e8..4beb5038 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -478,6 +478,7 @@ pub struct DaemonState { policy_quarantine: Arc>>, pending_policy_revocations: Arc>>>, exposures: RwLock, Exposure>>, + exposure_update: Mutex, dial_sockets: Mutex>, one_shot_dials: Mutex>>, dial_restore_failures: Mutex>, @@ -738,25 +739,33 @@ enum Exposure { path: PathBuf, /// Write the connecting peer's identity to the socket before its bytes. peer_header: bool, + persist: bool, }, Tcp { addr: String, + persist: bool, }, Exec { argv: Vec, limit: Arc, + persist: bool, }, } impl Exposure { + fn persisted(&self) -> bool { + match self { + Self::Socket { persist, .. } | Self::Tcp { persist, .. } | Self::Exec { persist, .. } => *persist, + } + } fn to_server_target(&self, peer: &PeerIdentity) -> tunnel::ServerTarget { match self { - Self::Socket { path, peer_header } => tunnel::ServerTarget::UnixSocket { + Self::Socket { path, peer_header, .. } => tunnel::ServerTarget::UnixSocket { path: path.clone(), header: peer_header.then(|| peer_header_line(peer)), }, - Self::Tcp { addr } => tunnel::ServerTarget::Tcp { addr: addr.clone() }, - Self::Exec { argv, limit } => tunnel::ServerTarget::Exec { + Self::Tcp { addr, .. } => tunnel::ServerTarget::Tcp { addr: addr.clone() }, + Self::Exec { argv, limit, .. } => tunnel::ServerTarget::Exec { argv: argv.clone(), limit: limit.clone(), }, @@ -769,7 +778,9 @@ fn load_persisted_exposures( services: &Services, ) -> Result, Exposure>> { let mut exposures = HashMap::new(); - for expose in FabricConfig::load(home)?.exposes() { + let config = FabricConfig::load(home)?; + let inherited = crate::identity::inherited_exposures(home)?; + for (expose, persist) in config.exposes().iter().map(|entry| (entry, true)).chain(inherited.iter().map(|entry| (entry, false))) { let alpn = validate_protocol(&expose.protocol)?; if matches_reserved_alpn(services, &alpn) { bail!( @@ -789,11 +800,12 @@ fn load_persisted_exposures( Exposure::Socket { path: socket.clone(), peer_header: *peer_header, + persist, } } PersistedExposeTarget::Tcp { addr } => { validate_tcp_addr(addr)?; - Exposure::Tcp { addr: addr.clone() } + Exposure::Tcp { addr: addr.clone(), persist } } PersistedExposeTarget::Exec { argv, max_children } => { if argv.is_empty() { @@ -805,6 +817,7 @@ fn load_persisted_exposures( Exposure::Exec { argv: argv.clone(), limit: tunnel::ExecLimit::new(*max_children), + persist, } } }; @@ -1234,6 +1247,7 @@ impl DaemonState { audit: Option, ) -> Result> { home.prepare()?; + crate::identity::reconcile_startup(&home)?; sweep_one_shot_dials(&home); let child_root = home.root().to_owned(); let child_owner = Arc::new( @@ -1311,6 +1325,7 @@ impl DaemonState { policy_quarantine: Arc::new(Mutex::new(None)), pending_policy_revocations: Arc::new(std::sync::Mutex::new(Vec::new())), exposures: RwLock::new(exposures), + exposure_update: Mutex::new(false), dial_sockets: Mutex::new(HashMap::new()), one_shot_dials: Mutex::new(Vec::new()), dial_restore_failures: Mutex::new(Vec::new()), @@ -1609,6 +1624,14 @@ impl DaemonState { PeerIdentity { id, name } } + async fn mutable_exposures(&self) -> Result> { + let guard = self.exposure_update.lock().await; + if *guard { + return Err(FabricError::new(ErrorCode::IdentityBusy, "exposure mutations are fenced by identity activation; retry the same guarded activation to recover, or restart this daemon").into()); + } + Ok(guard) + } + pub async fn expose(&self, protocol: &str, socket: PathBuf) -> Result<()> { self.expose_socket(protocol, socket, true, false).await } @@ -1624,6 +1647,7 @@ impl DaemonState { persist: bool, peer_header: bool, ) -> Result<()> { + let _identity_guard = self.mutable_exposures().await?; let alpn = validate_protocol(protocol)?; if matches_reserved_alpn(&self.services, &alpn) { bail!("{protocol:?} is reserved for fabric's built-in protocols"); @@ -1645,6 +1669,7 @@ impl DaemonState { Exposure::Socket { path: socket, peer_header, + persist, }, ); self.current_endpoint().set_alpns(accepted_alpns( @@ -1665,6 +1690,7 @@ impl DaemonState { addr: String, persist: bool, ) -> Result<()> { + let _identity_guard = self.mutable_exposures().await?; let alpn = validate_protocol(protocol)?; if matches_reserved_alpn(&self.services, &alpn) { bail!("{protocol:?} is reserved for fabric's built-in protocols"); @@ -1679,7 +1705,7 @@ impl DaemonState { } let mut exposures = self.exposures.write().await; - exposures.insert(alpn, Exposure::Tcp { addr }); + exposures.insert(alpn, Exposure::Tcp { addr, persist }); self.current_endpoint().set_alpns(accepted_alpns( &self.services, &exposures, @@ -1705,6 +1731,7 @@ impl DaemonState { max_children: usize, persist: bool, ) -> Result<()> { + let _identity_guard = self.mutable_exposures().await?; let alpn = validate_protocol(protocol)?; if matches_reserved_alpn(&self.services, &alpn) { bail!("{protocol:?} is reserved for fabric's built-in protocols"); @@ -1729,6 +1756,7 @@ impl DaemonState { Exposure::Exec { argv, limit: tunnel::ExecLimit::new(max_children), + persist, }, ); self.current_endpoint().set_alpns(accepted_alpns( @@ -1759,6 +1787,7 @@ impl DaemonState { } pub async fn unexpose(&self, protocol: &str) -> Result<()> { + let _identity_guard = self.mutable_exposures().await?; let alpn = validate_protocol(protocol)?; if matches_reserved_alpn(&self.services, &alpn) { bail!("{protocol:?} is reserved for fabric's built-in protocols"); @@ -3448,11 +3477,16 @@ pub async fn run_daemon_with_options(home: FabricHome, options: DaemonOptions) - result } -struct DaemonLease { +pub struct DaemonLease { _file: Arc, } pub fn daemon_lock_available(home: &FabricHome) -> Result { + Ok(try_daemon_lease(home)?.is_some()) +} + +/// Retain the returned file to exclude daemon startup until the identity swap. +pub fn try_daemon_lease(home: &FabricHome) -> Result> { let path = home.root().join("run/daemon.lock"); home.prepare()?; let file = OpenOptions::new() @@ -3464,10 +3498,10 @@ pub fn daemon_lock_available(home: &FabricHome) -> Result { { let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; if rc != 0 { - return Ok(false); + return Ok(None); } } - Ok(true) + Ok(Some(file)) } pub fn restart_down_decision(status_ok: bool, lease_available: bool) -> bool { @@ -3475,7 +3509,7 @@ pub fn restart_down_decision(status_ok: bool, lease_available: bool) -> bool { } impl DaemonLease { - fn acquire(home: &FabricHome) -> Result { + pub fn acquire(home: &FabricHome) -> Result { home.prepare()?; let path = home.root().join("run/daemon.lock"); let file = OpenOptions::new() @@ -3563,10 +3597,15 @@ async fn serve(state: Arc) -> Result<()> { .with_context(|| format!("failed to remove stale {}", sync_ipc_path.display()))?; } let sync_ipc_listener = IpcListener::bind(&sync_ipc_path)?; + // All fallible construction and listener binding precedes this serving-mode + // commit. A failed launch must not select stale runtime grants on recovery. + crate::identity::finish_startup_snapshot(&state.home)?; + crate::identity::record_runtime_mode(&state.home, state.runtime_peers.is_some())?; // Cancellation stops future sweeps but never drops an in-flight durable commit. // The existing policy drain owns its shutdown barrier. let _pairing_expiry = tokio::spawn(pairing::expire_loop(state.clone())); let _runtime_peer_expiry = tokio::spawn(runtime_control::runtime_peer_expiry_loop(state.clone())); + let _retired_keys = tokio::spawn(crate::identity::retired_key_cleanup(state.home.clone(), state.cancel.clone())); let result = tokio::select! { result = run_control_socket(control_listener, state.clone()) => result, @@ -4485,6 +4524,36 @@ async fn process_control_request( } => ControlResponse::Pairing { reply: pairing::remote(&state, gateway, addr, request).await, }, + ControlRequest::IdentitySnapshot { current, next } => { + if state.id() != current { + return Err(FabricError::new(ErrorCode::IdentityMismatch, "running daemon identity does not match activation guard").into()); + } + let mut handoff = state.exposure_update.lock().await; + let exposures = state.exposures.read().await; + let mut snapshot = Vec::with_capacity(exposures.len()); + for (alpn, exposure) in exposures.iter() { + if exposure.persisted() { continue; } + let protocol = String::from_utf8(alpn.clone())?; + snapshot.push(match exposure { + Exposure::Socket { path, peer_header, .. } => PersistedExpose::socket(protocol, path.clone(), *peer_header), + Exposure::Tcp { addr, .. } => PersistedExpose::tcp(protocol, addr.clone()), + Exposure::Exec { argv, limit, .. } => PersistedExpose::exec(protocol, argv.clone(), limit.max_children()), + }); + } + let home = state.home.clone(); + tokio::task::spawn_blocking(move || crate::identity::write_runtime_snapshot(&home, next, snapshot)).await??; + *handoff = true; + ControlResponse::IdentitySnapshot { runtime_peers: state.runtime_peers.is_some() } + } + ControlRequest::IdentityAbort { current } => { + if state.id() != current { + return Err(FabricError::new(ErrorCode::IdentityMismatch, "running daemon identity changed; cannot abort its handoff").into()); + } + let mut handoff = state.exposure_update.lock().await; + crate::identity::discard_runtime_snapshot(&state.home)?; + *handoff = false; + ControlResponse::Ok + } ControlRequest::Status => state.status_response().await?, ControlRequest::ReachabilityStatus => state.reachability_status_response().await?, ControlRequest::PeerEvents { diff --git a/src/identity.rs b/src/identity.rs new file mode 100644 index 00000000..f6453e57 --- /dev/null +++ b/src/identity.rs @@ -0,0 +1,605 @@ +//! Local identity rotation. Only identity.toml is authoritative for the active key. +use std::{fs::{self, File, OpenOptions}, io::Write, os::{fd::AsRawFd, unix::fs::OpenOptionsExt}, path::{Path, PathBuf}, time::{SystemTime, UNIX_EPOCH}}; +use anyhow::{Context, Result}; +use iroh::EndpointId; +use serde::Serialize; +use crate::{config::{FabricHome, generate_identity_file, load_or_create_identity}, error::{ErrorCode, FabricError}}; + +const RETENTION_SECONDS: u64 = 600; + +/// An advisory lock shared by every identity operation, including status. +pub struct IdentityLock(File); +impl IdentityLock { + pub fn acquire(home: &FabricHome) -> Result { + home.prepare()?; + let file = OpenOptions::new().read(true).write(true).create(true).truncate(false).mode(0o600).open(home.root().join("identity.lock"))?; + // SAFETY: flock borrows a live descriptor and does not retain pointers. + if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 { + let error = std::io::Error::last_os_error(); + if error.kind() == std::io::ErrorKind::WouldBlock { + return Err(FabricError::new(ErrorCode::IdentityBusy, "another identity operation is in progress").into()); + } + return Err(error.into()); + } + Ok(Self(file)) + } +} +impl Drop for IdentityLock { + fn drop(&mut self) { + // SAFETY: the descriptor remains live until after this call. + unsafe { libc::flock(self.0.as_raw_fd(), libc::LOCK_UN); } + } +} + +#[derive(Debug, Serialize)] +pub struct Status { + pub current: EndpointId, + pub next: Option, + pub state: &'static str, +} +#[derive(Debug, Serialize)] +pub struct Prepared { + pub current: EndpointId, + pub next: EndpointId, +} +#[derive(Debug, Serialize)] +pub struct Activated { + pub previous: EndpointId, + pub active: EndpointId, + pub outcome: &'static str, +} +fn pending(home: &FabricHome) -> PathBuf { home.root().join("identity.next.toml") } +fn sync_home(home: &FabricHome) -> Result<()> { File::open(home.root())?.sync_all().context("sync identity directory") } +fn read_id(path: &Path) -> Result { + #[derive(serde::Deserialize)] + struct Key { secret_key: iroh::SecretKey } + let key: Key = toml::from_str(&fs::read_to_string(path)?) + .map_err(|_| anyhow::anyhow!("invalid identity key file {}", path.display()))?; + Ok(key.secret_key.public()) +} +fn snapshot(home: &FabricHome) -> Result { + let current = read_id(&home.identity_path())?; + let next = match read_id(&pending(home)) { + Ok(id) => Some(id), + Err(error) if error.downcast_ref::().is_some_and(|e| e.kind() == std::io::ErrorKind::NotFound) => None, + Err(error) => { + tracing::warn!(%error, "ignoring unreadable pending identity; prepare replaces it"); + None + } + }; + Ok(Status { current, next, state: if next.is_some() { "prepared" } else { "active" } }) +} +pub fn status(home: &FabricHome) -> Result { + let _lock = IdentityLock::acquire(home)?; + load_or_create_identity(home)?; + if let Err(error) = cleanup_retired_locked(home, now()) { + tracing::warn!(%error, "retired identity cleanup failed"); + } + snapshot(home) +} +pub fn prepare(home: &FabricHome) -> Result { + let _lock = IdentityLock::acquire(home)?; + if let Err(error) = cleanup_retired_locked(home, now()) { + tracing::warn!(%error, "retired identity cleanup failed"); + } + load_or_create_identity(home)?; + File::open(home.identity_path())?.sync_all()?; + sync_home(home)?; + let state = snapshot(home)?; + if let Some(next) = state.next { return Ok(Prepared { current: state.current, next }); } + let temporary = scratch_path(home, "identity.next"); + let next = generate_identity_file(&temporary)?; + File::open(&temporary)?.sync_all()?; + fs::rename(&temporary, pending(home))?; + sync_home(home)?; + Ok(Prepared { current: state.current, next }) +} + +/// The lock remains held across the caller's restart barrier. A retry after the +/// rename never writes a key: it only asks the caller to reconcile the daemon. +pub struct Activation { + pub result: Activated, + _lock: IdentityLock, +} +struct Swap { + result: Activated, + durability_error: Option, +} +fn check_guards(home: &FabricHome, current: EndpointId, next: EndpointId) -> Result { + let active = read_id(&home.identity_path())?; + if active == next { return Ok(true); } + if active != current || current == next || read_id(&pending(home)).ok() != Some(next) { + return Err(FabricError::new(ErrorCode::IdentityMismatch, "identity does not match --expect-current and --expect-next; nothing changed").into()); + } + Ok(false) +} +pub fn activate(home: &FabricHome, expected_current: EndpointId, expected_next: EndpointId) -> Result { + let lock = IdentityLock::acquire(home)?; + let already = check_guards(home, expected_current, expected_next)?; + let swap = activate_locked(home, expected_current, expected_next, already, &lock)?; + Ok(Activation { result: swap.result, _lock: lock }) +} +fn activate_locked(home: &FabricHome, expected_current: EndpointId, expected_next: EndpointId, already: bool, _lock: &IdentityLock) -> Result { + if already { + return Ok(Swap { result: Activated { previous: expected_current, active: expected_next, outcome: "already_active" }, durability_error: sync_home(home).err() }); + } + // Retain the old secret before the only authoritative swap. The retired + // copy is never read as a serving key, even after a failed restart. + let retired = home.root().join(format!("identity.retired.{}.{}.toml", now().saturating_add(RETENTION_SECONDS), expected_current)); + match OpenOptions::new().write(true).create_new(true).mode(0o600).open(&retired) { + Ok(mut file) => { + file.write_all(&fs::read(home.identity_path())?)?; + file.sync_all()?; + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + anyhow::ensure!(read_id(&retired)? == expected_current, "retired identity does not match current key"); + } + Err(error) => return Err(error.into()), + } + // This is intent, not a second identity authority: startup acts on it only + // when identity.toml already contains expected_next. + write_private_atomic(home, "identity.activation", expected_next.to_string().as_bytes())?; + if check_guards(home, expected_current, expected_next)? { + return Err(FabricError::new(ErrorCode::IdentityMismatch, "active identity changed during preparation; activation aborted before rename").into()); + } + fs::rename(pending(home), home.identity_path())?; + // Once rename succeeded, the caller must retire the old daemon even when + // the directory durability barrier fails. + let durability_error = sync_home(home).err(); + Ok(Swap { result: Activated { previous: expected_current, active: expected_next, outcome: "activated" }, durability_error }) +} +fn now() -> u64 { SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() } +fn cleanup_retired_locked(home: &FabricHome, timestamp: u64) -> Result> { + let mut changed = false; + let mut next_deadline: Option = None; + for entry in fs::read_dir(home.root())? { + let entry = match entry { + Ok(entry) => entry, + Err(error) => { tracing::warn!(%error, "cannot inspect retired identity entry"); continue; } + }; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + let Some(rest) = name.strip_prefix("identity.retired.") else { continue }; + let Some((expiry, suffix)) = rest.split_once('.') else { continue }; + let Some(id) = suffix.strip_suffix(".toml") else { continue }; + if id.parse::().is_err() { continue } + let Ok(deadline) = expiry.parse::() else { continue }; + if timestamp >= deadline { + match fs::remove_file(entry.path()) { + Ok(()) => changed = true, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => tracing::warn!(%error, path = %entry.path().display(), "cannot remove retired identity"), + } + } else { + next_deadline = Some(next_deadline.map_or(deadline, |next| next.min(deadline))); + } + } + if changed { sync_home(home)?; } + Ok(next_deadline) +} +pub fn cleanup_retired(home: &FabricHome) -> Result<()> { + let _lock = IdentityLock::acquire(home)?; + cleanup_retired_locked(home, now()).map(|_| ()) +} + +/// Run only during exclusive daemon startup, before restoring configuration. +/// A crash before the rename leaves the old key and does not retire its dials. +pub fn reconcile_startup(home: &FabricHome) -> Result<()> { + let path = home.root().join("identity.activation"); + let expected = match fs::read_to_string(&path) { + Ok(value) => value.parse::()?, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error.into()), + }; + if read_id(&home.identity_path())? != expected { return Ok(()); } + let mut config = crate::config::FabricConfig::load(home)?; + let dials: Vec<_> = config.dials().iter().map(|dial| (dial.peer.clone(), dial.protocol.clone())).collect(); + for (peer, protocol) in dials { config.remove_dial(&peer, &protocol); } + config.save(home)?; + File::open(home.config_path())?.sync_all()?; + sync_home(home)?; + fs::remove_file(path)?; + sync_home(home) +} + +pub async fn activate_with_restart(home: &FabricHome, current: EndpointId, next: EndpointId, preserve: P, restart: F, recover: R) -> Result +where P: FnOnce() -> PFut, PFut: Future>, F: FnOnce() -> Fut, Fut: Future>, + R: FnOnce(anyhow::Error) -> RFut, RFut: Future { + let lock = IdentityLock::acquire(home)?; + let already = check_guards(home, current, next)?; + // A preserved daemon lease excludes concurrent startup through the rename. + let startup_guard = if !already { + match preserve().await { + Ok(guard) => Some(guard), + Err(error) => return Err(recover(error).await), + } + } else { None }; + let activation = activate_locked(home, current, next, already, &lock); + drop(startup_guard); + let activation = match activation { + Ok(activation) => activation, + Err(error) => return Err(recover(error).await), + }; + restart().await.map_err(|error| FabricError::new( + ErrorCode::RestartFailed, + format!("identity changed; retry the same guarded activation to restart: {error:#}"), + ).identity(current.to_string(), next.to_string()))?; + if let Some(error) = activation.durability_error { + return Err(FabricError::new(ErrorCode::RestartFailed, format!("identity renamed but directory durability could not be confirmed: {error:#}; retry the same activation")) + .identity(current.to_string(), next.to_string()).into()); + } + Ok(activation.result) +} + +#[derive(serde::Serialize, serde::Deserialize)] +pub struct RuntimeSnapshot { + pub next: EndpointId, + pub exposures: Vec, +} +pub fn write_runtime_snapshot(home: &FabricHome, next: EndpointId, exposures: Vec) -> Result<()> { + let bytes = serde_json::to_vec(&RuntimeSnapshot { next, exposures })?; + write_private_atomic(home, "identity.exposures", &bytes) +} +pub fn discard_runtime_snapshot(home: &FabricHome) -> Result<()> { + match fs::remove_file(home.root().join("identity.exposures")) { + Ok(()) => sync_home(home), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } +} + +fn scratch_path(home: &FabricHome, name: &str) -> PathBuf { + home.root().join(format!(".{name}.{}.tmp", iroh::SecretKey::generate().public())) +} + +fn write_private_atomic(home: &FabricHome, name: &str, bytes: &[u8]) -> Result<()> { + let path = home.root().join(name); + let temporary = scratch_path(home, name); + let result = (|| { + let mut file = OpenOptions::new().write(true).create_new(true).mode(0o600).open(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + fs::rename(&temporary, path)?; + sync_home(home) + })(); + if result.is_err() { let _ = fs::remove_file(&temporary); } + result +} +pub fn inherited_exposures(home: &FabricHome) -> Result> { + let path = home.root().join("identity.exposures"); + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.into()), + }; + let snapshot: RuntimeSnapshot = serde_json::from_slice(&bytes)?; + if snapshot.next == read_id(&home.identity_path())? { Ok(snapshot.exposures) } else { Ok(Vec::new()) } +} +pub fn finish_startup_snapshot(home: &FabricHome) -> Result<()> { + let bytes = match fs::read(home.root().join("identity.exposures")) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error.into()), + }; + let snapshot: RuntimeSnapshot = serde_json::from_slice(&bytes)?; + if snapshot.next != read_id(&home.identity_path())? { return Ok(()); } + match fs::remove_file(home.root().join("identity.exposures")) { + Ok(()) => sync_home(home), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } +} + +/// Persist the actual launch mode, not the mere presence of an old policy file. +pub fn record_runtime_mode(home: &FabricHome, enabled: bool) -> Result<()> { + let path = home.root().join("identity.runtime-peers"); + if enabled { + let file = OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(path)?; + file.sync_all()?; + } else { + match fs::remove_file(path) { + Ok(()) => {}, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}, + Err(error) => return Err(error.into()), + } + } + sync_home(home) +} + +pub async fn retired_key_cleanup(home: FabricHome, cancel: tokio_util::sync::CancellationToken) { + loop { + if cancel.is_cancelled() { return; } + let root = home.clone(); + let result = tokio::task::spawn_blocking(move || { + let _lock = IdentityLock::acquire(&root)?; + cleanup_retired_locked(&root, now()) + }).await; + let delay = match result { + // Every activation restarts the daemon, so a daemon with no retired + // keys needs no polling task. + Ok(Ok(None)) => return, + Ok(Ok(Some(deadline))) => std::time::Duration::from_secs(deadline.saturating_sub(now()).max(1)), + Ok(Err(error)) if crate::error::code_of(&error) == ErrorCode::IdentityBusy => std::time::Duration::from_millis(100), + Ok(Err(error)) => { + tracing::warn!(%error, "retired identity cleanup failed"); + std::time::Duration::from_secs(1) + }, + Err(error) => { + tracing::warn!(%error, "retired identity cleanup worker failed"); + std::time::Duration::from_secs(1) + }, + }; + tokio::select! { + biased; + _ = cancel.cancelled() => return, + _ = tokio::time::sleep(delay) => {}, + } + } +} + +#[cfg(test)] +mod tests { + #[tokio::test] + async fn pre_rename_recovery_retains_identity_lock_on_stop_and_write_failures() -> Result<()> { + for fail_stop in [false, true] { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + fs::create_dir(home.root().join("identity.activation"))?; + let recovered = std::cell::Cell::new(false); + let result = activate_with_restart(&home, prepared.current, prepared.next, + || async { + if fail_stop { anyhow::bail!("stop failed"); } + Ok(()) + }, + || async { panic!("pre-rename failure must not restart next") }, + |error| async { + let busy = IdentityLock::acquire(&home).err().context("recovery released identity lock").unwrap(); + assert_eq!(crate::error::code_of(&busy), ErrorCode::IdentityBusy); + assert_eq!(read_id(&home.identity_path()).unwrap(), prepared.current); + recovered.set(true); + error + }, + ).await; + assert!(result.is_err()); + assert!(recovered.get()); + } + Ok(()) + } + use super::*; + #[test] + fn prepare_is_durable_private_and_idempotent() -> Result<()> { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let first = prepare(&home)?; + let second = prepare(&home)?; + assert_eq!(first.next, second.next); + assert_eq!(first.current, second.current); + assert_eq!(status(&home)?.state, "prepared"); + assert_eq!(fs::metadata(pending(&home))?.permissions().mode() & 0o777, 0o600); + assert_eq!(load_or_create_identity(&home)?.public(), first.current); + Ok(()) + } + #[test] + fn activation_swap_crash_recovery_and_lost_reply_do_not_reswap() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let activation = activate(&home, prepared.current, prepared.next)?; + assert_eq!(activation.result.outcome, "activated"); + drop(activation); // crash before restart + assert_eq!(status(&home)?.current, prepared.next); + assert_eq!(status(&home)?.state, "active"); + let before = fs::read(home.identity_path())?; + let retry = activate(&home, prepared.current, prepared.next)?; + assert_eq!(retry.result.outcome, "already_active"); + assert_eq!(fs::read(home.identity_path())?, before); + Ok(()) + } + #[test] + fn mismatch_changes_nothing_and_operations_are_busy_while_locked() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let before = fs::read(home.identity_path())?; + let wrong = iroh::SecretKey::generate().public(); + assert_eq!(crate::error::code_of(&activate(&home, wrong, prepared.next).err().unwrap()), ErrorCode::IdentityMismatch); + assert_eq!(fs::read(home.identity_path())?, before); + assert_eq!(status(&home)?.next, Some(prepared.next)); + let _lock = IdentityLock::acquire(&home)?; + assert_eq!(crate::error::code_of(&prepare(&home).err().unwrap()), ErrorCode::IdentityBusy); + Ok(()) + } + #[test] + fn retired_secret_expires_without_touching_active_or_policy() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let _activation = activate(&home, prepared.current, prepared.next)?; + let active = fs::read(home.identity_path())?; + let retired: Vec<_> = fs::read_dir(home.root())?.filter_map(|e| e.ok()).filter(|e| e.file_name().to_string_lossy().starts_with("identity.retired.")).collect(); + assert_eq!(retired.len(), 1); + cleanup_retired_locked(&home, now().saturating_add(RETENTION_SECONDS))?; + assert!(!retired[0].path().exists()); + assert_eq!(fs::read(home.identity_path())?, active); + Ok(()) + } + + #[tokio::test] + async fn restart_failure_has_both_ids_and_retry_only_restarts() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let error = activate_with_restart(&home, prepared.current, prepared.next, || async { Ok(()) }, || async { + anyhow::bail!("injected restart failure") + }, |error| async { error }).await.unwrap_err(); + assert_eq!(crate::error::code_of(&error), ErrorCode::RestartFailed); + let report = crate::error::render(&error, true); + let json: serde_json::Value = serde_json::from_str(report.stdout.as_deref().unwrap())?; + assert_eq!(json["error"]["previous"], prepared.current.to_string()); + assert_eq!(json["error"]["active"], prepared.next.to_string()); + assert_eq!(status(&home)?.current, prepared.next); + let active = fs::read(home.identity_path())?; + let result = activate_with_restart(&home, prepared.current, prepared.next, || async { Err::<(), _>(anyhow::anyhow!("retry must not preserve again")) }, || async { + assert_eq!(load_or_create_identity(&home)?.public(), prepared.next); + Ok(()) + }, |error| async { error }).await?; + assert_eq!(result.outcome, "already_active"); + assert_eq!(fs::read(home.identity_path())?, active); + Ok(()) + } + + #[test] + fn startup_retires_dials_but_preserves_all_other_home_consumers() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let peer = iroh::SecretKey::generate().public(); + let config: crate::config::FabricConfig = toml::from_str(&format!( + "[[exposes]]\nprotocol = 'web'\ntype = 'socket'\nsocket = '/tmp/web.sock'\n[[dials]]\npeer = '{peer}'\nprotocol = 'web'\n" + ))?; + config.save(&home)?; + let original: toml::Value = toml::from_str(&fs::read_to_string(home.config_path())?)?; + let preserved = ["peers.toml", "runtime-peers.toml", "audit.jsonl", "syncs.toml"]; + for filename in preserved { + fs::write(home.root().join(filename), format!("unchanged {filename}"))?; + } + let activation = activate(&home, prepared.current, prepared.next)?; + reconcile_startup(&home)?; + assert!(crate::config::FabricConfig::load(&home)?.dials().is_empty()); + let after: toml::Value = toml::from_str(&fs::read_to_string(home.config_path())?)?; + let mut expected = original; + expected["dials"] = toml::Value::Array(Vec::new()); + assert_eq!(after, expected); + for filename in preserved { + assert_eq!(fs::read_to_string(home.root().join(filename))?, format!("unchanged {filename}")); + } + assert!(!home.root().join("identity.activation").exists()); + drop(activation); + Ok(()) + } + + #[tokio::test] + async fn failed_pre_swap_stop_keeps_prepared_identity_and_never_restarts() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let before = fs::read(home.identity_path())?; + let error = activate_with_restart(&home, prepared.current, prepared.next, + || async { Err::<(), _>(anyhow::anyhow!("old daemon stop barrier failed")) }, + || async { panic!("must not restart or rename after a failed stop barrier") }, + |error| async { error }, + ).await.unwrap_err(); + assert!(error.to_string().contains("stop barrier")); + assert_eq!(fs::read(home.identity_path())?, before); + assert_eq!(status(&home)?.state, "prepared"); + assert_eq!(status(&home)?.next, Some(prepared.next)); + Ok(()) + } + + #[tokio::test] + async fn startup_lease_spans_rename_and_is_released_before_restart() -> Result<()> { + struct Guard { lease: Option, home: FabricHome, next: EndpointId } + impl Drop for Guard { + fn drop(&mut self) { + assert_eq!(read_id(&self.home.identity_path()).unwrap(), self.next); + assert!(crate::daemon::try_daemon_lease(&self.home).unwrap().is_none()); + self.lease.take(); + } + } + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + activate_with_restart(&home, prepared.current, prepared.next, + || async { + Ok(Guard { lease: crate::daemon::try_daemon_lease(&home)?, home: home.clone(), next: prepared.next }) + }, + || async { + assert!(crate::daemon::try_daemon_lease(&home)?.is_some()); + Ok(()) + }, + |error| async { error }, + ).await?; + Ok(()) + } + + #[test] + fn incomplete_pre_rename_intent_staging_does_not_block_prepared_startup() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + // A crash may interrupt a staging write, but never publishes its bytes. + fs::write(home.root().join("identity.tmp"), b"partial")?; + reconcile_startup(&home)?; + assert_eq!(status(&home)?.current, prepared.current); + assert_eq!(status(&home)?.next, Some(prepared.next)); + drop(activate(&home, prepared.current, prepared.next)?); + assert_eq!(fs::read_to_string(home.root().join("identity.activation"))?, prepared.next.to_string()); + Ok(()) + } + + #[test] + fn cleanup_errors_and_corrupt_pending_do_not_block_status_or_prepare() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let first = prepare(&home)?; + fs::create_dir(home.root().join(format!("identity.retired.0.{}.toml", first.current)))?; + fs::write(pending(&home), "not a private key")?; + let state = status(&home)?; + assert_eq!(state.current, first.current); + assert_eq!(state.next, None); + let replacement = prepare(&home)?; + assert_eq!(replacement.current, first.current); + assert_ne!(replacement.next, first.next); + assert_eq!(prepare(&home)?.next, replacement.next); + Ok(()) + } + + #[test] + fn private_scratch_writes_do_not_follow_the_old_shared_symlink() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let sentinel = home.root().join("sentinel"); + fs::write(&sentinel, b"unchanged")?; + std::os::unix::fs::symlink(&sentinel, home.root().join("identity.tmp"))?; + write_runtime_snapshot(&home, prepared.next, Vec::new())?; + drop(activate(&home, prepared.current, prepared.next)?); + assert_eq!(fs::read(&sentinel)?, b"unchanged"); + Ok(()) + } + + #[tokio::test] + async fn activation_rechecks_pending_guard_after_preservation() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let error = activate_with_restart(&home, prepared.current, prepared.next, + || async { + fs::remove_file(pending(&home))?; + generate_identity_file(&pending(&home))?; + Ok(()) + }, + || async { panic!("changed pending guard must not restart or rename") }, + |error| async { error }, + ).await.unwrap_err(); + assert_eq!(crate::error::code_of(&error), ErrorCode::IdentityMismatch); + assert_eq!(status(&home)?.current, prepared.current); + Ok(()) + } + #[tokio::test] + async fn failed_runtime_launch_does_not_enable_retained_runtime_policy() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + prepare(&home)?; + record_runtime_mode(&home, false)?; + fs::write(home.config_path(), "invalid = [")?; + let result = crate::daemon::FabricNode::start_with_daemon_options(home.clone(), + crate::daemon::DaemonOptions { runtime_peers: true, ..Default::default() }, + ).await; + assert!(result.is_err()); + assert!(!home.root().join("identity.runtime-peers").exists()); + Ok(()) + } +} diff --git a/src/lib.rs b/src/lib.rs index 559ff280..d0f52f15 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,6 +18,7 @@ pub mod doctor; /// Stable error codes (docs/errors.md), shared with the sync companion. pub use fabric_config::error; pub mod join; +pub mod identity; pub mod macos_app; pub mod mux; pub mod pathwatch; diff --git a/src/main.rs b/src/main.rs index 634ff749..6ed54c0c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -71,6 +71,11 @@ enum CaCommands { #[derive(Debug, Subcommand)] enum Commands { + /// Prepare, inspect, or activate this machine's transport identity. + Identity { + #[command(subcommand)] + command: IdentityCommands, + }, /// Manage fabric identity key files. Key { #[command(subcommand)] @@ -494,6 +499,21 @@ enum KeyCommands { }, } +#[derive(Debug, Subcommand)] +enum IdentityCommands { + /// Durably prepare a new identity without changing the running daemon. + Prepare, + /// Inspect the authoritative active and prepared keys. + Status, + /// Activate only after peers trust the prepared NodeID. + Activate { + #[arg(long)] + expect_current: iroh::EndpointId, + #[arg(long)] + expect_next: iroh::EndpointId, + }, +} + #[derive(Debug, Subcommand)] enum ServiceCommands { /// Install and start a user service for the foreground daemon. @@ -627,6 +647,25 @@ async fn run(cli: Cli) -> Result<()> { let home = FabricHome::resolve(cli.home)?; match command { Commands::Key { .. } => unreachable!("key commands are handled before home setup"), + Commands::Identity { command } => { + match command { + IdentityCommands::Prepare => { + let prepared = fabric::identity::prepare(&home)?; + if json { print_json(&serde_json::to_value(&prepared)?)?; } + else { println!("current\t{}\nnext\t{}", prepared.current, prepared.next); } + } + IdentityCommands::Status => { + let status = fabric::identity::status(&home)?; + if json { print_json(&serde_json::to_value(&status)?)?; } + else { println!("current\t{}\nnext\t{}\nstate\t{}", status.current, status.next.map(|id| id.to_string()).unwrap_or_else(|| "-".into()), status.state); } + } + IdentityCommands::Activate { expect_current, expect_next } => { + let activated = activate_identity(&home, expect_current, expect_next).await?; + if json { print_json(&serde_json::to_value(&activated)?)?; } + else { println!("previous\t{}\nactive\t{}\noutcome\t{}", activated.previous, activated.active, activated.outcome); } + } + } + } Commands::Id => { let key = load_or_create_identity(&home)?; if json { @@ -1035,6 +1074,7 @@ async fn run(cli: Cli) -> Result<()> { runtime_peers, ); if foreground { + let _lease = fabric::daemon::DaemonLease::acquire(&home)?; init_daemon_tracing(&home)?; let node = FabricNode::start_with_daemon_options(home, options).await?; let peers = node.state().peer_reachability().await; @@ -3059,6 +3099,126 @@ fn run_restart_detacher(home: &FabricHome, _allow_shell: bool, runtime_peers: bo Ok(()) } +async fn identity_control(home: &FabricHome, request: ControlRequest) -> Result { + tokio::time::timeout(Duration::from_secs(2), send_control(home, request)).await + .map_err(|_| FabricError::new(ErrorCode::Timeout, "identity control request timed out; retry the same guarded activation to recover any exposure fence"))? +} + +async fn identity_live_status(home: &FabricHome) -> Result> { + match identity_control(home, ControlRequest::Status).await { + Ok(response @ ControlResponse::Status { .. }) => Ok(Some(response)), + Ok(response) => bail!("unexpected identity status response: {response:?}"), + Err(error) if fabric::error::code_of(&error) == ErrorCode::DaemonNotRunning => Ok(None), + Err(error) => Err(error), + } +} + +async fn identity_stop_lease(home: &FabricHome) -> Result { + tokio::time::timeout(Duration::from_secs(15), async { + loop { + // Every CLI daemon, foreground or background, retains this lease + // through teardown. A leftover pathname after a crash is harmless. + if let Some(lease) = fabric::daemon::try_daemon_lease(home)? { + return Ok::<_, anyhow::Error>(lease); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }).await.map_err(|_| FabricError::new(ErrorCode::DaemonStopTimeout, "daemon did not stop within 15 seconds"))? +} + +async fn activate_identity(home: &FabricHome, current: iroh::EndpointId, next: iroh::EndpointId) -> Result { + let snapshot_attempted = std::cell::Cell::new(false); + let stop_attempted = std::cell::Cell::new(false); + fabric::identity::activate_with_restart(home, current, next, || async { + let live = identity_live_status(home).await?; + if live.is_none() { + return identity_stop_lease(home).await; + } + snapshot_attempted.set(true); + let response = identity_control(home, ControlRequest::IdentitySnapshot { current, next }).await?; + let ControlResponse::IdentitySnapshot { runtime_peers } = response else { + bail!("unexpected identity snapshot response: {response:?}"); + }; + fabric::identity::record_runtime_mode(home, runtime_peers)?; + stop_attempted.set(true); + identity_control(home, ControlRequest::Shutdown).await?; + identity_stop_lease(home).await + }, || reconcile_identity_daemon(home, next), + |error| recover_identity_activation(home, current, snapshot_attempted.get(), stop_attempted.get(), error), + ).await +} + +async fn recover_identity_activation(home: &FabricHome, current: iroh::EndpointId, snapshot_attempted: bool, stop_attempted: bool, error: anyhow::Error) -> anyhow::Error { + if !snapshot_attempted || load_or_create_identity(home).map_or(true, |key| key.public() != current) { + return error; + } + let cleanup = fabric::identity::discard_runtime_snapshot(home); + // A live old daemon may have fenced exposures before a failed control + // reply. Clear that fence even when the stop request was never sent. + let abort = match identity_live_status(home).await { + Ok(Some(_)) => identity_control(home, ControlRequest::IdentityAbort { current }).await.map(|_| ()), + Ok(None) => Ok(()), + Err(error) => Err(error), + }; + let recovery = if stop_attempted { + match reconcile_identity_daemon(home, current).await { + Ok(()) => "current identity daemon restarted successfully".to_owned(), + Err(error) => format!("current identity restart failed: {error:#}"), + } + } else { + "current identity was not stopped".to_owned() + }; + error.context(format!( + "activation aborted before rename; {recovery}; handoff cleanup: {}; exposure fence: {}; retry the same guarded activation or restart the daemon if the fence remains", + cleanup.err().map(|e| e.to_string()).unwrap_or_else(|| "removed".into()), + abort.err().map(|e| e.to_string()).unwrap_or_else(|| "cleared".into()), + )) +} + +/// Reconcile, never roll back, after the authoritative identity rename. +async fn reconcile_identity_daemon(home: &FabricHome, expected: iroh::EndpointId) -> Result<()> { + let expected = expected.to_string(); + let mut live = identity_live_status(home).await?; + if matches!(&live, Some(ControlResponse::Status { node_id, .. }) if node_id == &expected) { + return Ok(()); + } + if live.is_none() && !fabric::daemon::daemon_lock_available(home)? { + // A supervisor may already be starting the right daemon. Give it a + // short readiness window rather than terminating it before it serves. + let until = tokio::time::Instant::now() + Duration::from_secs(2); + while tokio::time::Instant::now() < until { + live = identity_live_status(home).await?; + match &live { + Some(ControlResponse::Status { node_id, .. }) if node_id == &expected => return Ok(()), + Some(_) => break, + None => tokio::time::sleep(Duration::from_millis(50)).await, + } + } + } + if live.is_some() || !fabric::daemon::daemon_lock_available(home)? { + identity_control(home, ControlRequest::Shutdown).await?; + drop(identity_stop_lease(home).await?); + } + if service::restart_identity_service(home).await? { + // The native supervisor owns the replacement, including companion state. + } else { + // A crash after rename may leave no daemon to accept a Restart request. + // Preserve the daemon's opt-in runtime policy mode recorded before swap. + let runtime_peers = home.root().join("identity.runtime-peers").exists(); + tokio::time::timeout(Duration::from_secs(15), spawn_daemon_with_output(home, DaemonOptions { runtime_peers, ..DaemonOptions::default() }, false)) + .await.map_err(|_| FabricError::new(ErrorCode::Timeout, "replacement daemon spawn/readiness timed out"))??; + } + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if matches!(identity_control(home, ControlRequest::Status).await, Ok(ControlResponse::Status { node_id, .. }) if node_id == expected) { + return Ok::<_, anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }).await.map_err(|_| FabricError::new(ErrorCode::Timeout, "new identity did not become live within 30 seconds"))??; + Ok(()) +} + async fn run_restart_helper( home: &FabricHome, _allow_shell: bool, @@ -3566,8 +3726,12 @@ async fn request_shell_socket(home: &FabricHome, peer: &str) -> Result } async fn spawn_daemon(home: &FabricHome, options: DaemonOptions) -> Result<()> { + spawn_daemon_with_output(home, options, true).await +} + +async fn spawn_daemon_with_output(home: &FabricHome, options: DaemonOptions, announce: bool) -> Result<()> { if send_control(home, ControlRequest::Status).await.is_ok() { - println!("already running"); + if announce { println!("already running"); } return Ok(()); } @@ -3607,7 +3771,7 @@ async fn spawn_daemon(home: &FabricHome, options: DaemonOptions) -> Result<()> { let started = Instant::now(); loop { if send_control(home, ControlRequest::Status).await.is_ok() { - println!("started"); + if announce { println!("started"); } return Ok(()); } if started.elapsed() > Duration::from_secs(10) { @@ -3619,3 +3783,77 @@ async fn spawn_daemon(home: &FabricHome, options: DaemonOptions) -> Result<()> { tokio::time::sleep(Duration::from_millis(100)).await; } } + +#[cfg(test)] +mod identity_review_tests { + use super::*; + + #[tokio::test] + async fn identity_activation_waits_for_an_already_stopping_foreground_node() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = fabric::identity::prepare(&home)?; + let _lease = fabric::daemon::DaemonLease::acquire(&home)?; + let listener = tokio::net::UnixListener::bind(home.control_socket_path())?; + drop(listener); + let activation = activate_identity(&home, prepared.current, prepared.next); + tokio::pin!(activation); + assert!(tokio::time::timeout(Duration::from_millis(100), &mut activation).await.is_err()); + assert_eq!(load_or_create_identity(&home)?.public(), prepared.current); + assert!(home.root().join("identity.next.toml").exists()); + Ok(()) + } + + #[tokio::test] + async fn identity_foreground_stop_waits_for_lifetime_lease_after_socket_refusal() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + home.prepare()?; + let lease = fabric::daemon::DaemonLease::acquire(&home)?; + let listener = tokio::net::UnixListener::bind(home.control_socket_path())?; + drop(listener); + // A refused connection during cancellation does not prove endpoint + // teardown; the foreground process retains its lifetime lease. + assert!(identity_live_status(&home).await?.is_none()); + let stop = identity_stop_lease(&home); + tokio::pin!(stop); + assert!(tokio::time::timeout(Duration::from_millis(100), &mut stop).await.is_err()); + drop(lease); + drop(tokio::time::timeout(Duration::from_secs(1), stop).await??); + Ok(()) + } + + #[tokio::test] + async fn identity_control_wait_is_bounded_and_coded() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + home.prepare()?; + let listener = tokio::net::UnixListener::bind(home.control_socket_path())?; + let task = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + std::future::pending::<()>().await; + }); + let error = identity_control(&home, ControlRequest::Status).await.unwrap_err(); + assert_eq!(fabric::error::code_of(&error), ErrorCode::Timeout); + task.abort(); + Ok(()) + } + + #[tokio::test] + async fn identity_reconcile_waits_for_fresh_daemon_before_shutdown() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let expected = load_or_create_identity(&home)?.public(); + let _lease = fabric::daemon::try_daemon_lease(&home)?.context("lease")?; + let starting_home = home.clone(); + let task = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + fabric::daemon::FabricNode::start(starting_home).await + }); + reconcile_identity_daemon(&home, expected).await?; + let node = task.await??; + assert!(matches!(identity_live_status(&home).await?, Some(ControlResponse::Status { node_id, .. }) if node_id == expected.to_string())); + node.shutdown().await?; + Ok(()) + } +} diff --git a/src/service.rs b/src/service.rs index 415b7f0a..56c042f8 100644 --- a/src/service.rs +++ b/src/service.rs @@ -591,6 +591,83 @@ pub fn ensure_unsupervised_restart(home: &FabricHome) -> Result<()> { ) } +/// Restart a native owner without replacing its definition or changing enablement. +/// Returns false only when the selected home has no native owner. +pub async fn restart_identity_service(home: &FabricHome) -> Result { + if !home.is_default_state_root() { + return Ok(false); + } + match identity_service_enablement(false).await? { + ServiceEnablement::NotInstalled => return Ok(false), + ServiceEnablement::Unknown => bail!("cannot establish native service ownership"), + ServiceEnablement::Enabled | ServiceEnablement::PresentNotEnabled => {} + } + #[cfg(target_os = "linux")] + { + let companion = match identity_service_enablement(true).await? { + ServiceEnablement::NotInstalled => false, + ServiceEnablement::Unknown => bail!("cannot establish sync service ownership"), + _ => true, + }; + let (program, args) = systemd_restart_argv(companion); + let args: Vec<&str> = args.iter().map(String::as_str).collect(); + let output = identity_service_output(program, &args, Duration::from_secs(15)).await?; + if !output.status.success() { + bail!("native restart failed: {}", String::from_utf8_lossy(&output.stderr)); + } + } + #[cfg(target_os = "macos")] + { + let output = identity_service_output("launchctl", &["kickstart", "-k", &launchd_service_target()], Duration::from_secs(15)).await?; + if !output.status.success() { + bail!("native restart failed: {}", String::from_utf8_lossy(&output.stderr)); + } + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + bail!("native identity restart is unsupported on this platform"); + Ok(true) +} + +async fn identity_service_output(program: &str, args: &[&str], timeout: Duration) -> Result { + let mut command = tokio::process::Command::new(program); + command.args(args).kill_on_drop(true); + tokio::time::timeout(timeout, command.output()).await.map_err(|_| { + fabric_config::error::FabricError::new(fabric_config::error::ErrorCode::Timeout, format!("{program} timed out during identity restart")) + })?.with_context(|| format!("run {program} for identity restart")) +} + +async fn identity_service_enablement(sync: bool) -> Result { + #[cfg(target_os = "linux")] + { + let path = if sync { systemd_sync_user_unit_path() } else { systemd_user_unit_path() }; + let present = path.map(|path| path.exists()).unwrap_or(false); + let name = if sync { SYNC_SERVICE_NAME } else { SERVICE_NAME }; + identity_systemd_enablement("systemctl", name, present).await + } + #[cfg(target_os = "macos")] + { + let path = if sync { launch_sync_agent_path() } else { launch_agent_path() }; + let present = path.map(|path| path.exists()).unwrap_or(false); + let target = if sync { launchd_sync_service_target() } else { launchd_service_target() }; + let loaded = match identity_service_output("launchctl", &["print", &target], Duration::from_secs(5)).await { + Ok(output) => output.status.success(), + Err(error) if fabric_config::error::code_of(&error) == fabric_config::error::ErrorCode::Timeout => return Err(error), + Err(_) => false, + }; + Ok(interpret_launchd(present, loaded)) + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + bail!("native identity restart is unsupported on this platform") +} + +#[cfg(target_os = "linux")] +async fn identity_systemd_enablement(program: &str, name: &str, present: bool) -> Result { + match identity_service_output(program, &["--user", "is-enabled", name], Duration::from_secs(5)).await { + Ok(output) => Ok(interpret_systemd_is_enabled(present, true, &String::from_utf8_lossy(&output.stdout))), + Err(error) if fabric_config::error::code_of(&error) == fabric_config::error::ErrorCode::Timeout => Err(error), + Err(_) => Ok(interpret_systemd_is_enabled(present, false, "")), + } +} fn restart_ownership_decision( enablement: ServiceEnablement, native_restart_command: &str, @@ -1564,6 +1641,21 @@ fn xml_escape(value: &str) -> String { #[cfg(test)] mod tests { + #[cfg(target_os = "linux")] + #[tokio::test] + async fn identity_missing_native_tool_preserves_uninstalled_ownership() -> anyhow::Result<()> { + assert_eq!(super::identity_systemd_enablement("/fabric/no-such-systemctl", "fabric", false).await?, ServiceEnablement::NotInstalled); + assert_eq!(super::identity_systemd_enablement("/fabric/no-such-systemctl", "fabric", true).await?, ServiceEnablement::Unknown); + Ok(()) + } + + #[tokio::test] + async fn identity_native_command_timeout_is_bounded_and_coded() -> anyhow::Result<()> { + let error = super::identity_service_output("/bin/sh", &["-c", "exec sleep 5"], std::time::Duration::from_millis(50)).await.unwrap_err(); + assert_eq!(fabric_config::error::code_of(&error), fabric_config::error::ErrorCode::Timeout); + Ok(()) + } + use super::{ ControlSocketState, InstallVerification, ServiceEnablement, control_socket_state, interpret_launchd, interpret_systemd_is_enabled, rollback_service_command_result, diff --git a/tests/golden/json/identity_activate.json b/tests/golden/json/identity_activate.json new file mode 100644 index 00000000..f4b6b1e8 --- /dev/null +++ b/tests/golden/json/identity_activate.json @@ -0,0 +1,5 @@ +{ + "active": "", + "outcome": "activated", + "previous": "" +} diff --git a/tests/golden/json/identity_active.json b/tests/golden/json/identity_active.json new file mode 100644 index 00000000..a7474237 --- /dev/null +++ b/tests/golden/json/identity_active.json @@ -0,0 +1,5 @@ +{ + "current": "", + "next": null, + "state": "active" +} diff --git a/tests/golden/json/identity_already_active.json b/tests/golden/json/identity_already_active.json new file mode 100644 index 00000000..04218964 --- /dev/null +++ b/tests/golden/json/identity_already_active.json @@ -0,0 +1,5 @@ +{ + "active": "", + "outcome": "already_active", + "previous": "" +} diff --git a/tests/golden/json/identity_prepare.json b/tests/golden/json/identity_prepare.json new file mode 100644 index 00000000..4f38dda8 --- /dev/null +++ b/tests/golden/json/identity_prepare.json @@ -0,0 +1,4 @@ +{ + "current": "", + "next": "" +} diff --git a/tests/golden/json/identity_prepared.json b/tests/golden/json/identity_prepared.json new file mode 100644 index 00000000..e5bba2f1 --- /dev/null +++ b/tests/golden/json/identity_prepared.json @@ -0,0 +1,5 @@ +{ + "current": "", + "next": "", + "state": "prepared" +} diff --git a/tests/json_output.rs b/tests/json_output.rs index 93ee407a..19439454 100644 --- a/tests/json_output.rs +++ b/tests/json_output.rs @@ -154,6 +154,165 @@ fn ok(ran: &Ran) -> Result<&str> { Ok(&ran.stdout) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn identity_cli_prepare_activate_retry_crash_recovery_and_status_have_stable_json() -> Result<()> { + let dir = TempDir::new()?; + let home = FabricHome::new(dir.path()); + let prepared = fabric(&home, &["identity", "prepare", "--json"]).await?; + let ids: Value = serde_json::from_str(ok(&prepared)?)?; + let current = ids["current"].as_str().context("current")?; + let next = ids["next"].as_str().context("next")?; + let normaliser = Normaliser { replacements: vec![(current.into(), ""), (next.into(), "")] }; + normaliser.check("identity_prepare", ok(&prepared)?)?; + let repeated = fabric(&home, &["identity", "prepare", "--json"]).await?; + assert_eq!(ok(&prepared)?, ok(&repeated)?); + normaliser.check("identity_prepared", ok(&fabric(&home, &["identity", "status", "--json"]).await?)?)?; + let lock = fabric::identity::IdentityLock::acquire(&home)?; + let busy = fabric(&home, &["identity", "prepare", "--json"]).await?; + assert_eq!(busy.code, Some(69)); + let busy_error: Value = serde_json::from_str(&busy.stdout)?; + assert_eq!(busy_error["error"]["code"], "identity_busy"); + drop(lock); + ok(&fabric(&home, &["up", "--runtime-peers"]).await?)?; + struct Stop(FabricHome); + impl Drop for Stop { + fn drop(&mut self) { + let _ = Command::new(fabric_bin()).arg("--home").arg(self.0.root()).arg("down").output(); + } + } + let _stop = Stop(home.clone()); + let args = ["identity", "activate", "--expect-current", current, "--expect-next", next, "--json"]; + normaliser.check("identity_activate", ok(&fabric(&home, &args).await?)?)?; + normaliser.check("identity_already_active", ok(&fabric(&home, &args).await?)?)?; + normaliser.check("identity_active", ok(&fabric(&home, &["identity", "status", "--json"]).await?)?)?; + let wrong = iroh::SecretKey::generate().public().to_string(); + let mismatched = fabric(&home, &["identity", "activate", "--expect-current", current, "--expect-next", &wrong, "--json"]).await?; + assert_eq!(mismatched.code, Some(2)); + let error: Value = serde_json::from_str(&mismatched.stdout)?; + assert_eq!(error["error"]["code"], "identity_mismatch"); + normaliser.check("identity_active", ok(&fabric(&home, &["identity", "status", "--json"]).await?)?)?; + let second = fabric::identity::prepare(&home)?; + // Simulate the process dying after the single rename and before restart. + drop(fabric::identity::activate(&home, second.current, second.next)?); + let state: Value = serde_json::from_str(ok(&fabric(&home, &["identity", "status", "--json"]).await?)?)?; + assert_eq!(state["state"], "active"); + assert_eq!(state["current"], second.next.to_string()); + ok(&fabric(&home, &["down"]).await?)?; + let second_current = second.current.to_string(); + let second_next = second.next.to_string(); + let retry = fabric(&home, &["identity", "activate", "--expect-current", &second_current, "--expect-next", &second_next, "--json"]).await?; + let result: Value = serde_json::from_str(ok(&retry)?)?; + assert_eq!(result["outcome"], "already_active"); + let live: Value = serde_json::from_str(ok(&fabric(&home, &["addr", "--json"]).await?)?)?; + assert_eq!(live["id"], second_next); + let response = fabric::daemon::send_control(&home, fabric::control::ControlRequest::RuntimePeers { + request: fabric::runtime_peers::Request::List { cursor: None }, + }).await?; + let fabric::control::ControlResponse::RuntimePeers { reply } = response else { bail!("runtime mode lost across identity recovery") }; + assert!(reply.error.is_none(), "{:?}", reply.error); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn identity_foreground_activation_failed_swap_and_stale_socket_recovery_keep_runtime_off() -> Result<()> { + use fabric::{control::{ControlRequest, ControlResponse}, daemon::send_control}; + struct Stop(FabricHome); + impl Drop for Stop { + fn drop(&mut self) { + let _ = Command::new(fabric_bin()).arg("--home").arg(self.0.root()).arg("down").output(); + } + } + for fail_before_rename in [false, true] { + let dir = TempDir::new()?; + let home = FabricHome::new(dir.path()); + let prepared = fabric::identity::prepare(&home)?; + let mut foreground = tokio::process::Command::new(fabric_bin()) + .arg("--home").arg(home.root()).args(["up", "--foreground"]) + .stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null()) + .kill_on_drop(true).spawn()?; + let _stop = Stop(home.clone()); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if matches!(send_control(&home, ControlRequest::Status).await, + Ok(ControlResponse::Status { node_id, .. }) if node_id == prepared.current.to_string()) { break; } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }).await.context("foreground daemon readiness")?; + assert!(!fabric::daemon::daemon_lock_available(&home)?, "foreground must retain its lifetime lease through teardown"); + // A stale ON marker must not override the live daemon's OFF choice. + fs::write(home.root().join("identity.runtime-peers"), b"stale")?; + let mut blocked_retired_paths = Vec::new(); + if fail_before_rename { + // Fail retaining the old secret, without also making the old + // daemon's startup intent unreadable. Cover the bounded command's + // possible retirement timestamps rather than racing its clock. + let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs(); + for offset in 0..=30 { + let path = home.root().join(format!("identity.retired.{}.{}.toml", timestamp + 600 + offset, prepared.current)); + fs::create_dir(&path)?; + blocked_retired_paths.push(path); + } + } + let current = prepared.current.to_string(); + let next = prepared.next.to_string(); + let args = ["identity", "activate", "--expect-current", ¤t, "--expect-next", &next, "--json"]; + let activated = fabric(&home, &args).await?; + tokio::time::timeout(Duration::from_secs(5), foreground.wait()).await??; + if fail_before_rename { + assert_ne!(activated.code, Some(0)); + assert!(activated.stdout.contains("current identity daemon restarted successfully"), "{}", activated.stdout); + let state = fabric::identity::status(&home)?; + assert_eq!(state.current, prepared.current); + assert_eq!(state.next, Some(prepared.next)); + assert!(!home.root().join("identity.exposures").exists()); + for path in blocked_retired_paths { fs::remove_dir(path)?; } + ok(&fabric(&home, &args).await?)?; + } else { ok(&activated)?; } + let live = send_control(&home, ControlRequest::Status).await?; + assert!(matches!(live, ControlResponse::Status { node_id, .. } if node_id == next)); + let runtime = send_control(&home, ControlRequest::RuntimePeers { + request: fabric::runtime_peers::Request::List { cursor: None }, + }).await?; + let ControlResponse::RuntimePeers { reply } = runtime else { bail!("missing runtime reply") }; + assert!(reply.error.is_some(), "activation enabled runtime policy on an OFF daemon"); + assert!(!home.root().join("identity.runtime-peers").exists()); + ok(&fabric(&home, &["down"]).await?)?; + let prepared = fabric::identity::prepare(&home)?; + // A crashed process can leave a socket inode but no listener or lease. + let stale = tokio::net::UnixListener::bind(home.control_socket_path())?; + drop(stale); + let current = prepared.current.to_string(); + let next = prepared.next.to_string(); + ok(&fabric(&home, &["identity", "activate", "--expect-current", ¤t, "--expect-next", &next, "--json"]).await?)?; + assert_eq!(fabric::identity::status(&home)?.current, prepared.next); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn identity_snapshot_abort_removes_handoff_and_unfences_exposures() -> Result<()> { + use fabric::{control::{ControlRequest, ControlResponse}, daemon::send_control}; + let dir = TempDir::new()?; + let home = FabricHome::new(dir.path()); + let prepared = fabric::identity::prepare(&home)?; + let node = FabricNode::start(home.clone()).await?; + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if send_control(&home, ControlRequest::Status).await.is_ok() { break; } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }).await?; + let reply = send_control(&home, ControlRequest::IdentitySnapshot { current: prepared.current, next: prepared.next }).await?; + assert!(matches!(reply, ControlResponse::IdentitySnapshot { runtime_peers: false })); + let blocked = node.expose_tcp_ephemeral("handoff-check", "127.0.0.1:1".into()).await.unwrap_err(); + assert_eq!(fabric::error::code_of(&blocked), fabric::error::ErrorCode::IdentityBusy); + send_control(&home, ControlRequest::IdentityAbort { current: prepared.current }).await?; + assert!(!home.root().join("identity.exposures").exists()); + node.expose_tcp_ephemeral("handoff-check", "127.0.0.1:1".into()).await?; + node.shutdown().await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn every_json_command_prints_one_stable_object() -> Result<()> { let server_dir = TempDir::new()?; diff --git a/tests/local_slice.rs b/tests/local_slice.rs index bdd9b2bb..4f382e00 100644 --- a/tests/local_slice.rs +++ b/tests/local_slice.rs @@ -38,6 +38,138 @@ const LOCAL_IO_TIMEOUT: Duration = Duration::from_secs(60); const LARGE_PAYLOAD_TIMEOUT: Duration = Duration::from_secs(60); const LOCAL_SLICE_SETTLE: Duration = Duration::from_millis(500); +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn identity_activation_retires_old_transport_and_dials_preserving_policy_and_exposures() -> Result<()> { + use fabric::runtime_peers::{Operation, Request, Transaction}; + let _guard = local_slice_guard().await; + let server_dir = TempDir::new()?; + let client_dir = TempDir::new()?; + let server_home = FabricHome::new(server_dir.path()); + let client_home = FabricHome::new(client_dir.path()); + server_home.prepare()?; + let historical_audit = fabric::audit::Record::AuditLoss { + time: "2026-01-01T00:00:00Z".into(), count: 7, + first_time: "2026-01-01T00:00:00Z".into(), last_time: "2026-01-01T00:00:00Z".into(), + }; + let mut historical_line = fabric::audit::encode(&historical_audit)?; + historical_line.push(b'\n'); + fs::write(server_home.root().join("audit.jsonl"), historical_line)?; + let options = DaemonOptions { runtime_peers: true, ..DaemonOptions::default() }; + let server = FabricNode::start_with_daemon_options(server_home.clone(), options).await?; + let client = FabricNode::start_with_daemon_options(client_home.clone(), DaemonOptions { + server_session_max_total: Some(1), server_session_max_per_peer: Some(1), ..DaemonOptions::default() + }).await?; + trust_peer_allowing(&server_home, &server, client.id(), Some("client"), Some(client.addr()), &["tcp-echo", "echo"]).await?; + trust_peer_allowing(&client_home, &client, server.id(), Some("server"), Some(server.addr()), &["echo", "outgoing"]).await?; + let (backend, _, service) = spawn_tcp_echo_service().await?; + client.expose_tcp("outgoing", backend.clone()).await?; + server.expose_tcp("tcp-echo", backend.clone()).await?; + server.expose_tcp_ephemeral("ephemeral-echo", backend).await?; + let static_before = fs::read(server_home.peers_path())?; + let runtime_peer = iroh::SecretKey::generate().public(); + let revoked_peer = iroh::SecretKey::generate().public(); + let ControlResponse::RuntimePeers { reply } = send_control(&server_home, ControlRequest::RuntimePeers { request: Request::List { cursor: None } }).await? else { bail!("expected policy revision") }; + let revision = reply.snapshot.context("initial policy")?.policy_revision; + let transaction = Transaction { + owner: "identity-proof".into(), request_id: "preserved-receipt".into(), expected_revision: revision, + operations: vec![ + Operation::AddPeer { node_id: runtime_peer, name: Some("runtime".into()), direction: None, addr: None }, + Operation::SetGrants { node_id: runtime_peer, grants: vec!["echo".into()] }, + Operation::AddPeer { node_id: revoked_peer, name: None, direction: None, addr: None }, + ], + }; + let response = send_control(&server_home, ControlRequest::RuntimePeers { request: Request::Batch { transaction: transaction.clone() } }).await?; + let ControlResponse::RuntimePeers { reply } = response else { bail!("expected runtime reply") }; + assert!(reply.error.is_none(), "{:?}", reply.error); + let revoke = Transaction { + owner: "identity-proof".into(), request_id: "preserved-revocation".into(), expected_revision: revision + 1, + operations: vec![Operation::RemovePeer { node_id: revoked_peer }], + }; + let ControlResponse::RuntimePeers { reply } = send_control(&server_home, ControlRequest::RuntimePeers { request: Request::Batch { transaction: revoke.clone() } }).await? else { bail!("expected revoke reply") }; + assert!(reply.error.is_none(), "{:?}", reply.error); + let receipt_before = serde_json::to_value(&reply.mutation)?; + let prepared = fabric::identity::prepare(&server_home)?; + // Peers publish the new public identity before activation; old stays live. + trust_peer_allowing(&client_home, &client, prepared.next, Some("next"), None, &["echo", "outgoing"]).await?; + let old_addr = server.addr(); + let dial = client.dial_tcp("server", "tcp-echo", "127.0.0.1:0".into()).await?; + let mut stream = TcpStream::connect(&dial).await?; + stream.write_all(b"before").await?; + let mut before = [0; 6]; + tokio::time::timeout(LOCAL_IO_TIMEOUT, stream.read_exact(&mut before)).await.context("initial incoming echo timed out")??; + assert_eq!(&before, b"before"); + // A local declaration bound to the old local identity must not be restored. + let outgoing_addr = server.dial_tcp("client", "outgoing", "127.0.0.1:0".into()).await?; + let mut outgoing = TcpStream::connect(&outgoing_addr).await?; + outgoing.write_all(b"held").await?; + let mut held = [0; 4]; + tokio::time::timeout(LOCAL_IO_TIMEOUT, outgoing.read_exact(&mut held)).await.context("initial outgoing echo timed out")??; + assert_eq!(&held, b"held"); + let mut queued = TcpStream::connect(&outgoing_addr).await?; + queued.write_all(b"queued").await?; + let mut queued_byte = [0]; + assert!(tokio::time::timeout(Duration::from_millis(200), queued.read(&mut queued_byte)).await.is_err(), "second outgoing dial was not queued behind the one-session cap"); + assert!(!FabricConfig::load(&server_home)?.dials().is_empty()); + send_control(&server_home, ControlRequest::IdentitySnapshot { current: prepared.current, next: prepared.next }).await?; + let withdrawal = server.unexpose("tcp-echo").await.unwrap_err(); + assert_eq!(fabric::error::code_of(&withdrawal), fabric::error::ErrorCode::IdentityBusy); + server.shutdown().await?; + let activation = fabric::identity::activate(&server_home, prepared.current, prepared.next)?; + // The original daemon has stopped before the authoritative key rename. + let mut byte = [0]; + // The peer's local listener may retain its socket while retrying the now + // retired remote identity; it must not receive data from the old endpoint. + let closed = tokio::time::timeout(Duration::from_millis(300), stream.read(&mut byte)).await; + assert!(!matches!(closed, Ok(Ok(n)) if n > 0), "old tunnel remained usable"); + for pending in [&mut outgoing, &mut queued] { + let closed = tokio::time::timeout(Duration::from_secs(5), pending.read(&mut byte)).await.context("old outgoing/queued tunnel did not close")?; + assert!(matches!(closed, Ok(0) | Err(_)), "old outgoing or queued dial survived shutdown"); + } + // Start through the same fresh-construction path as the restart helper. + let server = FabricNode::start_with_daemon_options(server_home.clone(), DaemonOptions { runtime_peers: true, ..DaemonOptions::default() }).await?; + assert_eq!(server.id(), prepared.next); + wait_for_status(&server_home).await?; + assert!(FabricConfig::load(&server_home)?.dials().is_empty()); + assert_eq!(fs::read(server_home.peers_path())?, static_before); + let response = send_control(&server_home, ControlRequest::RuntimePeers { request: Request::Batch { transaction: revoke } }).await?; + let ControlResponse::RuntimePeers { reply } = response else { bail!("expected replay reply") }; + assert!(reply.error.is_none(), "{:?}", reply.error); + assert_eq!(serde_json::to_value(reply.mutation)?, receipt_before); + let ControlResponse::RuntimePeers { reply } = send_control(&server_home, ControlRequest::RuntimePeers { request: Request::List { cursor: None } }).await? else { bail!("expected policy snapshot") }; + let snapshot = reply.snapshot.context("effective policy")?; + assert!(snapshot.peers.iter().any(|peer| peer.node_id == runtime_peer && peer.services.contains_key("echo"))); + assert!(!snapshot.peers.iter().any(|peer| peer.node_id == revoked_peer)); + let ControlResponse::Status { exposed_protocols, .. } = send_control(&server_home, ControlRequest::Status).await? else { bail!("expected status") }; + assert!(exposed_protocols.iter().any(|protocol| protocol == "tcp-echo")); + assert!(exposed_protocols.iter().any(|protocol| protocol == "ephemeral-echo")); + let audit = fabric::audit::read(server_home.root(), None, None)?; + assert!(audit.records.iter().any(|record| matches!(record, fabric::audit::Record::AuditLoss { count: 7, .. }))); + // Address hints for the old public key cannot connect to the replacement. + let probe = iroh::Endpoint::builder(iroh::endpoint::presets::N0) + .secret_key(fabric::config::load_or_create_identity(&client_home)?) + .bind().await?; + assert!(!matches!(tokio::time::timeout(Duration::from_secs(2), probe.connect(old_addr, b"fabric/mux/2")).await, Ok(Ok(_)))); + probe.close().await; + trust_peer_allowing(&client_home, &client, server.id(), Some("next"), Some(server.addr()), &["echo", "outgoing"]).await?; + let new_dial = client.dial_tcp("next", "tcp-echo", "127.0.0.1:0".into()).await?; + let mut fresh = TcpStream::connect(new_dial).await?; + fresh.write_all(b"after").await?; + let mut after = [0; 5]; + tokio::time::timeout(LOCAL_IO_TIMEOUT, fresh.read_exact(&mut after)).await.context("replacement incoming echo timed out")??; + assert_eq!(&after, b"after"); + drop(activation); + server.shutdown().await?; + let server = FabricNode::start_with_daemon_options(server_home.clone(), DaemonOptions { runtime_peers: true, ..DaemonOptions::default() }).await?; + let ControlResponse::Status { exposed_protocols, .. } = wait_for_status(&server_home).await? else { bail!("expected ordinary restart status") }; + assert!(exposed_protocols.iter().any(|protocol| protocol == "tcp-echo")); + assert!(!exposed_protocols.iter().any(|protocol| protocol == "ephemeral-echo"), "activation made an ephemeral exposure permanent"); + assert!(!server_home.root().join("identity.exposures").exists(), "replacement startup did not consume the one-start handoff"); + server.shutdown().await?; + client.shutdown().await?; + service.abort(); + Ok(()) +} + /// The #304 workload: ten batches of twenty fresh HTTP connections through /// one dial listener. A successful bind must not conceal admission failures. #[tokio::test(flavor = "multi_thread", worker_threads = 4)]