[PM-39294] Gate shared unlock handshakes on transport reachability

Cargo.toml

@@ -100,7 +100,7 @@ reqwest-middleware = { version = "0.5.1", features = [
     "form",
     "query",
 ] }
-rsa = "0.10.0-rc.17"
+rsa = "0.10.0-rc.18"
 rustls = { version = "0.23.27", default-features = false, features = [
     "std",
     "tls12",
@@ -202,3 +202,9 @@ opt-level = 3
 # Stripping the binary reduces the size by ~30%, but the stacktraces won't be usable anymore.
 # This is fine as long as we don't have any unhandled panics, but let's keep it disabled for now
 # strip = true
+
+# The crates.io 0.7.x releases of crypto-bigint were yanked after 0.7.5 shipped a bugfix, so
+# stale registry indexes can fail to resolve `crypto-bigint = "^0.7"` (required by rsa). Override
+# to the published 0.7.5 from git to keep the build resolving regardless of index state.
+[patch.crates-io]
+crypto-bigint = { git = "https://github.com/RustCrypto/crypto-bigint", tag = "v0.7.5" }

crates/bitwarden-ipc/src/control.rs

@@ -0,0 +1,81 @@
+//! The IPC control plane: reserved topics for transport-level frames that travel over the raw
+//! transport and deliberately bypass the crypto channel (reachability ping/pong).
+//!
+//! Frames whose topic falls under [`CONTROL_TOPIC_PREFIX`] are peeled off by the
+//! [`ControlSplitter`](crate::control_splitter::ControlSplitter) before the crypto layer ever sees
+//! them, so they can measure liveness without (and without disturbing) a crypto session.
+
+use crate::{
+    endpoint::{Endpoint, Source},
+    message::OutgoingMessage,
+};
+
+/// Reserved topic namespace for transport control frames that bypass the crypto channel.
+pub(crate) const CONTROL_TOPIC_PREFIX: &str = "$bw.control.";
+/// Topic marking a plaintext reachability ping.
+pub(crate) const CONTROL_PING_TOPIC: &str = "$bw.control.ping";
+/// Topic marking a plaintext reachability pong (the reply to a ping).
+pub(crate) const CONTROL_PONG_TOPIC: &str = "$bw.control.pong";
+
+/// Whether `topic` belongs to the reserved control-topic namespace and must bypass crypto.
+pub(crate) fn is_control_topic(topic: Option<&str>) -> bool {
+    topic.is_some_and(|t| t.starts_with(CONTROL_TOPIC_PREFIX))
+}
+
+/// A transport control-plane message. These ride the reserved control topics over the raw
+/// transport rather than the crypto channel.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ControlMessage {
+    /// Liveness probe.
+    Ping,
+    /// Reply to a [`ControlMessage::Ping`].
+    Pong,
+}
+
+impl ControlMessage {
+    /// The reserved topic this message rides on.
+    pub(crate) fn topic(self) -> &'static str {
+        match self {
+            ControlMessage::Ping => CONTROL_PING_TOPIC,
+            ControlMessage::Pong => CONTROL_PONG_TOPIC,
+        }
+    }
+
+    /// Parse a control message from a frame's topic, returning `None` for non-control frames.
+    pub(crate) fn from_topic(topic: Option<&str>) -> Option<Self> {
+        match topic {
+            Some(CONTROL_PING_TOPIC) => Some(ControlMessage::Ping),
+            Some(CONTROL_PONG_TOPIC) => Some(ControlMessage::Pong),
+            _ => None,
+        }
+    }
+
+    /// Build the outgoing frame carrying this control message to `destination`.
+    pub(crate) fn to_outgoing(self, destination: Endpoint) -> OutgoingMessage {
+        OutgoingMessage {
+            payload: Vec::new(),
+            destination,
+            topic: Some(self.topic().to_owned()),
+        }
+    }
+}
+
+/// An inbound [`ControlMessage`] together with the peer it came from.
+pub(crate) struct IncomingControlMessage {
+    pub(crate) message: ControlMessage,
+    pub(crate) source: Source,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn is_control_topic_matches_namespace() {
+        assert!(is_control_topic(Some(CONTROL_PING_TOPIC)));
+        assert!(is_control_topic(Some(CONTROL_PONG_TOPIC)));
+        assert!(is_control_topic(Some("$bw.control.future")));
+        assert!(!is_control_topic(Some("some-rpc-topic")));
+        assert!(!is_control_topic(None));
+    }
+}

