diff --git a/crates/matrix-sdk-crypto/src/store/crypto_store_wrapper.rs b/crates/matrix-sdk-crypto/src/store/crypto_store_wrapper.rs index 2362c6a31..3ceea1907 100644 --- a/crates/matrix-sdk-crypto/src/store/crypto_store_wrapper.rs +++ b/crates/matrix-sdk-crypto/src/store/crypto_store_wrapper.rs @@ -8,7 +8,10 @@ use tokio::sync::{broadcast, Mutex}; use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}; use tracing::{debug, trace, warn}; -use super::{caches::SessionStore, DeviceChanges, IdentityChanges, LockableCryptoStore}; +use super::{ + caches::SessionStore, types::RoomKeyBundleInfo, DeviceChanges, IdentityChanges, + LockableCryptoStore, +}; use crate::{ olm::InboundGroupSession, store, @@ -46,6 +49,10 @@ pub(crate) struct CryptoStoreWrapper { /// identities which got updated or newly created. identities_broadcaster: broadcast::Sender<(Option, IdentityChanges, DeviceChanges)>, + + /// The sender side of a broadcast channel which sends out information about + /// historic room key bundles we have received. + historic_room_key_bundles_broadcaster: broadcast::Sender, } impl CryptoStoreWrapper { @@ -56,6 +63,7 @@ impl CryptoStoreWrapper { // The identities broadcaster is responsible for user identities as well as // devices, that's why we increase the capacity here. let identities_broadcaster = broadcast::Sender::new(20); + let historic_room_key_bundles_broadcaster = broadcast::Sender::new(10); Self { user_id: user_id.to_owned(), @@ -66,6 +74,7 @@ impl CryptoStoreWrapper { room_keys_withheld_received_sender, secrets_broadcaster, identities_broadcaster, + historic_room_key_bundles_broadcaster, } } @@ -107,6 +116,8 @@ impl CryptoStoreWrapper { let secrets = changes.secrets.to_owned(); let devices = changes.devices.to_owned(); let identities = changes.identities.to_owned(); + let room_key_bundle_updates: Vec<_> = + changes.received_room_key_bundles.iter().map(RoomKeyBundleInfo::from).collect(); if devices .changed @@ -159,6 +170,10 @@ impl CryptoStoreWrapper { let _ = self.secrets_broadcaster.send(secret); } + for bundle_info in room_key_bundle_updates { + let _ = self.historic_room_key_bundles_broadcaster.send(bundle_info); + } + if !devices.is_empty() || !identities.is_empty() { // Mapping the devices and user identities from the read-only variant to one's // that contain side-effects requires our own identity. This is @@ -330,6 +345,13 @@ impl CryptoStoreWrapper { Self::filter_errors_out_of_stream(stream, "secrets_stream") } + /// Receive notifications of historic room key bundles being received and + /// stored in the store as a [`Stream`]. + pub fn historic_room_key_stream(&self) -> impl Stream { + let stream = BroadcastStream::new(self.historic_room_key_bundles_broadcaster.subscribe()); + Self::filter_errors_out_of_stream(stream, "bundle_stream") + } + /// Returns a stream of newly created or updated cryptographic identities. /// /// This is just a helper method which allows us to build higher level diff --git a/crates/matrix-sdk-crypto/src/store/mod.rs b/crates/matrix-sdk-crypto/src/store/mod.rs index dae0d95ec..d44a138bf 100644 --- a/crates/matrix-sdk-crypto/src/store/mod.rs +++ b/crates/matrix-sdk-crypto/src/store/mod.rs @@ -60,6 +60,7 @@ use thiserror::Error; use tokio::sync::{Mutex, Notify, OwnedRwLockWriteGuard, RwLock}; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; use tracing::{error, info, instrument, trace, warn}; +use types::RoomKeyBundleInfo; use vodozemac::{megolm::SessionOrdering, Curve25519PublicKey}; use self::types::{ @@ -1313,6 +1314,52 @@ impl Store { self.inner.store.secrets_stream() } + /// Receive notifications of historic room key bundles as a [`Stream`]. + /// + /// Historic room key bundles are defined in [MSC4268](https://github.com/matrix-org/matrix-spec-proposals/pull/4268). + /// + /// Each time a historic room key bundle was received, an update will be + /// sent to the stream. This stream can be used to accept historic room key + /// bundles that arrive out of order, i.e. the bundle arrives after the + /// user has already accepted a room invitation. + /// + /// # Examples + /// + /// ```no_run + /// # use matrix_sdk_crypto::{ + /// # OlmMachine, + /// # store::types::StoredRoomKeyBundleData, + /// # types::room_history::RoomKeyBundle + /// # }; + /// # use ruma::{device_id, user_id}; + /// # use futures_util::{pin_mut, StreamExt}; + /// # let alice = user_id!("@alice:example.org").to_owned(); + /// # async { + /// # let machine = OlmMachine::new(&alice, device_id!("DEVICEID")).await; + /// let bundle_stream = machine.store().historic_room_key_stream(); + /// pin_mut!(bundle_stream); + /// + /// while let Some(bundle_info) = bundle_stream.next().await { + /// // Try to find the bundle content in the store and if it's valid accept it. + /// if let Some(bundle_content) = machine.store().get_received_room_key_bundle_data(&bundle_info.room_id, &bundle_info.sender).await? { + /// let StoredRoomKeyBundleData { sender_user, sender_data, bundle_data } = bundle_content; + /// // Download the bundle now and import it. + /// let bundle: RoomKeyBundle = todo!("Download the bundle"); + /// machine.store().receive_room_key_bundle( + /// &bundle_info.room_id, + /// &sender_user, + /// &sender_data, + /// bundle, + /// |_, _| {}, + /// ).await?; + /// } + /// } + /// # anyhow::Ok(()) }; + /// ``` + pub fn historic_room_key_stream(&self) -> impl Stream { + self.inner.store.historic_room_key_stream() + } + /// Import the given room keys into the store. /// /// # Arguments diff --git a/crates/matrix-sdk-crypto/src/store/types.rs b/crates/matrix-sdk-crypto/src/store/types.rs index b2c490b5c..5bca20b4d 100644 --- a/crates/matrix-sdk-crypto/src/store/types.rs +++ b/crates/matrix-sdk-crypto/src/store/types.rs @@ -478,3 +478,26 @@ pub struct RoomKeyWithheldInfo { /// withheld. pub withheld_event: RoomKeyWithheldEvent, } + +/// Information about a received historic room key bundle. +/// +/// This struct contains information needed to uniquely identify a room key +/// bundle. Only a single bundle per sender for a given room is persisted at a +/// time. +/// +/// It is used to notify listeners about received room key bundles. +#[derive(Debug, Clone)] +pub struct RoomKeyBundleInfo { + /// The user ID of the person that sent us the historic room key bundle. + pub sender: OwnedUserId, + + /// The ID of the room the bundle should be used in. + pub room_id: OwnedRoomId, +} + +impl From<&StoredRoomKeyBundleData> for RoomKeyBundleInfo { + fn from(value: &StoredRoomKeyBundleData) -> Self { + let StoredRoomKeyBundleData { sender_user, sender_data: _, bundle_data } = value; + Self { sender: sender_user.clone(), room_id: bundle_data.room_id.clone() } + } +} diff --git a/crates/matrix-sdk/src/encryption/mod.rs b/crates/matrix-sdk/src/encryption/mod.rs index 6909ab62a..ad16fe619 100644 --- a/crates/matrix-sdk/src/encryption/mod.rs +++ b/crates/matrix-sdk/src/encryption/mod.rs @@ -33,7 +33,7 @@ use futures_util::{ stream::{self, StreamExt}, }; use matrix_sdk_base::crypto::{ - store::types::RoomKeyInfo, + store::types::{RoomKeyBundleInfo, RoomKeyInfo}, types::requests::{ OutgoingRequest, OutgoingVerificationRequest, RoomMessageRequest, ToDeviceRequest, }, @@ -1478,6 +1478,43 @@ impl Encryption { Some(olm.store().room_keys_received_stream()) } + /// Receive notifications of historic room key bundles as a [`Stream`]. + /// + /// Historic room key bundles are defined in [MSC4268](https://github.com/matrix-org/matrix-spec-proposals/pull/4268). + /// + /// Each time a historic room key bundle was received, an update will be + /// sent to the stream. This stream is useful for informative purposes + /// exclusively, historic room key bundles are handled by the SDK + /// automatically. + /// + /// # Examples + /// + /// ```no_run + /// # use matrix_sdk::Client; + /// # use url::Url; + /// # async { + /// # let homeserver = Url::parse("http://example.com")?; + /// # let client = Client::new(homeserver).await?; + /// use futures_util::StreamExt; + /// + /// let Some(mut bundle_stream) = + /// client.encryption().historic_room_key_stream().await + /// else { + /// return Ok(()); + /// }; + /// + /// while let Some(bundle_info) = bundle_stream.next().await { + /// println!("Received a historic room key bundle {bundle_info:?}"); + /// } + /// # anyhow::Ok(()) }; + /// ``` + pub async fn historic_room_key_stream(&self) -> Option> { + let olm = self.client.olm_machine().await; + let olm = olm.as_ref()?; + + Some(olm.store().historic_room_key_stream()) + } + /// Get the secret storage manager of the client. pub fn secret_storage(&self) -> SecretStorage { SecretStorage { client: self.client.to_owned() } diff --git a/crates/matrix-sdk/src/room/shared_room_history.rs b/crates/matrix-sdk/src/room/shared_room_history.rs index 25827d8ed..251036b37 100644 --- a/crates/matrix-sdk/src/room/shared_room_history.rs +++ b/crates/matrix-sdk/src/room/shared_room_history.rs @@ -129,6 +129,8 @@ pub(super) async fn maybe_accept_key_bundle(room: &Room, inviter: &UserId) -> Re else { // No bundle received (yet). // TODO: deal with the bundle arriving later (https://github.com/matrix-org/matrix-rust-sdk/issues/4926) + // We need to check for all them bundles in the store when we create the client + // object and we need to process them when they arrive. return Ok(()); }; diff --git a/testing/matrix-sdk-integration-testing/src/tests/e2ee/shared_history.rs b/testing/matrix-sdk-integration-testing/src/tests/e2ee/shared_history.rs index d88d24b80..cc2238bd1 100644 --- a/testing/matrix-sdk-integration-testing/src/tests/e2ee/shared_history.rs +++ b/testing/matrix-sdk-integration-testing/src/tests/e2ee/shared_history.rs @@ -1,6 +1,9 @@ +use std::ops::Deref; + use anyhow::Result; use assert_matches2::assert_let; use assign::assign; +use futures::{pin_mut, FutureExt, StreamExt}; use matrix_sdk::{ assert_decrypted_message_eq, encryption::EncryptionSettings, @@ -68,6 +71,12 @@ async fn test_history_share_on_invite() -> Result<()> { .expect("We should be able to send a message to the room") .event_id; + let bundle_stream = bob + .encryption() + .historic_room_key_stream() + .await + .expect("We should be able to get the bundle stream"); + // Alice invites Bob to the room alice_room.invite_user_by_id(bob.user_id().unwrap()).await?; @@ -88,6 +97,17 @@ async fn test_history_share_on_invite() -> Result<()> { let bob_room = bob.get_room(alice_room.room_id()).expect("Bob should have received the invite"); + pin_mut!(bundle_stream); + + let info = bundle_stream + .next() + .now_or_never() + .flatten() + .expect("We should be notified about the received bundle"); + + assert_eq!(Some(info.sender.deref()), alice.user_id()); + assert_eq!(info.room_id, alice_room.room_id()); + bob_room .join() .instrument(bob_span.clone())