From bd45ec9cac98c217266855a696f772d364fe66a5 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Fri, 30 Oct 2026 08:38:46 +0100 Subject: [PATCH] fix identity activation recovery and pending-key cleanup agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- CHANGELOG.md | 9 +++ README.md | 7 ++ src/identity.rs | 169 +++++++++++++++++++++++++++++++++++++-- src/main.rs | 185 ++++++++++++++++++++++++++++++++++++++++++- tests/json_output.rs | 42 ++++++++++ 5 files changed, 404 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 96f9995..1065248 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,15 @@ Deprecated, Fixed or Security. ## [Unreleased] +### Fixed + +- Identity activation recovery waits for an already-stopping daemon's lifetime + lease even after its control socket closes. Interrupt, terminate, and hangup + signals cannot interrupt the stop/swap/restart recovery window. +- Identity preparation removes failed scratch-key writes and sweeps abandoned + scratch keys under its lock. Only malformed pending keys are replaced; + pending-key I/O failures are reported without replacing the key. + ## [0.9.3] - 2026-10-28 ### Added diff --git a/README.md b/README.md index 0bfc533..c61d4e0 100644 --- a/README.md +++ b/README.md @@ -150,6 +150,9 @@ 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). +Preparation removes failed scratch-key writes and sweeps abandoned scratch keys +under the identity lock. A malformed pending key can be replaced; other pending +key read failures are errors and leave the key unchanged. Only after peers have installed that batch: ```sh @@ -173,6 +176,10 @@ return `already_active`; they never swap again, and `previous` echoes the caller `identity_mismatch` without swapping the identity. The identity lock remains held through restart: concurrent `status`, `prepare`, or `activate` returns `identity_busy` rather than waiting. +Interrupt, terminate, and hangup signals cannot interrupt activation's +stop/swap/restart recovery window, including when the invoking SSH session drops. +If teardown has closed the control socket, recovery still waits for its lifetime +lease before restarting the current identity. 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 diff --git a/src/identity.rs b/src/identity.rs index f6453e5..172759e 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -53,7 +53,12 @@ fn sync_home(home: &FabricHome) -> Result<()> { File::open(home.root())?.sync_al 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)?) + #[cfg(test)] + if path.file_name().is_some_and(|name| name == "identity.next.toml") { + io_faults::check(io_faults::Operation::ReadPending)?; + } + let bytes = fs::read(path)?; + let key: Key = toml::from_str(std::str::from_utf8(&bytes)?) .map_err(|_| anyhow::anyhow!("invalid identity key file {}", path.display()))?; Ok(key.secret_key.public()) } @@ -62,8 +67,9 @@ fn snapshot(home: &FabricHome) -> Result { 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) if error.downcast_ref::().is_some() => return Err(error), Err(error) => { - tracing::warn!(%error, "ignoring unreadable pending identity; prepare replaces it"); + tracing::warn!(%error, "ignoring invalid pending identity; prepare replaces it"); None } }; @@ -79,6 +85,7 @@ pub fn status(home: &FabricHome) -> Result { } pub fn prepare(home: &FabricHome) -> Result { let _lock = IdentityLock::acquire(home)?; + cleanup_pending_scratch_locked(home)?; if let Err(error) = cleanup_retired_locked(home, now()) { tracing::warn!(%error, "retired identity cleanup failed"); } @@ -88,11 +95,36 @@ pub fn prepare(home: &FabricHome) -> Result { 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 }) + let result = (|| { + let next = generate_identity_file(&temporary)?; + #[cfg(test)] + io_faults::check(io_faults::Operation::SyncPending)?; + File::open(&temporary)?.sync_all()?; + #[cfg(test)] + io_faults::check(io_faults::Operation::RenamePending)?; + fs::rename(&temporary, pending(home))?; + sync_home(home)?; + Ok(Prepared { current: state.current, next }) + })(); + if result.is_err() { let _ = fs::remove_file(&temporary); } + result +} + +// A previous prepare may have crashed before publishing its private key. +// The identity lock excludes any live writer of these scratch files. +fn cleanup_pending_scratch_locked(home: &FabricHome) -> Result<()> { + let mut changed = false; + for entry in fs::read_dir(home.root())? { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + if name.starts_with(".identity.next.") && name.ends_with(".tmp") { + fs::remove_file(entry.path())?; + changed = true; + } + } + if changed { sync_home(home)?; } + Ok(()) } /// The lock remains held across the caller's restart barrier. A retry after the @@ -337,8 +369,131 @@ pub async fn retired_key_cleanup(home: FabricHome, cancel: tokio_util::sync::Can } } +#[cfg(test)] +mod io_faults { + #[derive(Clone, Copy, PartialEq, Eq)] + pub enum Operation { ReadPending, SyncPending, RenamePending } + + thread_local! { + static FAILURE: std::cell::Cell> = const { std::cell::Cell::new(None) }; + } + + pub fn fail_next(operation: Operation, errno: i32) { + FAILURE.with(|failure| { + assert!(failure.replace(Some((operation, errno))).is_none()); + }); + } + + pub fn check(operation: Operation) -> std::io::Result<()> { + FAILURE.with(|failure| { + if let Some((expected, errno)) = failure.get() { + if expected == operation { + failure.set(None); + return Err(std::io::Error::from_raw_os_error(errno)); + } + } + Ok(()) + }) + } +} + #[cfg(test)] mod tests { + #[test] + fn prepare_removes_private_scratch_after_sync_or_rename_failure() -> Result<()> { + for operation in [io_faults::Operation::SyncPending, io_faults::Operation::RenamePending] { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let current = load_or_create_identity(&home)?.public(); + let active = fs::read(home.identity_path())?; + io_faults::fail_next(operation, libc::EIO); + let error = prepare(&home).unwrap_err(); + assert_eq!(error.downcast_ref::().unwrap().raw_os_error(), Some(libc::EIO)); + assert_eq!(fs::read(home.identity_path())?, active); + assert!(!pending(&home).exists()); + for entry in fs::read_dir(home.root())? { + assert!(!entry?.file_name().to_string_lossy().starts_with(".identity.next.")); + } + let retry = prepare(&home)?; + assert_eq!(retry.current, current); + assert_eq!(read_id(&pending(&home))?, retry.next); + } + Ok(()) + } + + #[test] + fn prepare_sweeps_crashed_scratch_only_while_locked_even_when_already_prepared() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let active = fs::read(home.identity_path())?; + let next = fs::read(pending(&home))?; + let stale = scratch_path(&home, "identity.next"); + generate_identity_file(&stale)?; + let partial = scratch_path(&home, "identity.next"); + fs::write(&partial, b"partial secret")?; + let unrelated = [".identity.exposures.other.tmp", ".identity.next.other.toml", "identity.next.other.tmp"]; + for name in unrelated { fs::write(home.root().join(name), b"unchanged")?; } + let lock = IdentityLock::acquire(&home)?; + assert_eq!(crate::error::code_of(&prepare(&home).unwrap_err()), ErrorCode::IdentityBusy); + assert!(stale.exists()); + assert!(partial.exists()); + drop(lock); + assert_eq!(prepare(&home)?.next, prepared.next); + assert!(!stale.exists()); + assert!(!partial.exists()); + assert_eq!(fs::read(home.identity_path())?, active); + assert_eq!(fs::read(pending(&home))?, next); + for name in unrelated { assert_eq!(fs::read(home.root().join(name))?, b"unchanged"); } + Ok(()) + } + + #[test] + fn pending_read_io_failure_preserves_valid_key_and_propagates_existing_error_code() -> Result<()> { + for errno in [libc::EMFILE, libc::EIO, libc::EACCES] { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let prepared = prepare(&home)?; + let active = fs::read(home.identity_path())?; + let next = fs::read(pending(&home))?; + for read_status in [false, true] { + io_faults::fail_next(io_faults::Operation::ReadPending, errno); + let error = if read_status { status(&home).unwrap_err() } else { prepare(&home).unwrap_err() }; + assert_eq!(error.downcast_ref::().unwrap().raw_os_error(), Some(errno)); + assert_eq!(crate::error::code_of(&error), ErrorCode::Internal); + assert_eq!(fs::read(home.identity_path())?, active); + assert_eq!(fs::read(pending(&home))?, next); + } + assert_eq!(prepare(&home)?.next, prepared.next); + } + Ok(()) + } + + #[test] + fn prepare_replaces_pending_only_when_absent_or_invalid() -> Result<()> { + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let mut prepared = prepare(&home)?; + let active = fs::read(home.identity_path())?; + for invalid in [b"\xff".as_slice(), b"not a private key", b"secret_key = 'invalid'"] { + fs::write(pending(&home), invalid)?; + assert_eq!(status(&home)?.next, None); + let replacement = prepare(&home)?; + assert_eq!(replacement.current, prepared.current); + assert_ne!(replacement.next, prepared.next); + assert_eq!(read_id(&pending(&home))?, replacement.next); + assert_eq!(prepare(&home)?.next, replacement.next); + assert_eq!(fs::read(home.identity_path())?, active); + prepared = replacement; + } + fs::remove_file(pending(&home))?; + let replacement = prepare(&home)?; + assert_eq!(replacement.current, prepared.current); + assert_ne!(replacement.next, prepared.next); + assert_eq!(read_id(&pending(&home))?, replacement.next); + Ok(()) + } + #[tokio::test] async fn pre_rename_recovery_retains_identity_lock_on_stop_and_write_failures() -> Result<()> { for fail_stop in [false, true] { diff --git a/src/main.rs b/src/main.rs index 6ed54c0..8a9beb5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3126,7 +3126,59 @@ async fn identity_stop_lease(home: &FabricHome) -> Result { }).await.map_err(|_| FabricError::new(ErrorCode::DaemonStopTimeout, "daemon did not stop within 15 seconds"))? } +/// Activation must finish its rename/restart or old-key recovery after stopping +/// the daemon. Discard cancellation signals for this scope, then restore the +/// caller's dispositions. A caught no-op (not SIG_IGN) resets on exec, so a +/// replacement daemon does not inherit ignored termination signals. +struct IdentityActivationSignals { + previous: [Option; 3], +} + +impl IdentityActivationSignals { + const SIGNALS: [libc::c_int; 3] = [libc::SIGINT, libc::SIGTERM, libc::SIGHUP]; + + fn install() -> Result { + extern "C" fn finish_activation(_: libc::c_int) {} + + let mut guard = Self { previous: [None, None, None] }; + // SAFETY: zero initializes the platform sigaction structure; its mask is + // initialized below, and the handler neither accesses memory nor calls + // anything that could violate async-signal safety. + let mut action: libc::sigaction = unsafe { std::mem::zeroed() }; + action.sa_sigaction = finish_activation as *const () as libc::sighandler_t; + action.sa_flags = libc::SA_RESTART; + // SAFETY: action owns a valid, writable signal set. + if unsafe { libc::sigemptyset(&mut action.sa_mask) } != 0 { + return Err(std::io::Error::last_os_error().into()); + } + for (index, signal) in Self::SIGNALS.into_iter().enumerate() { + let mut previous = std::mem::MaybeUninit::::uninit(); + // SAFETY: both pointers refer to live correctly sized structures. + // sigaction initializes previous on success and retains no pointers. + if unsafe { libc::sigaction(signal, &action, previous.as_mut_ptr()) } != 0 { + return Err(std::io::Error::last_os_error().into()); + } + // SAFETY: the successful sigaction call initialized previous. + guard.previous[index] = Some(unsafe { previous.assume_init() }); + } + Ok(guard) + } +} + +impl Drop for IdentityActivationSignals { + fn drop(&mut self) { + for (signal, previous) in Self::SIGNALS.into_iter().zip(&self.previous) { + if let Some(previous) = previous { + // SAFETY: restore exactly the disposition returned by sigaction; + // the pointer is valid and no old-action output is requested. + unsafe { libc::sigaction(signal, previous, std::ptr::null_mut()); } + } + } + } +} + async fn activate_identity(home: &FabricHome, current: iroh::EndpointId, next: iroh::EndpointId) -> Result { + let _signals = IdentityActivationSignals::install().context("protect identity activation from cancellation signals")?; 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 { @@ -3196,7 +3248,14 @@ async fn reconcile_identity_daemon(home: &FabricHome, expected: iroh::EndpointId } } if live.is_some() || !fabric::daemon::daemon_lock_available(home)? { - identity_control(home, ControlRequest::Shutdown).await?; + match identity_control(home, ControlRequest::Shutdown).await { + Ok(_) => {} + // A previous Shutdown can have closed the control socket while + // teardown still owns the lease. Wait for that teardown, then + // restart instead of abandoning pre-rename recovery. + Err(error) if fabric::error::code_of(&error) == ErrorCode::DaemonNotRunning => {} + Err(error) => return Err(error), + } drop(identity_stop_lease(home).await?); } if service::restart_identity_service(home).await? { @@ -3788,6 +3847,130 @@ async fn spawn_daemon_with_output(home: &FabricHome, options: DaemonOptions, ann mod identity_review_tests { use super::*; + fn activation_signal_subprocess(signal: libc::c_int) -> Result<()> { + for recovery in [false, true] { + let output = ProcessCommand::new(std::env::current_exe()?) + .args(["--exact", "identity_review_tests::identity_activation_signal_child", "--nocapture"]) + .env("FABRIC_ACTIVATION_TEST_SIGNAL", signal.to_string()) + .env("FABRIC_ACTIVATION_TEST_RECOVERY", recovery.to_string()) + .output()?; + assert!(output.status.success(), + "activation signal {signal}, recovery={recovery}: {}\nstdout: {}\nstderr: {}", + output.status, String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr)); + } + Ok(()) + } + + #[test] + fn identity_activation_survives_sigint_through_restart_and_recovery() -> Result<()> { + activation_signal_subprocess(libc::SIGINT) + } + + #[test] + fn identity_activation_survives_sigterm_through_restart_and_recovery() -> Result<()> { + activation_signal_subprocess(libc::SIGTERM) + } + + #[test] + fn identity_activation_survives_sighup_through_restart_and_recovery() -> Result<()> { + activation_signal_subprocess(libc::SIGHUP) + } + + #[tokio::test] + async fn identity_activation_signal_child() -> Result<()> { + use std::os::unix::process::ExitStatusExt; + use tokio::io::{AsyncBufReadExt, BufReader}; + + let Ok(signal) = std::env::var("FABRIC_ACTIVATION_TEST_SIGNAL") else { return Ok(()); }; + let signal: libc::c_int = signal.parse()?; + let recovery = std::env::var("FABRIC_ACTIVATION_TEST_RECOVERY")? == "true"; + assert!(IdentityActivationSignals::SIGNALS.contains(&signal)); + // SAFETY: only this disposable subprocess is affected. Establish a + // terminating disposition so each regression fails without the guard. + unsafe { libc::signal(signal, libc::SIG_DFL); } + 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())?; + let server_home = home.clone(); + // Model the control boundary without starting network endpoints. Signals + // are delivered synchronously at Shutdown, strictly before the rename, + // and again during post-stop reconciliation / pre-rename recovery. + let server = tokio::spawn(async move { + let mut lease = Some(lease); + let mut stopped = false; + let mut post_stop_signals = 0; + loop { + let (stream, _) = listener.accept().await?; + let mut stream = BufReader::new(stream); + let mut line = String::new(); + stream.read_line(&mut line).await?; + let request: ControlRequest = serde_json::from_str(&line)?; + let response = match request { + ControlRequest::Status => { + if stopped { + // SAFETY: raise targets only this subprocess and + // uses one of the validated catchable signals. + assert_eq!(unsafe { libc::raise(signal) }, 0); + post_stop_signals += 1; + } + ControlResponse::Status { + node_id: load_or_create_identity(&server_home)?.public().to_string(), + endpoint_addr: serde_json::Value::Null, + exposed_protocols: vec![], dial_sockets: vec![], + allow_shell: false, allow_exec: false, audit: None, + } + } + ControlRequest::IdentitySnapshot { .. } => { + ControlResponse::IdentitySnapshot { runtime_peers: false } + } + ControlRequest::Shutdown => { + assert_eq!(load_or_create_identity(&server_home)?.public(), prepared.current); + drop(lease.take()); + stopped = true; + if recovery { + // Force the actual pre-rename error/recovery path. + fs::remove_file(server_home.root().join("identity.next.toml"))?; + } + // SAFETY: same subprocess-local signal delivery as above. + assert_eq!(unsafe { libc::raise(signal) }, 0); + // Exec'd replacement/service children must still obey + // termination; SIG_IGN would incorrectly survive exec. + let child = ProcessCommand::new("sh") + .args(["-c", &format!("kill -{signal} $$; exit 0")]) + .status()?; + assert_eq!(child.signal(), Some(signal)); + ControlResponse::Ok + } + ControlRequest::IdentityAbort { .. } => ControlResponse::Ok, + request => bail!("unexpected activation request: {request:?}"), + }; + stream.get_mut().write_all(&serde_json::to_vec(&response)?).await?; + stream.get_mut().shutdown().await?; + if stopped && ((!recovery && post_stop_signals == 1) || (recovery && post_stop_signals == 2)) { + return Ok::<_, anyhow::Error>(()); + } + } + }); + let result = activate_identity(&home, prepared.current, prepared.next).await; + if recovery { + let error = result.err().context("activation should fail before rename")?; + assert!(format!("{error:#}").contains("current identity daemon restarted successfully")); + assert_eq!(load_or_create_identity(&home)?.public(), prepared.current); + } else { + assert_eq!(result?.active, prepared.next); + assert_eq!(load_or_create_identity(&home)?.public(), prepared.next); + } + server.await??; + let mut restored = std::mem::MaybeUninit::::uninit(); + // SAFETY: querying with a null action only writes the live output. + assert_eq!(unsafe { libc::sigaction(signal, std::ptr::null(), restored.as_mut_ptr()) }, 0); + // SAFETY: the successful query initialized restored. + assert_eq!(unsafe { restored.assume_init() }.sa_sigaction, libc::SIG_DFL); + Ok(()) + } + #[tokio::test] async fn identity_activation_waits_for_an_already_stopping_foreground_node() -> Result<()> { let dir = tempfile::tempdir()?; diff --git a/tests/json_output.rs b/tests/json_output.rs index 1943945..ee9caed 100644 --- a/tests/json_output.rs +++ b/tests/json_output.rs @@ -289,6 +289,48 @@ async fn identity_foreground_activation_failed_swap_and_stale_socket_recovery_ke Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn identity_stop_timeout_recovery_waits_for_teardown_and_restarts_current() -> Result<()> { + use fabric::{control::{ControlRequest, ControlResponse}, daemon::{DaemonLease, 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(); + } + } + let dir = TempDir::new()?; + let home = FabricHome::new(dir.path()); + let prepared = fabric::identity::prepare(&home)?; + let lease = DaemonLease::acquire(&home)?; + let node = FabricNode::start(home.clone()).await?; + let _stop = Stop(home.clone()); + // Model a daemon which acknowledged Shutdown and stopped listening, but + // retains its lifetime lease beyond the activation's 15-second deadline. + let teardown = tokio::spawn(async move { + node.wait().await?; + tokio::time::sleep(Duration::from_secs(18)).await; + drop(lease); + Ok::<_, anyhow::Error>(()) + }); + let output = tokio::time::timeout(Duration::from_secs(45), + tokio::process::Command::new(fabric_bin()) + .arg("--home").arg(home.root()) + .args(["identity", "activate", "--expect-current", &prepared.current.to_string(), + "--expect-next", &prepared.next.to_string(), "--json"]) + .kill_on_drop(true).output() + ).await.context("activation recovery deadline")??; + let result: Value = serde_json::from_slice(&output.stdout)?; + assert!(!output.status.success()); + assert_eq!(result["error"]["code"], "daemon_stop_timeout"); + assert!(String::from_utf8_lossy(&output.stdout).contains("current identity daemon restarted successfully"), + "{}", String::from_utf8_lossy(&output.stdout)); + teardown.await??; + assert_eq!(fabric::identity::status(&home)?.next, Some(prepared.next)); + assert!(matches!(send_control(&home, ControlRequest::Status).await?, + ControlResponse::Status { node_id, .. } if node_id == prepared.current.to_string())); + 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};