crypto: Move the inbound group session lock into the session struct.

src/crypto/machine.rs

@@ -1139,7 +1139,7 @@ impl OlmMachine {
         // TODO check if the olm session is wedged and re-request the key.
         let session = session.ok_or(OlmError::MissingSession)?;
 
-        let (plaintext, _) = session.lock().await.decrypt(content.ciphertext.clone())?;
+        let (plaintext, _) = session.decrypt(content.ciphertext.clone()).await?;
         // TODO check the message index.
         // TODO check if this is from a verified device.
 

src/crypto/memory_stores.rs

@@ -60,7 +60,7 @@ impl SessionStore {
 
 #[derive(Debug)]
 pub struct GroupSessionStore {
-    entries: HashMap<RoomId, HashMap<String, HashMap<String, Arc<Mutex<InboundGroupSession>>>>>,
+    entries: HashMap<RoomId, HashMap<String, HashMap<String, InboundGroupSession>>>,
 }
 
 impl GroupSessionStore {
@@ -72,18 +72,19 @@ impl GroupSessionStore {
 
     pub fn add(&mut self, session: InboundGroupSession) -> bool {
         if !self.entries.contains_key(&session.room_id) {
-            self.entries
-                .insert(session.room_id.to_owned(), HashMap::new());
+            let room_id = &*session.room_id;
+            self.entries.insert(room_id.clone(), HashMap::new());
         }
 
         let room_map = self.entries.get_mut(&session.room_id).unwrap();
 
-        if !room_map.contains_key(&session.sender_key) {
-            room_map.insert(session.sender_key.to_owned(), HashMap::new());
+        if !room_map.contains_key(&*session.sender_key) {
+            let sender_key = &*session.sender_key;
+            room_map.insert(sender_key.to_owned(), HashMap::new());
         }
 
-        let sender_map = room_map.get_mut(&session.sender_key).unwrap();
-        let ret = sender_map.insert(session.session_id(), Arc::new(Mutex::new(session)));
+        let sender_map = room_map.get_mut(&*session.sender_key).unwrap();
+        let ret = sender_map.insert(session.session_id().to_owned(), session);
 
         ret.is_some()
     }
@@ -93,7 +94,7 @@ impl GroupSessionStore {
         room_id: &RoomId,
         sender_key: &str,
         session_id: &str,
-    ) -> Option<Arc<Mutex<InboundGroupSession>>> {
+    ) -> Option<InboundGroupSession> {
         self.entries
             .get(room_id)
             .and_then(|m| m.get(sender_key).and_then(|m| m.get(session_id).cloned()))

src/crypto/olm.rs

@@ -343,12 +343,14 @@ pub struct GroupSessionKey(pub String);
 ///
 /// Inbound group sessions are used to exchange room messages between a group of
 /// participants. Inbound group sessions are used to decrypt the room messages.
+#[derive(Clone)]
 pub struct InboundGroupSession {
-    inner: OlmInboundGroupSession,
-    pub(crate) sender_key: String,
-    pub(crate) signing_key: String,
-    pub(crate) room_id: RoomId,
-    forwarding_chains: Option<Vec<String>>,
+    inner: Arc<Mutex<OlmInboundGroupSession>>,
+    session_id: Arc<String>,
+    pub(crate) sender_key: Arc<String>,
+    pub(crate) signing_key: Arc<String>,
+    pub(crate) room_id: Arc<RoomId>,
+    forwarding_chains: Arc<Mutex<Option<Vec<String>>>>,
 }
 
 impl InboundGroupSession {
@@ -374,12 +376,16 @@ impl InboundGroupSession {
         room_id: &RoomId,
         session_key: GroupSessionKey,
     ) -> Result<Self, OlmGroupSessionError> {
+        let session = OlmInboundGroupSession::new(&session_key.0)?;
+        let session_id = session.session_id();
+
         Ok(InboundGroupSession {
-            inner: OlmInboundGroupSession::new(&session_key.0)?,
-            sender_key: sender_key.to_owned(),
-            signing_key: signing_key.to_owned(),
-            room_id: room_id.clone(),
-            forwarding_chains: None,
+            inner: Arc::new(Mutex::new(session)),
+            session_id: Arc::new(session_id),
+            sender_key: Arc::new(sender_key.to_owned()),
+            signing_key: Arc::new(signing_key.to_owned()),
+            room_id: Arc::new(room_id.clone()),
+            forwarding_chains: Arc::new(Mutex::new(None)),
         })
     }
 
@@ -389,8 +395,8 @@ impl InboundGroupSession {
     ///
     /// * `pickle_mode` - The mode that was used to pickle the group session,
     /// either an unencrypted mode or an encrypted using passphrase.
-    pub fn pickle(&self, pickle_mode: PicklingMode) -> String {
-        self.inner.pickle(pickle_mode)
+    pub async fn pickle(&self, pickle_mode: PicklingMode) -> String {
+        self.inner.lock().await.pickle(pickle_mode)
     }
 
     /// Restore a Session from a previously pickled string.
@@ -420,23 +426,26 @@ impl InboundGroupSession {
         room_id: RoomId,
     ) -> Result<Self, OlmGroupSessionError> {
         let session = OlmInboundGroupSession::unpickle(pickle, pickle_mode)?;
+        let session_id = session.session_id();
+
         Ok(InboundGroupSession {
-            inner: session,
-            sender_key,
-            signing_key,
-            room_id,
-            forwarding_chains: None,
+            inner: Arc::new(Mutex::new(session)),
+            session_id: Arc::new(session_id),
+            sender_key: Arc::new(sender_key),
+            signing_key: Arc::new(signing_key),
+            room_id: Arc::new(room_id),
+            forwarding_chains: Arc::new(Mutex::new(None)),
         })
     }
 
     /// Returns the unique identifier for this session.
-    pub fn session_id(&self) -> String {
-        self.inner.session_id()
+    pub fn session_id(&self) -> &str {
+        &self.session_id
     }
 
     /// Get the first message index we know how to decrypt.
-    pub fn first_known_index(&self) -> u32 {
-        self.inner.first_known_index()
+    pub async fn first_known_index(&self) -> u32 {
+        self.inner.lock().await.first_known_index()
     }
 
     /// Decrypt the given ciphertext.
@@ -447,8 +456,8 @@ impl InboundGroupSession {
     /// # Arguments
     ///
     /// * `message` - The message that should be decrypted.
-    pub fn decrypt(&self, message: String) -> Result<(String, u32), OlmGroupSessionError> {
-        self.inner.decrypt(message)
+    pub async fn decrypt(&self, message: String) -> Result<(String, u32), OlmGroupSessionError> {
+        self.inner.lock().await.decrypt(message)
     }
 }
 

src/crypto/store/memorystore.rs

@@ -77,7 +77,7 @@ impl CryptoStore for MemoryStore {
         room_id: &RoomId,
         sender_key: &str,
         session_id: &str,
-    ) -> Result<Option<Arc<Mutex<InboundGroupSession>>>> {
+    ) -> Result<Option<InboundGroupSession>> {
         Ok(self
             .inbound_group_sessions
             .get(room_id, sender_key, session_id))

src/crypto/store/mod.rs

@@ -67,21 +67,25 @@ pub type Result<T> = std::result::Result<T, CryptoStoreError>;
 pub trait CryptoStore: Debug + Send + Sync {
     async fn load_account(&mut self) -> Result<Option<Account>>;
     async fn save_account(&mut self, account: Account) -> Result<()>;
+
     async fn save_session(&mut self, session: Arc<Mutex<Session>>) -> Result<()>;
     async fn add_and_save_session(&mut self, session: Session) -> Result<()>;
     async fn get_sessions(
         &mut self,
         sender_key: &str,
     ) -> Result<Option<Arc<Mutex<Vec<Arc<Mutex<Session>>>>>>>;
+
     async fn save_inbound_group_session(&mut self, session: InboundGroupSession) -> Result<bool>;
     async fn get_inbound_group_session(
         &mut self,
         room_id: &RoomId,
         sender_key: &str,
         session_id: &str,
-    ) -> Result<Option<Arc<Mutex<InboundGroupSession>>>>;
+    ) -> Result<Option<InboundGroupSession>>;
+
     fn tracked_users(&self) -> &HashSet<UserId>;
     async fn add_user_for_tracking(&mut self, user: &UserId) -> Result<bool>;
+
     async fn save_device(&self, device: Device) -> Result<()>;
     async fn get_device(&self, user_id: &UserId, device_id: &str) -> Result<Option<Device>>;
     async fn get_user_devices(&self, user_id: &UserId) -> Result<UserDevices>;

src/crypto/store/sqlite.rs

@@ -366,7 +366,7 @@ impl CryptoStore for SqliteStore {
 
     async fn save_inbound_group_session(&mut self, session: InboundGroupSession) -> Result<bool> {
         let account_id = self.account_id.ok_or(CryptoStoreError::AccountUnset)?;
-        let pickle = session.pickle(self.get_pickle_mode());
+        let pickle = session.pickle(self.get_pickle_mode()).await;
         let mut connection = self.connection.lock().await;
         let session_id = session.session_id();
 
@@ -382,9 +382,9 @@ impl CryptoStore for SqliteStore {
         )
         .bind(session_id)
         .bind(account_id)
-        .bind(&session.sender_key)
-        .bind(&session.signing_key)
-        .bind(&session.room_id.to_string())
+        .bind(&*session.sender_key)
+        .bind(&*session.signing_key)
+        .bind(&*session.room_id.to_string())
         .bind(&pickle)
         .execute(&mut *connection)
         .await?;
@@ -397,7 +397,7 @@ impl CryptoStore for SqliteStore {
         room_id: &RoomId,
         sender_key: &str,
         session_id: &str,
-    ) -> Result<Option<Arc<Mutex<InboundGroupSession>>>> {
+    ) -> Result<Option<InboundGroupSession>> {
         Ok(self
             .inbound_group_sessions
             .get(room_id, sender_key, session_id))
@@ -653,7 +653,7 @@ mod test {
         )
         .expect("Can't create session");
 
-        let session_id = session.session_id();
+        let session_id = session.session_id().to_owned();
 
         store
             .save_inbound_group_session(session)