crates/bitwarden-ipc/src/control_splitter.rs

@@ -0,0 +1,105 @@
+//! A [`CommunicationBackend`] decorator that keeps reachability control frames off the crypto
+//! channel.
+//!
+//! Reachability ping/pong frames travel over the raw transport (so they can measure liveness
+//! without a crypto session), but the crypto layer reads the same transport and would try to decode
+//! them as Noise frames, aborting in-flight handshakes. This wrapper splits the inbound stream in
+//! two: the receivers handed to the crypto layer ([`CommunicationBackend::subscribe`]) drop control
+//! frames, while [`ControlSplitter::subscribe_control`] yields only the control messages. `send`
+//! and `reachability` pass straight through, so the crypto provider stays entirely unaware of
+//! reachability.
+
+use std::sync::Arc;
+
+use crate::{
+    control::{ControlMessage, IncomingControlMessage, is_control_topic},
+    endpoint::Endpoint,
+    message::{IncomingMessage, OutgoingMessage},
+    traits::{CommunicationBackend, CommunicationBackendReceiver, Reachability},
+};
+
+/// Wraps a communication backend so reachability control frames are separated from data frames. The
+/// crypto layer only ever sees data; the reachability tracker consumes the control stream.
+// Public (but hidden) only because it appears in `IpcClientImpl`'s public trait bounds: the client
+// hands this wrapper to the crypto provider so control frames never reach it. It is not part of the
+// supported API and should not be used directly.
+#[doc(hidden)]
+pub struct ControlSplitter<Com> {
+    backend: Arc<Com>,
+}
+
+impl<Com: CommunicationBackend> ControlSplitter<Com> {
+    pub(crate) fn new(backend: Arc<Com>) -> Self {
+        Self { backend }
+    }
+
+    /// Subscribe to the inbound control-plane stream. Mirrors [`CommunicationBackendReceiver`]:
+    /// call [`ControlReceiver::receive`] for the next control message. Data frames are filtered
+    /// out (they reach the crypto layer via [`CommunicationBackend::subscribe`]).
+    pub(crate) async fn subscribe_control(&self) -> ControlReceiver<Com::Receiver> {
+        ControlReceiver {
+            inner: self.backend.subscribe().await,
+        }
+    }
+}
+
+impl<Com: CommunicationBackend> CommunicationBackend for ControlSplitter<Com> {
+    type SendError = Com::SendError;
+    type Receiver = ControlSplitterReceiver<Com::Receiver>;
+
+    async fn send(&self, message: OutgoingMessage) -> Result<(), Self::SendError> {
+        self.backend.send(message).await
+    }
+
+    async fn subscribe(&self) -> Self::Receiver {
+        ControlSplitterReceiver {
+            inner: self.backend.subscribe().await,
+        }
+    }
+
+    async fn reachability(&self, endpoint: &Endpoint) -> Reachability {
+        self.backend.reachability(endpoint).await
+    }
+}
+
+/// Data-frame receiver handed to the crypto layer: drops control-topic frames so crypto never sees
+/// them.
+#[doc(hidden)]
+pub struct ControlSplitterReceiver<R> {
+    inner: R,
+}
+
+impl<R: CommunicationBackendReceiver> CommunicationBackendReceiver for ControlSplitterReceiver<R> {
+    type ReceiveError = R::ReceiveError;
+
+    async fn receive(&self) -> Result<IncomingMessage, Self::ReceiveError> {
+        loop {
+            let message = self.inner.receive().await?;
+            if is_control_topic(message.topic.as_deref()) {
+                continue;
+            }
+            return Ok(message);
+        }
+    }
+}
+
+/// Control-frame receiver: yields inbound [`ControlMessage`]s, skipping data frames.
+pub(crate) struct ControlReceiver<R> {
+    inner: R,
+}
+
+impl<R: CommunicationBackendReceiver> ControlReceiver<R> {
+    /// Receive the next control message. Blocks asynchronously, skipping data frames, until a
+    /// control frame arrives or the underlying receiver errors.
+    pub(crate) async fn receive(&self) -> Result<IncomingControlMessage, R::ReceiveError> {
+        loop {
+            let message = self.inner.receive().await?;
+            if let Some(control) = ControlMessage::from_topic(message.topic.as_deref()) {
+                return Ok(IncomingControlMessage {
+                    message: control,
+                    source: message.source,
+                });
+            }
+        }
+    }
+}

