crypto: Add the users for key claiming to the outgoing requests

This makes sure that we immediately claim one-time keys after we
receive the sync changes instead of waiting for a room message to be
sent by the user.

Users may not send a message in a long time which would mean that we'll
likely never share secrets or forward room keys if a Olm session was
missing with the requester.

matrix_sdk/src/client.rs

@@ -2120,6 +2120,14 @@ impl Client {
                                     .unwrap();
                             }
                         }
+                        OutgoingRequests::KeysClaim(request) => {
+                            if let Ok(resp) = self.send(request.clone(), None).await {
+                                self.base_client
+                                    .mark_request_as_sent(r.request_id(), &resp)
+                                    .await
+                                    .unwrap();
+                            }
+                        }
                     }
                 }
             }

matrix_sdk_crypto/src/gossiping/machine.rs

@@ -20,11 +20,12 @@
 // If we don't trust the device store an object that remembers the request and
 // let the users introspect that object.
 
-use std::sync::Arc;
+use std::{collections::BTreeMap, sync::Arc};
 
 use dashmap::{mapref::entry::Entry, DashMap, DashSet};
 use matrix_sdk_common::uuid::Uuid;
 use ruma::{
+    api::client::r0::keys::claim_keys::Request as KeysClaimRequest,
     events::{
         forwarded_room_key::ForwardedRoomKeyToDeviceEventContent,
         room_key_request::{Action, RequestedKeyInfo, RoomKeyRequestToDeviceEventContent},
@@ -36,7 +37,7 @@ use ruma::{
         },
         AnyToDeviceEvent, AnyToDeviceEventContent, ToDeviceEvent,
     },
-    DeviceId, DeviceIdBox, EventEncryptionAlgorithm, RoomId, UserId,
+    DeviceId, DeviceIdBox, DeviceKeyAlgorithm, EventEncryptionAlgorithm, RoomId, UserId,
 };
 use tracing::{debug, info, trace, warn};
 
@@ -112,6 +113,28 @@ impl GossipMachine {
             self.outgoing_requests.iter().map(|i| i.value().clone()).collect();
         key_requests.extend(key_forwards);
 
+        let users_for_key_claim: BTreeMap<_, _> = self
+            .users_for_key_claim
+            .iter()
+            .map(|i| {
+                let device_map = i
+                    .value()
+                    .iter()
+                    .map(|d| (d.key().to_owned(), DeviceKeyAlgorithm::SignedCurve25519))
+                    .collect();
+
+                (i.key().to_owned(), device_map)
+            })
+            .collect();
+
+        if !users_for_key_claim.is_empty() {
+            let key_claim_request = KeysClaimRequest::new(users_for_key_claim);
+            key_requests.push(OutgoingRequest {
+                request_id: Uuid::new_v4(),
+                request: Arc::new(key_claim_request.into()),
+            });
+        }
+
         Ok(key_requests)
     }
 
@@ -866,6 +889,7 @@ mod test {
     use std::{convert::TryInto, sync::Arc};
 
     use dashmap::DashMap;
+    use matches::assert_matches;
     use matrix_sdk_common::locks::Mutex;
     use matrix_sdk_test::async_test;
     use ruma::{
@@ -888,6 +912,7 @@ mod test {
         session_manager::GroupSessionCache,
         store::{Changes, CryptoStore, MemoryStore, Store},
         verification::VerificationMachine,
+        OutgoingRequests,
     };
 
     fn alice_id() -> UserId {
@@ -1507,8 +1532,12 @@ mod test {
         // Receive the room key request from alice.
         bob_machine.receive_incoming_key_request(&event);
         bob_machine.collect_incoming_key_requests().await.unwrap();
-        // Bob doesn't have an outgoing requests since we're lacking a session.
-        assert!(bob_machine.outgoing_to_device_requests().await.unwrap().is_empty());
+        // Bob only has a keys claim request, since we're lacking a session
+        assert_eq!(bob_machine.outgoing_to_device_requests().await.unwrap().len(), 1);
+        assert_matches!(
+            bob_machine.outgoing_to_device_requests().await.unwrap().first().unwrap().request(),
+            OutgoingRequests::KeysClaim(_)
+        );
         assert!(!bob_machine.users_for_key_claim.is_empty());
         assert!(!bob_machine.wait_queue.is_empty());
 

matrix_sdk_crypto/src/requests.rs

@@ -18,7 +18,7 @@ use matrix_sdk_common::uuid::Uuid;
 use ruma::{
     api::client::r0::{
         keys::{
-            claim_keys::Response as KeysClaimResponse,
+            claim_keys::{Request as KeysClaimRequest, Response as KeysClaimResponse},
             get_keys::Response as KeysQueryResponse,
             upload_keys::{Request as KeysUploadRequest, Response as KeysUploadResponse},
             upload_signatures::{
@@ -183,6 +183,10 @@ pub enum OutgoingRequests {
     /// The keys query request, fetching the device and cross singing keys of
     /// other users.
     KeysQuery(KeysQueryRequest),
+    /// The request to claim one-time keys for a user/device pair from the
+    /// server, after the response is received an 1-to-1 Olm session will be
+    /// established with the user/device pair.
+    KeysClaim(KeysClaimRequest),
     /// The to-device requests, this request is used for a couple of different
     /// things, the main use is key requests/forwards and interactive device
     /// verification.
@@ -211,6 +215,12 @@ impl From<KeysQueryRequest> for OutgoingRequests {
     }
 }
 
+impl From<KeysClaimRequest> for OutgoingRequests {
+    fn from(r: KeysClaimRequest) -> Self {
+        Self::KeysClaim(r)
+    }
+}
+
 impl From<KeysUploadRequest> for OutgoingRequests {
     fn from(request: KeysUploadRequest) -> Self {
         OutgoingRequests::KeysUpload(request)

matrix_sdk_crypto/src/session_manager/sessions.rs

@@ -279,7 +279,22 @@ impl SessionManager {
             }
         }
 
-        Ok(self.store.save_changes(changes).await?)
+        // TODO turn this into a single save_changes() call.
+        self.store.save_changes(changes).await?;
+
+        match self.key_request_machine.collect_incoming_key_requests().await {
+            Ok(sessions) => {
+                let changes = Changes { sessions, ..Default::default() };
+                self.store.save_changes(changes).await?
+            }
+            // We don't propagate the error here since the next sync will retry
+            // this.
+            Err(e) => {
+                warn!(error =? e, "Error while trying to collect the incoming secret requests")
+            }
+        }
+
+        Ok(())
     }
 }
 

matrix_sdk_crypto/src/verification/event_enums.rs

@@ -770,6 +770,7 @@ impl TryFrom<OutgoingRequest> for OutgoingContent {
             crate::OutgoingRequests::KeysQuery(_) => Err("Invalid request type".to_owned()),
             crate::OutgoingRequests::ToDeviceRequest(r) => Self::try_from(r.clone()),
             crate::OutgoingRequests::SignatureUpload(_) => Err("Invalid request type".to_owned()),
+            crate::OutgoingRequests::KeysClaim(_) => Err("Invalid request type".to_owned()),
             crate::OutgoingRequests::RoomMessage(r) => Ok(Self::from(r.clone())),
         }
     }