crypto: Initial scaffolding for incoming key share handling.

matrix_sdk/src/client.rs

@@ -1662,13 +1662,17 @@ impl Client {
     /// ```
     #[cfg(feature = "encryption")]
     #[cfg_attr(feature = "docs", doc(cfg(encryption)))]
-    pub async fn get_device(&self, user_id: &UserId, device_id: &DeviceId) -> Option<Device> {
+    pub async fn get_device(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+    ) -> StdResult<Option<Device>, CryptoStoreError> {
         let device = self.base_client.get_device(user_id, device_id).await?;
 
-        Some(Device {
+        Ok(device.map(|d| Device {
-            inner: device,
+            inner: d,
             http_client: self.http_client.clone(),
-        })
+        }))
     }
 
     /// Get a map holding all the devices of an user.

matrix_sdk_base/src/client.rs

@@ -1809,9 +1809,19 @@ impl BaseClient {
     /// ```
     #[cfg(feature = "encryption")]
     #[cfg_attr(feature = "docs", doc(cfg(encryption)))]
-    pub async fn get_device(&self, user_id: &UserId, device_id: &DeviceId) -> Option<Device> {
+    pub async fn get_device(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+    ) -> StdResult<Option<Device>, CryptoStoreError> {
         let olm = self.olm.lock().await;
-        olm.as_ref()?.get_device(user_id, device_id).await
+
+        if let Some(olm) = olm.as_ref() {
+            olm.get_device(user_id, device_id).await
+        } else {
+            // TODO remove this panic.
+            panic!("The client hasn't been logged in")
+        }
     }
 
     /// Get a map holding all the devices of an user.

matrix_sdk_crypto/src/identities/device.rs

@@ -40,7 +40,7 @@ use crate::{
     error::{EventError, OlmError, OlmResult, SignatureError},
     identities::{OwnUserIdentity, UserIdentities},
     olm::Utility,
-    store::{caches::ReadOnlyUserDevices, Result as StoreResult},
+    store::{caches::ReadOnlyUserDevices, Result as StoreResult, Store},
     verification::VerificationMachine,
     Sas, ToDeviceRequest,
 };
@@ -120,43 +120,9 @@ impl Device {
         event_type: EventType,
         content: Value,
     ) -> OlmResult<EncryptedEventContent> {
-        let sender_key = if let Some(k) = self.inner.get_key(DeviceKeyAlgorithm::Curve25519) {
+        self.inner
-            k
+            .encrypt(self.verification_machine.store.clone(), event_type, content)
-        } else {
+            .await
-            warn!(
-                "Trying to encrypt a Megolm session for user {} on device {}, \
-                but the device doesn't have a curve25519 key",
-                self.user_id(),
-                self.device_id()
-            );
-            return Err(EventError::MissingSenderKey.into());
-        };
-
-        let mut session = if let Some(s) = self
-            .verification_machine
-            .store
-            .get_sessions(sender_key)
-            .await?
-        {
-            let session = &s.lock().await[0];
-            session.clone()
-        } else {
-            warn!(
-                "Trying to encrypt a Megolm session for user {} on device {}, \
-                but no Olm session is found",
-                self.user_id(),
-                self.device_id()
-            );
-            return Err(OlmError::MissingSession);
-        };
-
-        let message = session.encrypt(&self.inner, event_type, content).await;
-        self.verification_machine
-            .store
-            .save_sessions(&[session])
-            .await?;
-
-        message
     }
 }
 
@@ -310,7 +276,7 @@ impl ReadOnlyDevice {
         self.deleted.load(Ordering::Relaxed)
     }
 
-    fn trust_state(
+    pub(crate) fn trust_state(
         &self,
         own_identity: &Option<OwnUserIdentity>,
         device_owner: &Option<UserIdentities>,
@@ -352,6 +318,43 @@ impl ReadOnlyDevice {
         }
     }
 
+    pub(crate) async fn encrypt(
+        &self,
+        store: Store,
+        event_type: EventType,
+        content: Value,
+    ) -> OlmResult<EncryptedEventContent> {
+        let sender_key = if let Some(k) = self.get_key(DeviceKeyAlgorithm::Curve25519) {
+            k
+        } else {
+            warn!(
+                "Trying to encrypt a Megolm session for user {} on device {}, \
+                but the device doesn't have a curve25519 key",
+                self.user_id(),
+                self.device_id()
+            );
+            return Err(EventError::MissingSenderKey.into());
+        };
+
+        let mut session = if let Some(s) = store.get_sessions(sender_key).await? {
+            let session = &s.lock().await[0];
+            session.clone()
+        } else {
+            warn!(
+                "Trying to encrypt a Megolm session for user {} on device {}, \
+                but no Olm session is found",
+                self.user_id(),
+                self.device_id()
+            );
+            return Err(OlmError::MissingSession);
+        };
+
+        let message = session.encrypt(&self, event_type, content).await;
+        store.save_sessions(&[session]).await?;
+
+        message
+    }
+
     /// Update a device with a new device keys struct.
     pub(crate) fn update_device(&mut self, device_keys: &DeviceKeys) -> Result<(), SignatureError> {
         self.verify_device_keys(device_keys)?;

matrix_sdk_crypto/src/identities/manager.rs

@@ -380,8 +380,9 @@ pub(crate) mod test {
     }
 
     fn manager() -> IdentityManager {
-        let store = Store::new(Box::new(MemoryStore::new()));
+        let user_id = Arc::new(user_id());
-        IdentityManager::new(Arc::new(user_id()), Arc::new(device_id()), store)
+        let store = Store::new(user_id.clone(), Box::new(MemoryStore::new()));
+        IdentityManager::new(user_id, Arc::new(device_id()), store)
     }
 
     pub(crate) fn other_key_query() -> KeyQueryResponse {

matrix_sdk_crypto/src/identities/user.rs

@@ -659,7 +659,7 @@ impl OwnUserIdentity {
 
 #[cfg(test)]
 pub(crate) mod test {
-    use std::convert::TryFrom;
+    use std::{convert::TryFrom, sync::Arc};
 
     use crate::{
         identities::{
@@ -736,7 +736,10 @@ pub(crate) mod test {
 
         let verification_machine = VerificationMachine::new(
             Account::new(second.user_id(), second.device_id()),
-            Store::new(Box::new(MemoryStore::new())),
+            Store::new(
+                Arc::new(second.user_id().clone()),
+                Box::new(MemoryStore::new()),
+            ),
         );
 
         let first = Device {

matrix_sdk_crypto/src/key_request.rs

@@ -23,16 +23,20 @@
 //
 // If we don't trust the device store an object that remembers the request and
 // let the users introspect that object.
+
+#![allow(dead_code)]
+
 use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
-use serde_json::value::to_raw_value;
+use serde_json::{value::to_raw_value, Value};
-use std::{collections::BTreeMap, sync::Arc};
+use std::{collections::BTreeMap, convert::TryInto, sync::Arc};
 use tracing::{info, trace};
 
 use matrix_sdk_common::{
     api::r0::to_device::DeviceIdOrAllDevices,
     events::{
         forwarded_room_key::ForwardedRoomKeyEventContent,
+        room::encrypted::EncryptedEventContent,
         room_key_request::{Action, RequestedKeyInfo, RoomKeyRequestEventContent},
         AnyToDeviceEvent, EventType, ToDeviceEvent,
     },
@@ -42,17 +46,45 @@ use matrix_sdk_common::{
 };
 
 use crate::{
+    error::OlmResult,
+    identities::{OwnUserIdentity, ReadOnlyDevice, UserIdentities},
     olm::InboundGroupSession,
     requests::{OutgoingRequest, ToDeviceRequest},
     store::{CryptoStoreError, Store},
 };
 
+struct Device {
+    inner: ReadOnlyDevice,
+    store: Store,
+    own_identity: Option<OwnUserIdentity>,
+    device_owner_identity: Option<UserIdentities>,
+}
+
+impl Device {
+    fn trust_state(&self) -> bool {
+        self.inner
+            .trust_state(&self.own_identity, &self.device_owner_identity)
+    }
+
+    pub(crate) async fn encrypt(
+        &self,
+        event_type: EventType,
+        content: Value,
+    ) -> OlmResult<EncryptedEventContent> {
+        self.inner
+            .encrypt(self.store.clone(), event_type, content)
+            .await
+    }
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct KeyRequestMachine {
     user_id: Arc<UserId>,
     device_id: Arc<DeviceIdBox>,
     store: Store,
     outgoing_to_device_requests: Arc<DashMap<Uuid, OutgoingRequest>>,
+    incoming_key_requests:
+        Arc<DashMap<(UserId, DeviceIdBox, String), ToDeviceEvent<RoomKeyRequestEventContent>>>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -116,6 +148,7 @@ impl KeyRequestMachine {
             device_id,
             store,
             outgoing_to_device_requests: Arc::new(DashMap::new()),
+            incoming_key_requests: Arc::new(DashMap::new()),
         }
     }
 
@@ -131,6 +164,111 @@ impl KeyRequestMachine {
             .collect()
     }
 
+    /// Receive a room key request event.
+    pub fn receive_incoming_key_request(&self, event: &ToDeviceEvent<RoomKeyRequestEventContent>) {
+        let sender = event.sender.clone();
+        let device_id = event.content.requesting_device_id.clone();
+        let request_id = event.content.request_id.clone();
+        self.incoming_key_requests
+            .insert((sender, device_id, request_id), event.clone());
+    }
+
+    pub async fn collect_incoming_key_requests(&self) -> Result<(), CryptoStoreError> {
+        for item in self.incoming_key_requests.iter() {
+            let event = item.value();
+
+            // TODO move this into the handle key request method.
+            match event.content.action {
+                Action::Request => {
+                    self.handle_key_request(event).await?;
+                }
+                _ => (),
+            }
+
+            println!("HELLO {:?}", event);
+        }
+
+        self.incoming_key_requests.clear();
+
+        Ok(())
+    }
+
+    async fn handle_key_request(
+        &self,
+        event: &ToDeviceEvent<RoomKeyRequestEventContent>,
+    ) -> Result<(), CryptoStoreError> {
+        let key_info = event.content.body.as_ref().unwrap();
+        let session = self
+            .store
+            .get_inbound_group_session(
+                &key_info.room_id,
+                &key_info.sender_key,
+                &key_info.session_id,
+            )
+            .await?;
+
+        let session = if let Some(s) = session {
+            s
+        } else {
+            info!("Received a key request for an unknown inbound group session.");
+            return Ok(());
+        };
+
+        let device = self
+            .store
+            .get_device_and_users(&event.sender, &event.content.requesting_device_id)
+            .await?
+            .map(|(d, o, u)| Device {
+                inner: d,
+                store: self.store.clone(),
+                own_identity: o,
+                device_owner_identity: u,
+            });
+
+        if let Some(device) = device {
+            if self.user_id() == &event.sender {
+                self.handle_key_request_from_own_user(event, session, device)
+                    .await?;
+            } else {
+                self.handle_key_request_from_others(event, session, device)
+                    .await?;
+            }
+        } else {
+            info!("Received a key request from an unknown device.");
+            self.store.update_tracked_user(&event.sender, true).await?;
+        }
+
+        Ok(())
+    }
+
+    async fn handle_key_request_from_own_user(
+        &self,
+        event: &ToDeviceEvent<RoomKeyRequestEventContent>,
+        session: InboundGroupSession,
+        device: Device,
+    ) -> Result<(), CryptoStoreError> {
+        // TODO should we create yet another Device type that holds a store
+        // but not a verification machine?
+        if device.trust_state() {
+            let export = session.export().await;
+            let content: ForwardedRoomKeyEventContent = export.try_into().unwrap();
+
+            todo!("Queue up a key to be shared");
+        } else {
+            info!("Received a key request from an untrusted device.");
+        }
+        todo!()
+    }
+
+    async fn handle_key_request_from_others(
+        &self,
+        event: &ToDeviceEvent<RoomKeyRequestEventContent>,
+        session: InboundGroupSession,
+        device: Device,
+    ) -> Result<(), CryptoStoreError> {
+        todo!()
+    }
+
     /// Create a new outgoing key request for the key with the given session id.
     ///
     /// This will queue up a new to-device request and store the key info so
@@ -360,9 +498,10 @@ mod test {
     }
 
     fn get_machine() -> KeyRequestMachine {
-        let store = Store::new(Box::new(MemoryStore::new()));
+        let user_id = Arc::new(alice_id());
+        let store = Store::new(user_id.clone(), Box::new(MemoryStore::new()));
 
-        KeyRequestMachine::new(Arc::new(alice_id()), Arc::new(alice_device_id()), store)
+        KeyRequestMachine::new(user_id, Arc::new(alice_device_id()), store)
     }
 
     #[test]

matrix_sdk_crypto/src/machine.rs

@@ -124,9 +124,10 @@ impl OlmMachine {
         store: Box<dyn CryptoStore>,
         account: Account,
     ) -> Self {
-        let store = Store::new(store);
-        let verification_machine = VerificationMachine::new(account.clone(), store.clone());
         let user_id = Arc::new(user_id.clone());
+
+        let store = Store::new(user_id.clone(), store);
+        let verification_machine = VerificationMachine::new(account.clone(), store.clone());
         let device_id: Arc<DeviceIdBox> = Arc::new(device_id);
         let key_request_machine =
             KeyRequestMachine::new(user_id.clone(), device_id.clone(), store.clone());
@@ -1226,30 +1227,21 @@ impl OlmMachine {
     /// println!("{:?}", device);
     /// # });
     /// ```
-    pub async fn get_device(&self, user_id: &UserId, device_id: &DeviceId) -> Option<Device> {
+    pub async fn get_device(
-        let device = self
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+    ) -> StoreResult<Option<Device>> {
+        Ok(self
             .store
-            .get_device(user_id, device_id)
+            .get_device_and_users(user_id, device_id)
-            .await
+            .await?
-            .ok()
+            .map(|(d, o, u)| Device {
-            .flatten()?;
+                inner: d,
-
-        let own_identity = self
-            .store
-            .get_user_identity(self.user_id())
-            .await
-            .ok()
-            .flatten()
-            .map(|i| i.own().cloned())
-            .flatten();
-        let device_owner_identity = self.store.get_user_identity(user_id).await.ok().flatten();
-
-        Some(Device {
-            inner: device,
                 verification_machine: self.verification_machine.clone(),
-            own_identity,
+                own_identity: o,
-            device_owner_identity,
+                device_owner_identity: u,
-        })
+            }))
     }
 
     /// Get a map holding all the devices of an user.
@@ -1566,6 +1558,7 @@ pub(crate) mod test {
         let bob_device = alice
             .get_device(&bob.user_id, &bob.device_id)
             .await
+            .unwrap()
             .unwrap();
 
         let event = ToDeviceEvent {
@@ -1848,6 +1841,7 @@ pub(crate) mod test {
         let bob_device = alice
             .get_device(&bob.user_id, &bob.device_id)
             .await
+            .unwrap()
             .unwrap();
 
         let event = ToDeviceEvent {
@@ -2031,6 +2025,7 @@ pub(crate) mod test {
         let bob_device = alice
             .get_device(bob.user_id(), bob.device_id())
             .await
+            .unwrap()
             .unwrap();
 
         assert!(!bob_device.is_trusted());
@@ -2096,6 +2091,7 @@ pub(crate) mod test {
         let alice_device = bob
             .get_device(alice.user_id(), alice.device_id())
             .await
+            .unwrap()
             .unwrap();
 
         assert!(!alice_device.is_trusted());

matrix_sdk_crypto/src/store/memorystore.rs

@@ -119,6 +119,18 @@ impl CryptoStore for MemoryStore {
     }
 
     async fn update_tracked_user(&self, user: &UserId, dirty: bool) -> Result<bool> {
+        // TODO to prevent a race between the sync and a key query in flight we
+        // need to have an additional state to mention that the user changed.
+        //
+        // A simple counter could be used for this or enum with two states, e.g.
+        // The counter would work as follows:
+        // * 0 -> User is synced, no need for a key query.
+        // * 1 -> A sync has marked the user as dirty.
+        // * 2 -> A sync has marked the user again as dirty, before we got a
+        // successful key query response.
+        //
+        // The counter would top out at 2 since there won't be a race between 3
+        // different key queries syncs.
         if dirty {
             self.users_for_key_query.insert(user.clone());
         } else {

matrix_sdk_crypto/src/store/mod.rs

@@ -71,12 +71,15 @@ use matrix_sdk_common_macros::async_trait;
 use matrix_sdk_common_macros::send_sync;
 
 use super::{
-    identities::{ReadOnlyDevice, UserIdentities},
+    identities::{OwnUserIdentity, ReadOnlyDevice, UserIdentities},
     olm::{Account, InboundGroupSession, Session},
 };
 
 use crate::error::SessionUnpicklingError;
 
+/// A `CryptoStore` specific result type.
+pub type Result<T> = std::result::Result<T, CryptoStoreError>;
+
 /// A wrapper for our CryptoStore trait object.
 ///
 /// This is needed because we want to have a generic interface so we can
@@ -84,11 +87,48 @@ use crate::error::SessionUnpicklingError;
 /// generics don't mix let the CryptoStore store strings and this wrapper
 /// adds the generic interface on top.
 #[derive(Debug, Clone)]
-pub(crate) struct Store(Arc<Box<dyn CryptoStore>>);
+pub(crate) struct Store {
+    user_id: Arc<UserId>,
+    inner: Arc<Box<dyn CryptoStore>>,
+}
 
 impl Store {
-    pub fn new(store: Box<dyn CryptoStore>) -> Self {
+    pub fn new(user_id: Arc<UserId>, store: Box<dyn CryptoStore>) -> Self {
-        Self(Arc::new(store))
+        Self {
+            user_id,
+            inner: Arc::new(store),
+        }
+    }
+
+    pub async fn get_device_and_users(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+    ) -> Result<
+        Option<(
+            ReadOnlyDevice,
+            Option<OwnUserIdentity>,
+            Option<UserIdentities>,
+        )>,
+    > {
+        let device = self.get_device(user_id, device_id).await?;
+
+        let device = if let Some(d) = device {
+            d
+        } else {
+            return Ok(None);
+        };
+
+        let own_identity = self
+            .get_user_identity(&self.user_id)
+            .await
+            .ok()
+            .flatten()
+            .map(|i| i.own().cloned())
+            .flatten();
+        let device_owner_identity = self.get_user_identity(user_id).await.ok().flatten();
+
+        Ok(Some((device, own_identity, device_owner_identity)))
     }
 
     #[allow(dead_code)]
@@ -108,7 +148,7 @@ impl Store {
 
     #[allow(dead_code)]
     pub async fn delete_object(&self, key: &str) -> Result<()> {
-        self.0.remove_value(key).await?;
+        self.inner.remove_value(key).await?;
         Ok(())
     }
 }
@@ -117,13 +157,10 @@ impl Deref for Store {
     type Target = dyn CryptoStore;
 
     fn deref(&self) -> &Self::Target {
-        &**self.0
+        &**self.inner
     }
 }
 
-/// A `CryptoStore` specific result type.
-pub type Result<T> = std::result::Result<T, CryptoStoreError>;
-
 #[derive(Error, Debug)]
 /// The crypto store's error type.
 pub enum CryptoStoreError {

matrix_sdk_crypto/src/store/sqlite.rs

@@ -598,6 +598,8 @@ impl SqliteStore {
     async fn save_tracked_user(&self, user: &UserId, dirty: bool) -> Result<()> {
         let account_id = self.account_id().ok_or(CryptoStoreError::AccountUnset)?;
         let mut connection = self.connection.lock().await;
+        // TODO see the todo in the memory store, we need to avoid a race
+        // between a sync and key query.
 
         query(
             "INSERT INTO tracked_users (

matrix_sdk_crypto/src/verification/machine.rs

@@ -221,6 +221,7 @@ mod test {
 
     use std::{
         convert::TryFrom,
+        sync::Arc,
         time::{Duration, Instant},
     };
 
@@ -257,7 +258,7 @@ mod test {
         let alice = Account::new(&alice_id(), &alice_device_id());
         let bob = Account::new(&bob_id(), &bob_device_id());
         let store = MemoryStore::new();
-        let bob_store = Store::new(Box::new(MemoryStore::new()));
+        let bob_store = Store::new(Arc::new(bob_id()), Box::new(MemoryStore::new()));
 
         let bob_device = ReadOnlyDevice::from_account(&bob).await;
         let alice_device = ReadOnlyDevice::from_account(&alice).await;
@@ -268,7 +269,8 @@ mod test {
             .await
             .unwrap();
 
-        let machine = VerificationMachine::new(alice, Store::new(Box::new(store)));
+        let machine =
+            VerificationMachine::new(alice, Store::new(Arc::new(alice_id()), Box::new(store)));
         let (bob_sas, start_content) = Sas::start(bob, alice_device, bob_store, None);
         machine
             .receive_event(&mut wrap_any_to_device_content(
@@ -284,8 +286,9 @@ mod test {
     #[test]
     fn create() {
         let alice = Account::new(&alice_id(), &alice_device_id());
+        let user_id = Arc::new(alice_id());
         let store = MemoryStore::new();
-        let _ = VerificationMachine::new(alice, Store::new(Box::new(store)));
+        let _ = VerificationMachine::new(alice, Store::new(user_id, Box::new(store)));
     }
 
     #[tokio::test]

matrix_sdk_crypto/src/verification/sas/mod.rs

@@ -646,7 +646,7 @@ impl InnerSas {
 
 #[cfg(test)]
 mod test {
-    use std::convert::TryFrom;
+    use std::{convert::TryFrom, sync::Arc};
 
     use matrix_sdk_common::{
         events::{EventContent, ToDeviceEvent},
@@ -776,8 +776,8 @@ mod test {
         let bob = Account::new(&bob_id(), &bob_device_id());
         let bob_device = ReadOnlyDevice::from_account(&bob).await;
 
-        let alice_store = Store::new(Box::new(MemoryStore::new()));
+        let alice_store = Store::new(Arc::new(alice_id()), Box::new(MemoryStore::new()));
-        let bob_store = Store::new(Box::new(MemoryStore::new()));
+        let bob_store = Store::new(Arc::new(bob_id()), Box::new(MemoryStore::new()));
 
         bob_store
             .save_devices(&[alice_device.clone()])