crates/bitwarden-ipc/src/ipc_client.rs

@@ -7,6 +7,7 @@ use tokio::select;
 
 use crate::{
     constants::CHANNEL_BUFFER_CAPACITY,
+    control_splitter::ControlSplitter,
     error::{
         AlreadyRunningError, IpcErrorKind, ReceiveError, SendError, SubscribeError,
         TypedReceiveError,
@@ -15,6 +16,7 @@ use crate::{
         IncomingMessage, OutgoingMessage, PayloadTypeName, TypedIncomingMessage,
         TypedOutgoingMessage,
     },
+    reachability::ReachabilityTracker,
     rpc::{
         exec::{handler::ErasedRpcHandler, handler_registry::RpcHandlerRegistry},
         request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestPayload},
@@ -45,13 +47,17 @@ pub struct IpcClientTypedSubscription<Payload: DeserializeOwned + PayloadTypeNam
 /// Internal shared state for the IPC client.
 struct IpcClientInner<Crypto, Com, Ses>
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
     crypto: Crypto,
-    communication: Com,
+    /// The raw backend wrapped so reachability control frames bypass the crypto layer.
+    communication: ControlSplitter<Com>,
     sessions: Ses,
+    /// Owns the per-endpoint ping loops and the auto-pong responder; handed to consumers via
+    /// [`IpcClient::reachability`](crate::IpcClient::reachability).
+    reachability: Arc<ReachabilityTracker>,
 
     handlers: RpcHandlerRegistry,
     incoming: Mutex<Option<tokio::sync::broadcast::Receiver<IncomingMessage>>>,
@@ -65,7 +71,7 @@ where
 /// This is the concrete implementation of the [`IpcClient`](crate::IpcClient) trait.
 pub struct IpcClientImpl<Crypto, Com, Ses>
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
@@ -74,7 +80,7 @@ where
 
 impl<Crypto, Com, Ses> Clone for IpcClientImpl<Crypto, Com, Ses>
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
@@ -87,18 +93,22 @@ where
 
 impl<Crypto, Com, Ses> IpcClientImpl<Crypto, Com, Ses>
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
     /// Create a new IPC client with the provided crypto provider, communication backend, and
     /// session repository.
     pub fn new(crypto: Crypto, communication: Com, sessions: Ses) -> Self {
+        let backend = Arc::new(communication);
+        let reachability = Arc::new(ReachabilityTracker::from_backend(backend.clone()));
+        let communication = ControlSplitter::new(backend);
         Self {
             inner: Arc::new(IpcClientInner {
                 crypto,
                 communication,
                 sessions,
+                reachability,
 
                 handlers: RpcHandlerRegistry::new(),
                 incoming: Mutex::new(None),
@@ -111,7 +121,7 @@ where
 #[async_trait::async_trait]
 impl<Crypto, Com, Ses> crate::ipc_client_trait::IpcClient for IpcClientImpl<Crypto, Com, Ses>
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
@@ -130,6 +140,14 @@ where
             .expect("Failed to lock cancellation token mutex")
             .replace(cancellation_token.clone());
 
+        // Feed the reachability tracker from the control-frame stream (auto-pong + liveness
+        // recording). It shares the client's cancellation token so it stops on shutdown rather than
+        // leaking, and runs even when nothing is tracked so passive peers still answer pings.
+        let control_receiver = self.inner.communication.subscribe_control().await;
+        self.inner
+            .reachability
+            .start(control_receiver, cancellation_token.clone());
+
         let com_receiver = self.inner.communication.subscribe().await;
         let (client_tx, client_rx) = tokio::sync::broadcast::channel(CHANNEL_BUFFER_CAPACITY);
 
@@ -246,11 +264,15 @@ where
             .register_erased(name.to_owned(), handler)
             .await;
     }
+
+    fn reachability(&self) -> Arc<ReachabilityTracker> {
+        self.inner.reachability.clone()
+    }
 }
 
 fn stop_inner<Crypto, Com, Ses>(inner: &IpcClientInner<Crypto, Com, Ses>)
 where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
@@ -268,7 +290,7 @@ fn handle_rpc_request<Crypto, Com, Ses>(
     inner: &Arc<IpcClientInner<Crypto, Com, Ses>>,
     incoming_message: IncomingMessage,
 ) where
-    Crypto: CryptoProvider<Com, Ses>,
+    Crypto: CryptoProvider<ControlSplitter<Com>, Ses>,
     Com: CommunicationBackend,
     Ses: SessionRepository<Crypto::Session>,
 {
@@ -426,15 +448,17 @@ mod tests {
     }
 
     type TestSessionRepository = InMemorySessionRepository<String>;
-    impl CryptoProvider<TestCommunicationBackend, TestSessionRepository> for TestCryptoProvider {
+    // Generic over the backend so it satisfies the client's `CryptoProvider<ControlSplitter<Com>>`
+    // bound as well as a bare backend in other tests.
+    impl<Com: CommunicationBackend> CryptoProvider<Com, TestSessionRepository> for TestCryptoProvider {
         type Session = String;
         type SendError = TestCryptoError;
         type ReceiveError = TestCryptoError;
 
         async fn receive(
             &self,
-            _receiver: &<TestCommunicationBackend as CommunicationBackend>::Receiver,
-            _communication: &TestCommunicationBackend,
+            _receiver: &Com::Receiver,
+            _communication: &Com,
             _sessions: &TestSessionRepository,
         ) -> Result<IncomingMessage, Self::ReceiveError> {
             match &self.receive_result {
@@ -459,7 +483,7 @@ mod tests {
 
         async fn send(
             &self,
-            _communication: &TestCommunicationBackend,
+            _communication: &Com,
             _sessions: &TestSessionRepository,
             _message: OutgoingMessage,
         ) -> Result<(), Self::SendError> {

crates/bitwarden-ipc/src/ipc_client_trait.rs

@@ -1,9 +1,12 @@
+use std::sync::Arc;
+
 use bitwarden_threading::cancellation_token::CancellationToken;
 
 use crate::{
     error::{AlreadyRunningError, SendError, SubscribeError},
     ipc_client::IpcClientSubscription,
     message::OutgoingMessage,
+    reachability::ReachabilityTracker,
     rpc::exec::handler::ErasedRpcHandler,
 };
 
@@ -46,4 +49,11 @@ pub trait IpcClient: Send + Sync {
     /// instead.
     #[doc(hidden)]
     async fn register_rpc_handler_erased(&self, name: &str, handler: Box<dyn ErasedRpcHandler>);
+
+    /// The reachability tracker for this client.
+    ///
+    /// Consumers that need to know whether a peer is reachable call
+    /// [`ReachabilityTracker::track`] with the peer's endpoint (typically discovered at runtime)
+    /// and hold the returned handle for as long as they care; dropping it stops tracking.
+    fn reachability(&self) -> Arc<ReachabilityTracker>;
 }