diff --git a/crates/bitwarden-ipc/src/crypto_provider/noise/crypto_provider.rs b/crates/bitwarden-ipc/src/crypto_provider/noise/crypto_provider.rs index cfcc817088..64afc6ed4d 100644 --- a/crates/bitwarden-ipc/src/crypto_provider/noise/crypto_provider.rs +++ b/crates/bitwarden-ipc/src/crypto_provider/noise/crypto_provider.rs @@ -2,7 +2,7 @@ use std::{sync::LazyLock, time::Duration}; use bitwarden_threading::time::timeout; use serde::{Deserialize, Serialize}; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use crate::{ crypto_provider::noise::{ @@ -12,13 +12,14 @@ use crate::{ }, transport_state::{PersistentTransportState, TransportFrame}, }, - error::IpcErrorKind, + error::{ErrorKind, IpcErrorKind}, message::{IncomingMessage, OutgoingMessage}, traits::{ CommunicationBackend, CommunicationBackendReceiver, CryptoProvider, SessionRepository, }, }; +/// A `CryptoProvider` that encrypts IPC traffic using the Noise protocol. pub struct NoiseCryptoProvider; #[derive(Debug)] @@ -27,33 +28,47 @@ pub enum NoiseCryptoProviderError { HandshakeProtocol, /// A timeout waiting for a message Timeout, - /// Could not send via the underlying transport. `fatal` is derived from the underlying - /// backend error's [`IpcErrorKind`] classification. - TransportSend { fatal: bool }, - /// Could not receive via the underlying transport. `fatal` is derived from the underlying - /// backend error's [`IpcErrorKind`] classification. - TransportReceive { fatal: bool }, + /// The destination could not be reached (the underlying transport is not connected). + TransportUnreachable, + /// Could not send via the underlying transport. `kind` is the underlying backend error's + /// [`IpcErrorKind`] classification. + TransportSend { kind: ErrorKind }, + /// Could not receive via the underlying transport. `kind` is the underlying backend error's + /// [`IpcErrorKind`] classification. + TransportReceive { kind: ErrorKind }, /// A cryptographic error. In most cases, such messages are just dropped. DecryptionFailure, } impl IpcErrorKind for NoiseCryptoProviderError { - fn is_fatal(&self) -> bool { + fn kind(&self) -> ErrorKind { match self { // A bad/missing handshake frame from one peer does not affect the shared client; the // peer can retry the handshake. - NoiseCryptoProviderError::HandshakeProtocol => false, + NoiseCryptoProviderError::HandshakeProtocol => ErrorKind::Other, // The handshake is retryable on a subsequent send. - NoiseCryptoProviderError::Timeout => false, + NoiseCryptoProviderError::Timeout => ErrorKind::Other, // A decryption failure only affects the offending message, which is dropped. - NoiseCryptoProviderError::DecryptionFailure => false, + NoiseCryptoProviderError::DecryptionFailure => ErrorKind::Other, + // An unreachable destination; the message simply could not be delivered. + NoiseCryptoProviderError::TransportUnreachable => ErrorKind::Unreachable, // Defer to the underlying backend's classification, captured at construction. - NoiseCryptoProviderError::TransportSend { fatal } => *fatal, - NoiseCryptoProviderError::TransportReceive { fatal } => *fatal, + NoiseCryptoProviderError::TransportSend { kind } + | NoiseCryptoProviderError::TransportReceive { kind } => *kind, } } } +/// Classify a transport send failure: an unreachable destination becomes the dedicated +/// [`NoiseCryptoProviderError::TransportUnreachable`], while every other failure preserves the +/// underlying backend's fatal/recoverable classification. +fn transport_send_error(e: E) -> NoiseCryptoProviderError { + match e.kind() { + ErrorKind::Unreachable => NoiseCryptoProviderError::TransportUnreachable, + kind => NoiseCryptoProviderError::TransportSend { kind }, + } +} + // Serialize send operations to prevent concurrent reads of the same persisted // transport state, which can cause nonce reuse. static CRYPTO_STATE_GUARD: LazyLock> = @@ -69,7 +84,7 @@ impl NoiseCryptoProvider { Com: CommunicationBackend, Ses: SessionRepository, { - info!("Starting noise handshake with {:?}", destination); + debug!("Starting noise handshake with {:?}", destination); let mut initiator = HandshakeInitiator::new(&CipherSuite::default()); let message = initiator @@ -85,18 +100,15 @@ impl NoiseCryptoProvider { topic: None, }) .await - .map_err(|e| NoiseCryptoProviderError::TransportSend { - fatal: e.is_fatal(), - })?; + .map_err(transport_send_error)?; // Wait for the handshake response (with timeout) timeout(Duration::from_secs(HANDSHAKE_TIMEOUT_SECS), async { loop { - let incoming = receiver.receive().await.map_err(|e| { - NoiseCryptoProviderError::TransportReceive { - fatal: e.is_fatal(), - } - })?; + let incoming = receiver + .receive() + .await + .map_err(|e| NoiseCryptoProviderError::TransportReceive { kind: e.kind() })?; // For concurrent handshakes, ignore messages if incoming.source.to_endpoint() != destination { @@ -204,17 +216,21 @@ where if should_handshake { if crypto_state.is_none() { - info!( + debug!( "Noise handshake with {:?} initiated for new session establishment", destination ); } else { - info!( + debug!( "Noise re-handshake with {:?} due to re-handshake interval", destination ); } + // Propagate every handshake failure, including an unreachable transport. The + // unreachable case surfaces as `NoiseCryptoProviderError::TransportUnreachable` + // (non-fatal), which the logging layers intentionally do not log — so it no longer + // needs to be swallowed here to avoid spam. Self::perform_handshake(communication, sessions, destination.clone()).await?; } @@ -229,16 +245,53 @@ where .state .send(message.payload.into()) .map_err(|_| NoiseCryptoProviderError::DecryptionFailure)?; - communication + if let Err(e) = communication .send(OutgoingMessage { payload: Frame::TransportFrame(transport_frame).to_cbor(), destination: destination.clone(), topic: message.topic, }) .await - .map_err(|e| NoiseCryptoProviderError::TransportSend { - fatal: e.is_fatal(), - })?; + .map_err(transport_send_error) + { + match e.kind() { + ErrorKind::Fatal => { + error!( + "{:?} fatal error sending message. Clearing cryptographic sessions.", + destination + ); + sessions + .remove(destination.clone()) + .await + .expect("Delete session should not fail"); + return Err(e); + } + ErrorKind::Unreachable => { + // If a destination goes offline, the cryptographic session is torn down. + // The next time the destination comes back online, a new handshake will be + // performed. If this were not done, then the first message + // would always be dropped by the destination, + // after the destination process-reloads because it would not be decryptable by + // the destination. + info!( + "{:?} is unreachable. Clearing cryptographic sessions.", + destination + ); + sessions + .remove(destination.clone()) + .await + .expect("Delete session should not fail"); + return Err(e); + } + // Every other recoverable send failure is still surfaced. + ErrorKind::Other => { + error!( + "Recoverable error sending message to {:?}: {:?}", + destination, e + ); + } + } + } sessions .save(destination, crypto_state) @@ -255,11 +308,10 @@ where sessions: &Ses, ) -> Result { loop { - let message = receiver.receive().await.map_err(|e| { - NoiseCryptoProviderError::TransportReceive { - fatal: e.is_fatal(), - } - })?; + let message = receiver + .receive() + .await + .map_err(|e| NoiseCryptoProviderError::TransportReceive { kind: e.kind() })?; // Ensure session exists let source_endpoint: crate::endpoint::Endpoint = message.source.clone().into(); @@ -287,9 +339,7 @@ where topic: None, }) .await - .map_err(|e| NoiseCryptoProviderError::TransportSend { - fatal: e.is_fatal(), - })?; + .map_err(transport_send_error)?; let crypto_state = NoiseCryptoProviderState { state: (&mut responder).into(), @@ -306,7 +356,7 @@ where .await .expect("Get session should not fail"); let Some(mut state) = crypto_state else { - info!("No session for {:?}, waiting for handshake", message.source); + debug!("No session for {:?}, waiting for handshake", message.source); let frame = Frame::CryptoInvalidated.to_cbor(); communication .send(OutgoingMessage { @@ -315,9 +365,7 @@ where topic: None, }) .await - .map_err(|e| NoiseCryptoProviderError::TransportSend { - fatal: e.is_fatal(), - })?; + .map_err(transport_send_error)?; continue; }; diff --git a/crates/bitwarden-ipc/src/error.rs b/crates/bitwarden-ipc/src/error.rs index fe89938840..6174744b0a 100644 --- a/crates/bitwarden-ipc/src/error.rs +++ b/crates/bitwarden-ipc/src/error.rs @@ -3,30 +3,43 @@ use thiserror::Error; use crate::rpc::error::RpcError; -/// Classifies an IPC error as either fatal or recoverable. +/// Classification of an IPC error, returned by [`IpcErrorKind::kind`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorKind { + /// The client can no longer make progress, so the shared processing loop should stop. + Fatal, + /// The destination could not be reached (e.g. the peer transport is not connected); + /// The client should continue to process messages. A peer may become reachable later. + Unreachable, + /// Any other, recoverable failure: only the current operation failed, so the client should stay + /// running and continue processing other messages. + Other, +} + +/// Classifies an IPC error into an [`ErrorKind`]. /// /// The IPC client runs a single long-lived processing loop that is shared across every peer and /// every message. Historically *any* transport or crypto error tore that loop down, which meant a /// single transient failure (a handshake timeout, a peer disconnecting mid-send, a malformed /// frame) permanently disabled the shared client and it never recovered. /// -/// This trait lets each layer classify its own errors so the client can distinguish the two cases: -/// - **Fatal** (`is_fatal() == true`): the client can no longer make progress, so the processing -/// loop should stop. -/// - **Recoverable** (`is_fatal() == false`): only the current operation failed, so the client -/// should stay running and continue processing other messages. +/// This trait lets each layer classify its own errors so the client can distinguish the cases: +/// - [`ErrorKind::Fatal`]: the client can no longer make progress, so the processing loop should +/// stop. +/// - [`ErrorKind::Unreachable`] and [`ErrorKind::Other`]: only the current operation failed, so the +/// client should stay running and continue processing other messages. /// /// Implementations should classify errors at construction, where the most context is available, -/// and default ambiguous cases to recoverable. Failing open keeps the shared client alive, which -/// is almost always the safer choice. +/// and default ambiguous cases to [`ErrorKind::Other`]. Failing open keeps the shared client alive, +/// which is almost always the safer choice. pub trait IpcErrorKind { - /// Returns `true` if the error is fatal and the IPC client should stop processing messages, or - /// `false` if the error is recoverable and the client should keep running. - fn is_fatal(&self) -> bool; + /// Classifies the error so the IPC client can decide whether to stop the processing loop or + /// keep running. + fn kind(&self) -> ErrorKind; } impl IpcErrorKind for std::convert::Infallible { - fn is_fatal(&self) -> bool { + fn kind(&self) -> ErrorKind { // `Infallible` can never be constructed, so this is unreachable. match *self {} } @@ -34,8 +47,8 @@ impl IpcErrorKind for std::convert::Infallible { #[cfg(any(test, feature = "test-support"))] impl IpcErrorKind for () { - fn is_fatal(&self) -> bool { - false + fn kind(&self) -> ErrorKind { + ErrorKind::Other } } @@ -46,11 +59,16 @@ impl IpcErrorKind for () { #[bitwarden_error(basic)] pub struct AlreadyRunningError; -/// Error returned by [`IpcClient::send`](crate::IpcClient::send). Wraps the underlying transport -/// error as a string. +/// Error returned by [`IpcClient::send`](crate::IpcClient::send). #[derive(Debug, Error, Clone, PartialEq, Eq)] -#[error("{0}")] -pub struct SendError(pub(crate) String); +pub enum SendError { + /// The destination could not be reached (e.g. the peer transport is not connected). + #[error("Destination unreachable")] + Unreachable, + /// Any other send failure, carrying the underlying error's debug representation. + #[error("{0}")] + Other(String), +} #[derive(Debug, Error, Clone, PartialEq, Eq)] #[bitwarden_error(flat)] @@ -117,6 +135,18 @@ pub enum RequestError { #[error("Failed to send message: {0}")] Send(String), + #[error("Destination unreachable")] + Unreachable, + #[error("Error occurred on the remote target: {0}")] Rpc(#[from] RpcError), } + +impl From for RequestError { + fn from(error: SendError) -> Self { + match error { + SendError::Unreachable => RequestError::Unreachable, + SendError::Other(message) => RequestError::Send(message), + } + } +} diff --git a/crates/bitwarden-ipc/src/ipc_client.rs b/crates/bitwarden-ipc/src/ipc_client.rs index f130e21a0b..a6516df445 100644 --- a/crates/bitwarden-ipc/src/ipc_client.rs +++ b/crates/bitwarden-ipc/src/ipc_client.rs @@ -8,7 +8,7 @@ use tokio::select; use crate::{ constants::CHANNEL_BUFFER_CAPACITY, error::{ - AlreadyRunningError, IpcErrorKind, ReceiveError, SendError, SubscribeError, + AlreadyRunningError, ErrorKind, IpcErrorKind, ReceiveError, SendError, SubscribeError, TypedReceiveError, }, message::{ @@ -159,7 +159,7 @@ where break; }; } - Err(error) if error.is_fatal() => { + Err(error) if matches!(error.kind(), ErrorKind::Fatal) => { tracing::error!(?error, "Fatal error receiving message, stopping IPC client"); break; } @@ -209,18 +209,27 @@ where .await; if let Err(ref error) = result { - if error.is_fatal() { - tracing::error!(?error, "Fatal error sending message, stopping IPC client"); - stop_inner(&self.inner); - } else { - tracing::warn!( - ?error, - "Recoverable error sending message, IPC client will continue running" - ); + match error.kind() { + ErrorKind::Fatal => { + tracing::error!(?error, "Fatal error sending message, stopping IPC client"); + stop_inner(&self.inner); + } + // An unreachable destination is an expected condition and not logged + ErrorKind::Unreachable => {} + // Every other recoverable send failure is still surfaced. + ErrorKind::Other => { + tracing::warn!( + ?error, + "Recoverable error sending message, IPC client will continue running" + ); + } } } - result.map_err(|e| SendError(format!("{e:?}"))) + result.map_err(|e| match e.kind() { + ErrorKind::Unreachable => SendError::Unreachable, + _ => SendError::Other(format!("{e:?}")), + }) } async fn subscribe( @@ -413,8 +422,12 @@ mod tests { } impl IpcErrorKind for TestCryptoError { - fn is_fatal(&self) -> bool { - self.fatal + fn kind(&self) -> ErrorKind { + if self.fatal { + ErrorKind::Fatal + } else { + ErrorKind::Other + } } } diff --git a/crates/bitwarden-ipc/src/ipc_client_ext.rs b/crates/bitwarden-ipc/src/ipc_client_ext.rs index ffa051ca4a..f16abb1570 100644 --- a/crates/bitwarden-ipc/src/ipc_client_ext.rs +++ b/crates/bitwarden-ipc/src/ipc_client_ext.rs @@ -53,9 +53,7 @@ pub trait IpcClientExt: IpcClient { RequestError::Rpc(RpcError::RequestSerialization(e.to_string())) })?; - self.send(message) - .await - .map_err(|e| RequestError::Send(format!("{e:?}"))) + self.send(message).await.map_err(RequestError::from) } } @@ -111,9 +109,7 @@ pub trait IpcClientExt: IpcClient { RequestError::Rpc(RpcError::RequestSerialization(e.to_string())) })?; - self.send(message) - .await - .map_err(|e| RequestError::Send(format!("{e:?}")))?; + self.send(message).await.map_err(RequestError::from)?; let response = loop { let received = response_subscription diff --git a/crates/bitwarden-ipc/src/lib.rs b/crates/bitwarden-ipc/src/lib.rs index 0a5a1485fb..bbc71921a4 100644 --- a/crates/bitwarden-ipc/src/lib.rs +++ b/crates/bitwarden-ipc/src/lib.rs @@ -17,9 +17,12 @@ mod traits; #[cfg(feature = "wasm")] pub mod wasm; +#[cfg(any(test, feature = "test-support"))] +pub use crypto_provider::noise::crypto_provider::{NoiseCryptoProvider, NoiseCryptoProviderState}; pub use endpoint::{Endpoint, HostId, Source}; pub use error::{ - IpcErrorKind, ReceiveError, RequestError, SendError, SubscribeError, TypedReceiveError, + ErrorKind, IpcErrorKind, ReceiveError, RequestError, SendError, SubscribeError, + TypedReceiveError, }; pub use ipc_client::{IpcClientImpl, IpcClientSubscription, IpcClientTypedSubscription}; pub use ipc_client_ext::IpcClientExt; @@ -31,9 +34,12 @@ pub use message::{ pub use rpc::exec::handler::ErasedRpcHandler; pub use rpc::{exec::handler::RpcHandler, request::RpcRequest}; #[cfg(any(test, feature = "test-support"))] -pub use traits::NoEncryptionCryptoProvider; -#[cfg(any(test, feature = "test-support"))] pub use traits::TestCommunicationBackend; +#[cfg(any(test, feature = "test-support"))] +pub use traits::{ + CommunicationBackend, CommunicationBackendReceiver, NoEncryptionCryptoProvider, + SessionRepository, +}; pub use traits::{InMemorySessionRepository, NoopCommunicationBackend}; // Test configuration of the IPC client, always available in test and test-support contexts. diff --git a/crates/bitwarden-ipc/src/traits/communication_backend.rs b/crates/bitwarden-ipc/src/traits/communication_backend.rs index a034afcc76..6ae062cb78 100644 --- a/crates/bitwarden-ipc/src/traits/communication_backend.rs +++ b/crates/bitwarden-ipc/src/traits/communication_backend.rs @@ -9,14 +9,16 @@ use crate::{ /// It is up to the platform to implement this trait and any necessary thread synchronization and /// broadcasting. pub trait CommunicationBackend: Send + Sync + 'static { + /// Error returned by [`send`](Self::send), classified via [`IpcErrorKind`]. type SendError: Debug + Send + Sync + 'static + IpcErrorKind; + /// Receiver type returned by [`subscribe`](Self::subscribe). type Receiver: CommunicationBackendReceiver; /// Send a message to the destination specified in the message. This function may be called /// from any thread at any time. /// /// Both recoverable and fatal errors may be returned, classified via - /// [`IpcErrorKind::is_fatal()`]. A recoverable error (e.g. a transient transport failure) is + /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. a transient transport failure) is /// logged and the IPC client keeps running; a fatal error stops the client from processing any /// further messages. Ambiguous cases should be classified as recoverable. /// @@ -44,12 +46,13 @@ pub trait CommunicationBackend: Send + Sync + 'static { /// receive(). /// - The receiver buffers messages between calls to receive(). pub trait CommunicationBackendReceiver: Send + Sync + 'static { + /// Error returned by [`receive`](Self::receive), classified via [`IpcErrorKind`]. type ReceiveError: Debug + Send + Sync + 'static + IpcErrorKind; /// Receive a message. This function will block asynchronously until a message is received. /// /// Both recoverable and fatal errors may be returned, classified via - /// [`IpcErrorKind::is_fatal()`]. A recoverable error (e.g. the receiver lagging) is logged and + /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. the receiver lagging) is logged and /// the IPC client's processing loop continues; a fatal error (e.g. the channel being closed) /// stops the loop. Ambiguous cases should be classified as recoverable. /// diff --git a/crates/bitwarden-ipc/src/traits/crypto_provider.rs b/crates/bitwarden-ipc/src/traits/crypto_provider.rs index f153a67f3c..3c0e46e16d 100644 --- a/crates/bitwarden-ipc/src/traits/crypto_provider.rs +++ b/crates/bitwarden-ipc/src/traits/crypto_provider.rs @@ -25,7 +25,7 @@ where /// original message. The implementation of this function should handle this logic. /// /// Both recoverable and fatal errors may be returned, classified via - /// [`IpcErrorKind::is_fatal()`]. A recoverable error (e.g. a handshake timeout or a transient + /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. a handshake timeout or a transient /// transport failure) is logged and the IPC client keeps running; a fatal error (e.g. the /// session storage being inaccessible) stops the client from processing any further messages. /// Ambiguous cases should be classified as recoverable. @@ -45,7 +45,7 @@ where /// logic. /// /// Both recoverable and fatal errors may be returned, classified via - /// [`IpcErrorKind::is_fatal()`]. A recoverable error (e.g. a malformed frame from one peer or a + /// [`IpcErrorKind::kind()`]. A recoverable error (e.g. a malformed frame from one peer or a /// transient transport failure) is logged and the IPC client's processing loop continues; a /// fatal error (e.g. the session storage being inaccessible) stops the loop. Ambiguous cases /// should be classified as recoverable. diff --git a/crates/bitwarden-ipc/src/traits/session_repository.rs b/crates/bitwarden-ipc/src/traits/session_repository.rs index 3d2ca230c9..9766551241 100644 --- a/crates/bitwarden-ipc/src/traits/session_repository.rs +++ b/crates/bitwarden-ipc/src/traits/session_repository.rs @@ -4,20 +4,28 @@ use tokio::sync::RwLock; use crate::endpoint::Endpoint; +/// Persists per-destination crypto sessions so they survive across sends and, where the +/// implementation is durable, across restarts. pub trait SessionRepository: Send + Sync + 'static { + /// Error returned when a session could not be read. type GetError: Debug + Send + Sync + 'static; + /// Error returned when a session could not be persisted. type SaveError: Debug + Send + Sync + 'static; + /// Error returned when a session could not be removed. type RemoveError: Debug + Send + Sync + 'static; + /// Load the session for the given destination, if one exists. fn get( &self, destination: Endpoint, ) -> impl std::future::Future, Self::GetError>> + Send + Sync; + /// Store (or overwrite) the session for the given destination. fn save( &self, destination: Endpoint, session: Session, ) -> impl std::future::Future> + Send + Sync; + /// Remove the session for the given destination, if one exists. fn remove( &self, destination: Endpoint, diff --git a/crates/bitwarden-ipc/src/wasm/communication_backend.rs b/crates/bitwarden-ipc/src/wasm/communication_backend.rs index bec1ffa96f..aff3d7398a 100644 --- a/crates/bitwarden-ipc/src/wasm/communication_backend.rs +++ b/crates/bitwarden-ipc/src/wasm/communication_backend.rs @@ -7,7 +7,7 @@ use tokio::sync::RwLock; use wasm_bindgen::prelude::*; use crate::{ - error::IpcErrorKind, + error::{ErrorKind, IpcErrorKind}, message::{IncomingMessage, OutgoingMessage}, traits::{CommunicationBackend, CommunicationBackendReceiver}, }; @@ -42,6 +42,11 @@ pub enum WasmCommunicationError { #[error("incoming message channel lagged, {0} messages were dropped")] Lagged(u64), + /// The destination is not reachable: the JS backend reported "Destination unreachable" (e.g. + /// the desktop app is not connected). + #[error("Destination unreachable")] + Unreachable, + /// The communication channel was closed because all senders were dropped. This is fatal: the /// IPC client's processing loop stops cleanly instead of busy-looping on the closed channel. #[error("incoming message channel closed")] @@ -49,8 +54,12 @@ pub enum WasmCommunicationError { } impl IpcErrorKind for WasmCommunicationError { - fn is_fatal(&self) -> bool { - matches!(self, WasmCommunicationError::Closed) + fn kind(&self) -> ErrorKind { + match self { + WasmCommunicationError::Unreachable => ErrorKind::Unreachable, + WasmCommunicationError::Closed => ErrorKind::Fatal, + _ => ErrorKind::Other, + } } } @@ -132,7 +141,16 @@ impl CommunicationBackend for JsCommunicationBackend { }) .await .map_err(|e| WasmCommunicationError::Js(e.to_string()))? - .map_err(WasmCommunicationError::Js) + .map_err(|message| { + // The TS backend throws `Error("Destination unreachable")` when the peer transport + // is not connected. Map it to a dedicated variant so callers can suppress logging + // for this expected case while still surfacing every other send failure. + if message.contains("Destination unreachable") { + WasmCommunicationError::Unreachable + } else { + WasmCommunicationError::Js(message) + } + }) } async fn subscribe(&self) -> Self::Receiver { @@ -187,7 +205,7 @@ mod tests { let error = receiver.receive().await.unwrap_err(); assert!(matches!(error, WasmCommunicationError::Closed)); - assert!(error.is_fatal()); + assert_eq!(error.kind(), ErrorKind::Fatal); } #[tokio::test] @@ -202,6 +220,6 @@ mod tests { let error = receiver.receive().await.unwrap_err(); assert!(matches!(error, WasmCommunicationError::Lagged(_))); - assert!(!error.is_fatal()); + assert_ne!(error.kind(), ErrorKind::Fatal); } } diff --git a/crates/bitwarden-shared-unlock/Cargo.toml b/crates/bitwarden-shared-unlock/Cargo.toml index 5310f8ba29..04a1988756 100644 --- a/crates/bitwarden-shared-unlock/Cargo.toml +++ b/crates/bitwarden-shared-unlock/Cargo.toml @@ -47,7 +47,7 @@ js-sys = { workspace = true } [dev-dependencies] bitwarden-encoding = { workspace = true } bitwarden-ipc = { workspace = true, features = ["test-support"] } -tokio = { workspace = true, features = ["rt"] } +tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } [lints] workspace = true diff --git a/crates/bitwarden-shared-unlock/src/follower.rs b/crates/bitwarden-shared-unlock/src/follower.rs index 7a1fcddee2..94b429fab0 100644 --- a/crates/bitwarden-shared-unlock/src/follower.rs +++ b/crates/bitwarden-shared-unlock/src/follower.rs @@ -1,7 +1,9 @@ use std::{ops::Add, sync::Arc}; use bitwarden_error::bitwarden_error; -use bitwarden_ipc::{Endpoint, IpcClient, IpcClientExt, SubscribeError, TypedIncomingMessage}; +use bitwarden_ipc::{ + Endpoint, IpcClient, IpcClientExt, RequestError, SubscribeError, TypedIncomingMessage, +}; use bitwarden_threading::{cancellation_token, time::sleep}; use thiserror::Error; @@ -238,7 +240,20 @@ impl Follower { async fn send_message(&self, message: FollowerMessage, recipient: Endpoint) { if let Err(error) = self.0.ipc_client.send_typed(message, recipient).await { - tracing::error!(?error, "Failed to send shared unlock IPC message"); + match error { + RequestError::Unreachable => { + // The leader is not connected; the message simply could not be delivered. + } + RequestError::Timeout(_) => { + tracing::warn!( + ?error, + "Timeout sending shared unlock follower message to leader" + ); + } + _ => { + tracing::error!(?error, "Failed to send shared unlock IPC message"); + } + } } } } diff --git a/crates/bitwarden-shared-unlock/src/tests.rs b/crates/bitwarden-shared-unlock/src/tests.rs index 7477af9041..68fc556101 100644 --- a/crates/bitwarden-shared-unlock/src/tests.rs +++ b/crates/bitwarden-shared-unlock/src/tests.rs @@ -560,6 +560,306 @@ async fn test_web_source_lock_state_update_with_matching_origin_is_accepted() { ); } +/// Tests that exercise the full encrypted (Noise) transport, including reconnection after the +/// leader has been offline. Unlike the manual-pump harness above (which uses +/// [`NoEncryptionCryptoProvider`] and hand-delivers already-decoded payloads), these tests run real +/// [`IpcClient`] receive loops over a controllable in-memory link so the Noise handshake and +/// session-cleanup-on-send-failure logic are actually driven. +mod reconnect { + use std::sync::atomic::{AtomicBool, Ordering}; + + use bitwarden_ipc::{ + CommunicationBackend, CommunicationBackendReceiver, Endpoint, ErrorKind, IpcClient, + IpcClientImpl, IpcErrorKind, NoiseCryptoProvider, NoiseCryptoProviderState, + OutgoingMessage, + }; + use bitwarden_threading::{cancellation_token::CancellationToken, time::sleep}; + use tokio::sync::{Mutex, RwLock, broadcast}; + + use super::*; + + /// Send error for [`Link`]. When the link is offline the destination is unreachable, which the + /// Noise provider treats as a benign, non-fatal condition. + #[derive(Debug)] + enum LinkError { + Unreachable, + } + + impl IpcErrorKind for LinkError { + fn kind(&self) -> ErrorKind { + match self { + LinkError::Unreachable => ErrorKind::Unreachable, + } + } + } + + /// One end of a bidirectional in-memory transport between two peers. `send` pushes onto the + /// peer's incoming channel and `subscribe` reads from our own; a shared `online` flag lets a + /// test simulate the peer going offline (sends then fail as unreachable). + #[derive(Clone)] + struct Link { + online: Arc, + /// The peer's incoming channel: what our sends are delivered into. + to_peer: broadcast::Sender, + /// Our own incoming channel: receivers subscribe to this. + from_peer: broadcast::Sender, + /// Identity stamped onto messages we receive (i.e. the peer's source). + peer_source: Source, + } + + struct LinkReceiver { + incoming: Mutex>, + peer_source: Source, + } + + impl Link { + /// Create a connected pair of links plus the shared online flag that controls both. + fn pair() -> (Link, Link, Arc) { + let online = Arc::new(AtomicBool::new(true)); + let (follower_incoming, _) = broadcast::channel(256); + let (leader_incoming, _) = broadcast::channel(256); + + let follower = Link { + online: online.clone(), + to_peer: leader_incoming.clone(), + from_peer: follower_incoming.clone(), + // The follower only ever talks to the leader (desktop). + peer_source: Source::DesktopMain, + }; + let leader = Link { + online: online.clone(), + to_peer: follower_incoming, + from_peer: leader_incoming, + peer_source: follower_source(), + }; + (follower, leader, online) + } + } + + impl CommunicationBackend for Link { + type SendError = LinkError; + type Receiver = LinkReceiver; + + async fn send(&self, message: OutgoingMessage) -> Result<(), Self::SendError> { + if !self.online.load(Ordering::SeqCst) { + return Err(LinkError::Unreachable); + } + // A missing receiver just means the message is dropped, not that the peer is + // unreachable, so the error is intentionally ignored. + let _ = self.to_peer.send(message); + Ok(()) + } + + async fn subscribe(&self) -> Self::Receiver { + LinkReceiver { + incoming: Mutex::new(self.from_peer.subscribe()), + peer_source: self.peer_source.clone(), + } + } + } + + impl CommunicationBackendReceiver for LinkReceiver { + type ReceiveError = LinkError; + + async fn receive(&self) -> Result { + loop { + let received = { self.incoming.lock().await.recv().await }; + match received { + Ok(message) => { + return Ok(IncomingMessage { + payload: message.payload, + destination: message.destination, + source: self.peer_source.clone(), + topic: message.topic, + }); + } + Err(broadcast::error::RecvError::Lagged(_)) => continue, + // The channel is only closed when the harness is torn down; block forever + // rather than reporting a fatal error mid-test. + Err(broadcast::error::RecvError::Closed) => std::future::pending().await, + } + } + } + } + + /// A session repository whose backing store can be inspected and cleared by the test, used to + /// simulate a leader that restarted (and thus lost its Noise session) while offline. + #[derive(Clone, Default)] + struct SharedSessions(Arc>>); + + impl bitwarden_ipc::SessionRepository for SharedSessions { + type GetError = (); + type SaveError = (); + type RemoveError = (); + + async fn get( + &self, + destination: Endpoint, + ) -> Result, Self::GetError> { + Ok(self.0.read().await.get(&destination).cloned()) + } + + async fn save( + &self, + destination: Endpoint, + session: NoiseCryptoProviderState, + ) -> Result<(), Self::SaveError> { + self.0.write().await.insert(destination, session); + Ok(()) + } + + async fn remove(&self, destination: Endpoint) -> Result<(), Self::RemoveError> { + self.0.write().await.remove(&destination); + Ok(()) + } + } + + /// Poll a driver's lock state until it matches `expected` or a generous timeout elapses. + async fn wait_for_state(driver: &MockDriver, user: UserId, expected: &LockState) -> bool { + for _ in 0..250 { + if &driver.get_state(user) == expected { + return true; + } + sleep(Duration::from_millis(20)).await; + } + &driver.get_state(user) == expected + } + + /// A leader + follower connected over an encrypted link, each running real IPC receive loops. + struct EncryptedHarness { + follower: Follower, + leader_lock: MockDriver, + leader_sessions: SharedSessions, + online: Arc, + // Retained for the harness's lifetime; the spawned IPC/leader tasks are torn down when the + // test's tokio runtime shuts down at the end of the test. + _token: CancellationToken, + } + + impl EncryptedHarness { + async fn new( + leader_states: HashMap, + follower_driver: MockDriver, + ) -> Self { + let (follower_link, leader_link, online) = Link::pair(); + let token = CancellationToken::new(); + + // Leader: an IPC client whose internal loop performs the Noise handshake, plus the + // shared-unlock leader that reacts to decoded follower messages. + let leader_lock = MockDriver::new(leader_states); + let leader_sessions = SharedSessions::default(); + let leader_ipc: Arc = Arc::new(IpcClientImpl::new( + NoiseCryptoProvider, + leader_link, + leader_sessions.clone(), + )); + leader_ipc + .start(Some(token.clone())) + .await + .expect("Leader IPC client should start"); + let leader = Leader::create(leader_lock.clone(), leader_ipc); + leader + .start(Some(token.clone())) + .await + .expect("Leader should start"); + + // Follower: only ever sends, so its IPC client does not need its own receive loop. + let follower_ipc: Arc = Arc::new(IpcClientImpl::new( + NoiseCryptoProvider, + follower_link, + bitwarden_ipc::InMemorySessionRepository::::new( + HashMap::new(), + ), + )); + let follower = Follower::create(follower_driver, follower_ipc); + + Self { + follower, + leader_lock, + leader_sessions, + online, + _token: token, + } + } + + fn set_online(&self, online: bool) { + self.online.store(online, Ordering::SeqCst); + } + } + + /// Full reconnection cycle: + /// 1. The follower shares its unlocked state; the leader unlocks. + /// 2. The leader goes offline. The follower's next send fails, which clears the follower's + /// Noise session. The (restarted) leader loses its session too. + /// 3. The leader comes back online. The follower's first send re-handshakes and successfully + /// re-shares the unlock without needing a second attempt. + #[tokio::test] + async fn test_reconnect_after_leader_offline_reshares_unlock() { + let user = user_a(); + let key = test_user_key(); + + // Leader starts locked; the follower is unlocked and will share its state. + let follower_driver = MockDriver::new(HashMap::from([( + user, + LockState::Unlocked { + user_key: key.clone(), + }, + )])); + let harness = + EncryptedHarness::new(HashMap::from([(user, LockState::Locked)]), follower_driver) + .await; + + // --- 1. Share unlock over a freshly established encrypted session --- + harness.follower.start_sessions().await; + assert!( + wait_for_state( + &harness.leader_lock, + user, + &LockState::Unlocked { + user_key: key.clone() + } + ) + .await, + "Leader should unlock after the follower shares its unlocked state" + ); + + // --- 2. Leader goes offline; the follower's send fails and clears its crypto state --- + harness.set_online(false); + // Simulate the leader relocking and losing its Noise session (e.g. a restart) while down. + harness + .leader_lock + .states + .lock() + .unwrap() + .insert(user, LockState::Locked); + harness.leader_sessions.0.write().await.clear(); + + // This send cannot be delivered; the Noise provider discards the follower's session so the + // next send is forced to re-handshake. The follower swallows the unreachable error. + harness.follower.start_sessions().await; + assert_eq!( + harness.leader_lock.get_state(user), + LockState::Locked, + "Leader must stay locked while offline" + ); + + // --- 3. Leader is back; the first reconnect attempt re-handshakes and re-shares unlock --- + harness.set_online(true); + harness.follower.start_sessions().await; + assert!( + wait_for_state( + &harness.leader_lock, + user, + &LockState::Unlocked { + user_key: key.clone() + } + ) + .await, + "Leader should unlock again after the follower reconnects on the first attempt" + ); + } +} + #[tokio::test] async fn test_non_web_source_skips_origin_validation() { let user = user_a();