Skip to main content

matrix_sdk_crypto/store/
memorystore.rs

1// Copyright 2020, 2026 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{
16    collections::{BTreeMap, HashMap, HashSet},
17    convert::Infallible,
18    sync::Arc,
19};
20
21use async_trait::async_trait;
22use matrix_sdk_common::{
23    cross_process_lock::{
24        CrossProcessLockGeneration,
25        memory_store_helper::{Lease, try_take_leased_lock},
26    },
27    locks::RwLock as StdRwLock,
28};
29use ruma::{
30    DeviceId, OwnedDeviceId, OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId, TransactionId,
31    UserId, events::secret::request::SecretName,
32};
33use tokio::sync::{Mutex, RwLock};
34use tracing::warn;
35use vodozemac::Curve25519PublicKey;
36use zeroize::Zeroizing;
37
38use super::{
39    Account, CryptoStore, InboundGroupSession, Session,
40    caches::DeviceStore,
41    types::{
42        BackupKeys, Changes, DehydratedDeviceKey, PendingChanges, RoomKeyCounts, RoomSettings,
43        StoredRoomKeyBundleData, TrackedUser,
44    },
45};
46use crate::{
47    gossiping::{GossipRequest, SecretInfo},
48    identities::{DeviceData, UserIdentityData},
49    olm::{
50        OutboundGroupSession, PickledAccount, PickledInboundGroupSession, PickledSession,
51        PrivateCrossSigningIdentity, SenderDataType, StaticAccountData,
52    },
53    store::types::{RoomKeyWithheldEntry, RoomPendingKeyBundleDetails},
54};
55
56fn encode_key_info(info: &SecretInfo) -> String {
57    match info {
58        SecretInfo::KeyRequest(info) => {
59            format!("{}{}{}", info.room_id(), info.algorithm(), info.session_id())
60        }
61        SecretInfo::SecretRequest(i) => i.as_ref().to_owned(),
62    }
63}
64
65type SessionId = String;
66
67/// The "version" of a backup - newtype wrapper around a String.
68#[derive(Clone, Debug, PartialEq)]
69struct BackupVersion(String);
70
71impl BackupVersion {
72    fn from(s: &str) -> Self {
73        Self(s.to_owned())
74    }
75
76    fn as_str(&self) -> &str {
77        &self.0
78    }
79}
80
81/// An in-memory only store that will forget all the E2EE key once it's dropped.
82#[derive(Default, Debug)]
83pub struct MemoryStore {
84    static_account: Arc<StdRwLock<Option<StaticAccountData>>>,
85
86    account: StdRwLock<Option<String>>,
87    // Map of sender_key to map of session_id to serialized pickle
88    sessions: StdRwLock<BTreeMap<String, BTreeMap<String, String>>>,
89    inbound_group_sessions: StdRwLock<BTreeMap<OwnedRoomId, HashMap<String, String>>>,
90
91    /// Map room id -> session id -> backup order number The latest backup in
92    /// which this session is stored. Equivalent to `backed_up_to` in
93    /// [`IndexedDbCryptoStore`]
94    inbound_group_sessions_backed_up_to:
95        StdRwLock<HashMap<OwnedRoomId, HashMap<SessionId, BackupVersion>>>,
96
97    outbound_group_sessions: StdRwLock<BTreeMap<OwnedRoomId, OutboundGroupSession>>,
98    private_identity: StdRwLock<Option<PrivateCrossSigningIdentity>>,
99    tracked_users: StdRwLock<HashMap<OwnedUserId, TrackedUser>>,
100    olm_hashes: StdRwLock<HashMap<String, HashSet<String>>>,
101    devices: DeviceStore,
102    identities: StdRwLock<HashMap<OwnedUserId, String>>,
103    outgoing_key_requests: StdRwLock<HashMap<OwnedTransactionId, GossipRequest>>,
104    key_requests_by_info: StdRwLock<HashMap<String, OwnedTransactionId>>,
105    direct_withheld_info: StdRwLock<HashMap<OwnedRoomId, HashMap<String, RoomKeyWithheldEntry>>>,
106    custom_values: StdRwLock<HashMap<String, Vec<u8>>>,
107    leases: StdRwLock<HashMap<String, Lease>>,
108    secret_inbox: StdRwLock<HashMap<String, Vec<Zeroizing<String>>>>,
109    backup_keys: RwLock<BackupKeys>,
110    dehydrated_device_pickle_key: RwLock<Option<DehydratedDeviceKey>>,
111    next_batch_token: RwLock<Option<String>>,
112    room_settings: StdRwLock<HashMap<OwnedRoomId, RoomSettings>>,
113    room_key_bundles:
114        StdRwLock<HashMap<OwnedRoomId, HashMap<OwnedUserId, StoredRoomKeyBundleData>>>,
115    room_key_backups_fully_downloaded: StdRwLock<HashSet<OwnedRoomId>>,
116    rooms_pending_key_bundle: StdRwLock<HashMap<OwnedRoomId, RoomPendingKeyBundleDetails>>,
117
118    save_changes_lock: Arc<Mutex<()>>,
119}
120
121impl MemoryStore {
122    /// Create a new empty `MemoryStore`.
123    pub fn new() -> Self {
124        Self::default()
125    }
126
127    fn get_static_account(&self) -> Option<StaticAccountData> {
128        self.static_account.read().clone()
129    }
130
131    pub(crate) fn save_devices(&self, devices: Vec<DeviceData>) {
132        for device in devices {
133            let _ = self.devices.add(device);
134        }
135    }
136
137    fn delete_devices(&self, devices: Vec<DeviceData>) {
138        for device in devices {
139            let _ = self.devices.remove(device.user_id(), device.device_id());
140        }
141    }
142
143    fn save_sessions(&self, sessions: Vec<(String, PickledSession)>) {
144        let mut session_store = self.sessions.write();
145
146        for (session_id, pickle) in sessions {
147            let entry = session_store.entry(pickle.sender_key.to_base64()).or_default();
148
149            // insert or replace if exists
150            entry.insert(
151                session_id,
152                serde_json::to_string(&pickle).expect("Failed to serialize olm session"),
153            );
154        }
155    }
156
157    fn save_outbound_group_sessions(&self, sessions: Vec<OutboundGroupSession>) {
158        self.outbound_group_sessions
159            .write()
160            .extend(sessions.into_iter().map(|s| (s.room_id().to_owned(), s)));
161    }
162
163    fn save_private_identity(&self, private_identity: Option<PrivateCrossSigningIdentity>) {
164        *self.private_identity.write() = private_identity;
165    }
166
167    /// Return all the [`InboundGroupSession`]s we have, paired with the
168    /// `backed_up_to` value for each one (or "" where it is missing, which
169    /// should never happen).
170    async fn get_inbound_group_sessions_and_backed_up_to(
171        &self,
172    ) -> Result<Vec<(InboundGroupSession, Option<BackupVersion>)>> {
173        let lookup = |s: &InboundGroupSession| {
174            self.inbound_group_sessions_backed_up_to
175                .read()
176                .get(&s.room_id)?
177                .get(s.session_id())
178                .cloned()
179        };
180
181        Ok(self
182            .get_inbound_group_sessions()
183            .await?
184            .into_iter()
185            .map(|s| {
186                let v = lookup(&s);
187                (s, v)
188            })
189            .collect())
190    }
191}
192
193type Result<T> = std::result::Result<T, Infallible>;
194
195#[cfg_attr(target_family = "wasm", async_trait(?Send))]
196#[cfg_attr(not(target_family = "wasm"), async_trait)]
197impl CryptoStore for MemoryStore {
198    type Error = Infallible;
199
200    async fn close(&self) -> Result<()> {
201        Ok(())
202    }
203
204    async fn reopen(&self) -> Result<()> {
205        Ok(())
206    }
207
208    async fn load_account(&self) -> Result<Option<Account>> {
209        let pickled_account: Option<PickledAccount> = self.account.read().as_ref().map(|acc| {
210            serde_json::from_str(acc)
211                .expect("Deserialization failed: invalid pickled account JSON format")
212        });
213
214        if let Some(pickle) = pickled_account {
215            let account =
216                Account::from_pickle(pickle).expect("From pickle failed: invalid pickle format");
217
218            *self.static_account.write() = Some(account.static_data().clone());
219
220            Ok(Some(account))
221        } else {
222            Ok(None)
223        }
224    }
225
226    async fn load_identity(&self) -> Result<Option<PrivateCrossSigningIdentity>> {
227        Ok(self.private_identity.read().clone())
228    }
229
230    async fn next_batch_token(&self) -> Result<Option<String>> {
231        Ok(self.next_batch_token.read().await.clone())
232    }
233
234    async fn delete_next_batch_token(&self) -> Result<()> {
235        *self.next_batch_token.write().await = None;
236        Ok(())
237    }
238
239    async fn save_pending_changes(&self, changes: PendingChanges) -> Result<()> {
240        let _guard = self.save_changes_lock.lock().await;
241
242        let pickled_account = if let Some(account) = changes.account {
243            *self.static_account.write() = Some(account.static_data().clone());
244            Some(account.pickle())
245        } else {
246            None
247        };
248
249        *self.account.write() = pickled_account.map(|pickle| {
250            serde_json::to_string(&pickle)
251                .expect("Serialization failed: invalid pickled account JSON format")
252        });
253
254        Ok(())
255    }
256
257    async fn save_changes(&self, changes: Changes) -> Result<()> {
258        let _guard = self.save_changes_lock.lock().await;
259
260        let mut pickled_session: Vec<(String, PickledSession)> = Vec::new();
261        for session in changes.sessions {
262            let session_id = session.session_id().to_owned();
263            let pickle = session.pickle().await;
264            pickled_session.push((session_id.clone(), pickle));
265        }
266        self.save_sessions(pickled_session);
267
268        self.save_inbound_group_sessions(changes.inbound_group_sessions, None).await?;
269        self.save_outbound_group_sessions(changes.outbound_group_sessions);
270        self.save_private_identity(changes.private_identity);
271
272        self.save_devices(changes.devices.new);
273        self.save_devices(changes.devices.changed);
274        self.delete_devices(changes.devices.deleted);
275
276        {
277            let mut identities = self.identities.write();
278            for identity in changes.identities.new.into_iter().chain(changes.identities.changed) {
279                identities.insert(
280                    identity.user_id().to_owned(),
281                    serde_json::to_string(&identity)
282                        .expect("UserIdentityData should always serialize to json"),
283                );
284            }
285        }
286
287        {
288            let mut olm_hashes = self.olm_hashes.write();
289            for hash in changes.message_hashes {
290                olm_hashes.entry(hash.sender_key.to_owned()).or_default().insert(hash.hash);
291            }
292        }
293
294        {
295            let mut outgoing_key_requests = self.outgoing_key_requests.write();
296            let mut key_requests_by_info = self.key_requests_by_info.write();
297
298            for key_request in changes.key_requests {
299                let id = key_request.request_id.clone();
300                let info_string = encode_key_info(&key_request.info);
301
302                // If we have an old request for the same key/secret, remove it.
303                if let Some(old_id) = key_requests_by_info.get(&info_string) {
304                    outgoing_key_requests.remove(old_id);
305                    key_requests_by_info.remove(&info_string);
306                }
307
308                outgoing_key_requests.insert(id.clone(), key_request);
309                key_requests_by_info.insert(info_string, id);
310            }
311        }
312
313        if let Some(key) = changes.backup_decryption_key {
314            self.backup_keys.write().await.decryption_key = Some(key);
315        }
316
317        if let Some(version) = changes.backup_version {
318            self.backup_keys.write().await.backup_version = Some(version);
319        }
320
321        if let Some(pickle_key) = changes.dehydrated_device_pickle_key {
322            let mut lock = self.dehydrated_device_pickle_key.write().await;
323            *lock = Some(pickle_key);
324        }
325
326        {
327            let mut secret_inbox = self.secret_inbox.write();
328            for secret in changes.secrets {
329                secret_inbox.entry(secret.secret_name.to_string()).or_default().push(secret.secret);
330            }
331        }
332
333        {
334            let mut direct_withheld_info = self.direct_withheld_info.write();
335            for (room_id, data) in changes.withheld_session_info {
336                for (session_id, event) in data {
337                    direct_withheld_info
338                        .entry(room_id.to_owned())
339                        .or_default()
340                        .insert(session_id, event);
341                }
342            }
343        }
344
345        if let Some(next_batch_token) = changes.next_batch_token {
346            *self.next_batch_token.write().await = Some(next_batch_token);
347        }
348
349        if !changes.room_settings.is_empty() {
350            let mut settings = self.room_settings.write();
351            settings.extend(changes.room_settings);
352        }
353
354        if !changes.received_room_key_bundles.is_empty() {
355            let mut room_key_bundles = self.room_key_bundles.write();
356            for bundle in changes.received_room_key_bundles {
357                room_key_bundles
358                    .entry(bundle.bundle_data.room_id.clone())
359                    .or_default()
360                    .insert(bundle.sender_user.clone(), bundle);
361            }
362        }
363
364        if !changes.room_key_backups_fully_downloaded.is_empty() {
365            let mut room_key_backups_fully_downloaded =
366                self.room_key_backups_fully_downloaded.write();
367            for room in changes.room_key_backups_fully_downloaded {
368                room_key_backups_fully_downloaded.insert(room);
369            }
370        }
371
372        if !changes.rooms_pending_key_bundle.is_empty() {
373            let mut lock = self.rooms_pending_key_bundle.write();
374            for (room, details) in changes.rooms_pending_key_bundle {
375                if let Some(details) = details {
376                    lock.insert(room, details);
377                } else {
378                    lock.remove(&room);
379                }
380            }
381        }
382
383        Ok(())
384    }
385
386    async fn save_inbound_group_sessions(
387        &self,
388        sessions: Vec<InboundGroupSession>,
389        backed_up_to_version: Option<&str>,
390    ) -> Result<()> {
391        for session in sessions {
392            let room_id = session.room_id();
393            let session_id = session.session_id();
394
395            // Sanity-check that the data in the sessions corresponds to
396            // backed_up_version
397            let backed_up = session.backed_up();
398            if backed_up != backed_up_to_version.is_some() {
399                warn!(
400                    backed_up,
401                    backed_up_to_version,
402                    "Session backed-up flag does not correspond to backup version setting",
403                );
404            }
405
406            if let Some(backup_version) = backed_up_to_version {
407                self.inbound_group_sessions_backed_up_to
408                    .write()
409                    .entry(room_id.to_owned())
410                    .or_default()
411                    .insert(session_id.to_owned(), BackupVersion::from(backup_version));
412            }
413
414            let pickle = session.pickle().await;
415            self.inbound_group_sessions
416                .write()
417                .entry(session.room_id().to_owned())
418                .or_default()
419                .insert(
420                    session.session_id().to_owned(),
421                    serde_json::to_string(&pickle)
422                        .expect("Pickle pickle data should serialize to json"),
423                );
424        }
425        Ok(())
426    }
427
428    async fn get_sessions(&self, sender_key: &str) -> Result<Option<Vec<Session>>> {
429        let device_keys = self.get_own_device().await?.as_device_keys().clone();
430
431        if let Some(pickles) = self.sessions.read().get(sender_key) {
432            let mut sessions: Vec<Session> = Vec::new();
433            for serialized_pickle in pickles.values() {
434                let pickle: PickledSession = serde_json::from_str(serialized_pickle.as_str())
435                    .expect("Pickle pickle deserialization should work");
436                let session = Session::from_pickle(device_keys.clone(), pickle)
437                    .expect("Expect from pickle to always work");
438                sessions.push(session);
439            }
440            Ok(Some(sessions))
441        } else {
442            Ok(None)
443        }
444    }
445
446    async fn get_inbound_group_session(
447        &self,
448        room_id: &RoomId,
449        session_id: &str,
450    ) -> Result<Option<InboundGroupSession>> {
451        let pickle: Option<PickledInboundGroupSession> = self
452            .inbound_group_sessions
453            .read()
454            .get(room_id)
455            .and_then(|m| m.get(session_id))
456            .and_then(|ser| {
457                serde_json::from_str(ser).expect("Pickle pickle deserialization should work")
458            });
459
460        Ok(pickle.map(|p| {
461            InboundGroupSession::from_pickle(p).expect("Expect from pickle to always work")
462        }))
463    }
464
465    async fn get_withheld_info(
466        &self,
467        room_id: &RoomId,
468        session_id: &str,
469    ) -> Result<Option<RoomKeyWithheldEntry>> {
470        Ok(self
471            .direct_withheld_info
472            .read()
473            .get(room_id)
474            .and_then(|e| Some(e.get(session_id)?.to_owned())))
475    }
476
477    async fn get_withheld_sessions_by_room_id(
478        &self,
479        room_id: &RoomId,
480    ) -> crate::store::Result<Vec<RoomKeyWithheldEntry>, Self::Error> {
481        Ok(self
482            .direct_withheld_info
483            .read()
484            .get(room_id)
485            .map(|e| e.values().cloned().collect())
486            .unwrap_or_default())
487    }
488
489    async fn get_inbound_group_sessions(&self) -> Result<Vec<InboundGroupSession>> {
490        let inbounds = self
491            .inbound_group_sessions
492            .read()
493            .values()
494            .flat_map(HashMap::values)
495            .map(|ser| {
496                let pickle: PickledInboundGroupSession =
497                    serde_json::from_str(ser).expect("Pickle deserialization should work");
498                InboundGroupSession::from_pickle(pickle).expect("Expect from pickle to always work")
499            })
500            .collect();
501        Ok(inbounds)
502    }
503
504    async fn inbound_group_session_counts(
505        &self,
506        backup_version: Option<&str>,
507    ) -> Result<RoomKeyCounts> {
508        let backed_up = if let Some(backup_version) = backup_version {
509            self.get_inbound_group_sessions_and_backed_up_to()
510                .await?
511                .into_iter()
512                // Count the sessions backed up in the required backup
513                .filter(|(_, o)| o.as_ref().is_some_and(|o| o.as_str() == backup_version))
514                .count()
515        } else {
516            // We asked about a nonexistent backup version - this doesn't make
517            // much sense, but we can easily answer that nothing is backed up in
518            // this nonexistent backup.
519            0
520        };
521
522        let total = self.inbound_group_sessions.read().values().map(HashMap::len).sum();
523        Ok(RoomKeyCounts { total, backed_up })
524    }
525
526    async fn get_inbound_group_sessions_by_room_id(
527        &self,
528        room_id: &RoomId,
529    ) -> Result<Vec<InboundGroupSession>> {
530        let inbounds = match self.inbound_group_sessions.read().get(room_id) {
531            None => Vec::new(),
532            Some(v) => v
533                .values()
534                .map(|ser| {
535                    let pickle: PickledInboundGroupSession =
536                        serde_json::from_str(ser).expect("Pickle deserialization should work");
537                    InboundGroupSession::from_pickle(pickle)
538                        .expect("Expect from pickle to always work")
539                })
540                .collect(),
541        };
542        Ok(inbounds)
543    }
544
545    async fn get_inbound_group_sessions_for_device_batch(
546        &self,
547        sender_key: Curve25519PublicKey,
548        sender_data_type: SenderDataType,
549        after_session_id: Option<String>,
550        limit: usize,
551    ) -> Result<Vec<InboundGroupSession>> {
552        // First, find all InboundGroupSessions, filtering for those that match
553        // the device and sender_data type.
554        let mut sessions: Vec<_> = self
555            .get_inbound_group_sessions()
556            .await?
557            .into_iter()
558            .filter(|session: &InboundGroupSession| {
559                session.creator_info.curve25519_key == sender_key
560                    && session.sender_data.to_type() == sender_data_type
561            })
562            .collect();
563
564        // Then, sort the sessions in order of ascending session ID...
565        sessions.sort_by_key(|s| s.session_id().to_owned());
566
567        // Figure out where in the array to start returning results from
568        let start_index = {
569            match after_session_id {
570                None => 0,
571                Some(id) => {
572                    // We're looking for the first session with a session ID
573                    // strictly after `id`; if there are none, the end of the
574                    // array.
575                    sessions
576                        .iter()
577                        .position(|session| session.session_id() > id.as_str())
578                        .unwrap_or(sessions.len())
579                }
580            }
581        };
582
583        // Return up to `limit` items from the array, starting from
584        // `start_index`
585        Ok(sessions.drain(start_index..).take(limit).collect())
586    }
587
588    async fn inbound_group_sessions_for_backup(
589        &self,
590        backup_version: &str,
591        limit: usize,
592    ) -> Result<Vec<InboundGroupSession>> {
593        Ok(self
594            .get_inbound_group_sessions_and_backed_up_to()
595            .await?
596            .into_iter()
597            .filter_map(|(session, backed_up_to)| {
598                if let Some(ref existing_version) = backed_up_to
599                    && existing_version.as_str() == backup_version
600                {
601                    // This session is already backed up in the required backup
602                    None
603                } else {
604                    // It's not backed up, or it's backed up in a different
605                    // backup
606                    Some(session)
607                }
608            })
609            .take(limit)
610            .collect())
611    }
612
613    async fn mark_inbound_group_sessions_as_backed_up(
614        &self,
615        backup_version: &str,
616        room_and_session_ids: &[(&RoomId, &str)],
617    ) -> Result<()> {
618        for &(room_id, session_id) in room_and_session_ids {
619            let session = self.get_inbound_group_session(room_id, session_id).await?;
620
621            if let Some(session) = session {
622                session.mark_as_backed_up();
623
624                self.inbound_group_sessions_backed_up_to
625                    .write()
626                    .entry(room_id.to_owned())
627                    .or_default()
628                    .insert(session_id.to_owned(), BackupVersion::from(backup_version));
629
630                // Save it back
631                let updated_pickle = session.pickle().await;
632
633                self.inbound_group_sessions.write().entry(room_id.to_owned()).or_default().insert(
634                    session_id.to_owned(),
635                    serde_json::to_string(&updated_pickle)
636                        .expect("Pickle serialization should work"),
637                );
638            }
639        }
640
641        Ok(())
642    }
643
644    async fn reset_backup_state(&self) -> Result<()> {
645        // Nothing to do here, because we remember which backup versions we
646        // backed up to in `mark_inbound_group_sessions_as_backed_up`, so we
647        // don't need to reset anything here because the required version is
648        // passed in to `inbound_group_sessions_for_backup`, and we can compare
649        // against the version we stored.
650
651        Ok(())
652    }
653
654    async fn load_backup_keys(&self) -> Result<BackupKeys> {
655        Ok(self.backup_keys.read().await.to_owned())
656    }
657
658    async fn load_dehydrated_device_pickle_key(&self) -> Result<Option<DehydratedDeviceKey>> {
659        Ok(self.dehydrated_device_pickle_key.read().await.to_owned())
660    }
661
662    async fn delete_dehydrated_device_pickle_key(&self) -> Result<()> {
663        let mut lock = self.dehydrated_device_pickle_key.write().await;
664        *lock = None;
665        Ok(())
666    }
667
668    async fn get_outbound_group_session(
669        &self,
670        room_id: &RoomId,
671    ) -> Result<Option<OutboundGroupSession>> {
672        Ok(self.outbound_group_sessions.read().get(room_id).cloned())
673    }
674
675    async fn load_tracked_users(&self) -> Result<Vec<TrackedUser>> {
676        Ok(self.tracked_users.read().values().cloned().collect())
677    }
678
679    async fn save_tracked_users(&self, tracked_users: &[(&UserId, bool)]) -> Result<()> {
680        self.tracked_users.write().extend(tracked_users.iter().map(|(user_id, dirty)| {
681            let user_id: OwnedUserId = user_id.to_owned().into();
682            (user_id.clone(), TrackedUser { user_id, dirty: *dirty })
683        }));
684        Ok(())
685    }
686
687    async fn get_device(
688        &self,
689        user_id: &UserId,
690        device_id: &DeviceId,
691    ) -> Result<Option<DeviceData>> {
692        Ok(self.devices.get(user_id, device_id))
693    }
694
695    async fn get_user_devices(
696        &self,
697        user_id: &UserId,
698    ) -> Result<HashMap<OwnedDeviceId, DeviceData>> {
699        Ok(self.devices.user_devices(user_id))
700    }
701
702    async fn get_own_device(&self) -> Result<DeviceData> {
703        let account =
704            self.get_static_account().expect("Expect account to exist when getting own device");
705
706        Ok(self
707            .devices
708            .get(&account.user_id, &account.device_id)
709            .expect("Invalid state: Should always have a own device"))
710    }
711
712    async fn get_user_identity(&self, user_id: &UserId) -> Result<Option<UserIdentityData>> {
713        let serialized = self.identities.read().get(user_id).cloned();
714        match serialized {
715            None => Ok(None),
716            Some(serialized) => {
717                let id: UserIdentityData = serde_json::from_str(serialized.as_str())
718                    .expect("Only valid serialized identity are saved");
719                Ok(Some(id))
720            }
721        }
722    }
723
724    async fn is_message_known(&self, message_hash: &crate::olm::OlmMessageHash) -> Result<bool> {
725        Ok(self
726            .olm_hashes
727            .write()
728            .entry(message_hash.sender_key.to_owned())
729            .or_default()
730            .contains(&message_hash.hash))
731    }
732
733    async fn get_outgoing_secret_requests(
734        &self,
735        request_id: &TransactionId,
736    ) -> Result<Option<GossipRequest>> {
737        Ok(self.outgoing_key_requests.read().get(request_id).cloned())
738    }
739
740    async fn get_secret_request_by_info(
741        &self,
742        key_info: &SecretInfo,
743    ) -> Result<Option<GossipRequest>> {
744        let key_info_string = encode_key_info(key_info);
745
746        Ok(self
747            .key_requests_by_info
748            .read()
749            .get(&key_info_string)
750            .and_then(|i| self.outgoing_key_requests.read().get(i).cloned()))
751    }
752
753    async fn get_unsent_secret_requests(&self) -> Result<Vec<GossipRequest>> {
754        Ok(self
755            .outgoing_key_requests
756            .read()
757            .values()
758            .filter(|req| !req.sent_out)
759            .cloned()
760            .collect())
761    }
762
763    async fn delete_outgoing_secret_requests(&self, request_id: &TransactionId) -> Result<()> {
764        let req = self.outgoing_key_requests.write().remove(request_id);
765        if let Some(i) = req {
766            let key_info_string = encode_key_info(&i.info);
767            self.key_requests_by_info.write().remove(&key_info_string);
768        }
769
770        Ok(())
771    }
772
773    async fn get_secrets_from_inbox(
774        &self,
775        secret_name: &SecretName,
776    ) -> Result<Vec<Zeroizing<String>>> {
777        Ok(self.secret_inbox.write().entry(secret_name.to_string()).or_default().to_owned())
778    }
779
780    async fn delete_secrets_from_inbox(&self, secret_name: &SecretName) -> Result<()> {
781        self.secret_inbox.write().remove(secret_name.as_str());
782
783        Ok(())
784    }
785
786    async fn get_room_settings(&self, room_id: &RoomId) -> Result<Option<RoomSettings>> {
787        Ok(self.room_settings.read().get(room_id).cloned())
788    }
789
790    async fn get_received_room_key_bundle_data(
791        &self,
792        room_id: &RoomId,
793        user_id: &UserId,
794    ) -> Result<Option<StoredRoomKeyBundleData>> {
795        let guard = self.room_key_bundles.read();
796
797        let result = guard.get(room_id).and_then(|bundles| bundles.get(user_id).cloned());
798
799        Ok(result)
800    }
801
802    async fn get_pending_key_bundle_details_for_room(
803        &self,
804        room_id: &RoomId,
805    ) -> Result<Option<RoomPendingKeyBundleDetails>> {
806        Ok(self.rooms_pending_key_bundle.read().get(room_id).cloned())
807    }
808
809    async fn get_all_rooms_pending_key_bundles(&self) -> Result<Vec<RoomPendingKeyBundleDetails>> {
810        Ok(self.rooms_pending_key_bundle.read().values().cloned().collect())
811    }
812
813    async fn has_downloaded_all_room_keys(&self, room_id: &RoomId) -> Result<bool> {
814        let guard = self.room_key_backups_fully_downloaded.read();
815        Ok(guard.contains(room_id))
816    }
817
818    async fn get_custom_value(&self, key: &str) -> Result<Option<Vec<u8>>> {
819        Ok(self.custom_values.read().get(key).cloned())
820    }
821
822    async fn set_custom_value(&self, key: &str, value: Vec<u8>) -> Result<()> {
823        self.custom_values.write().insert(key.to_owned(), value);
824        Ok(())
825    }
826
827    async fn remove_custom_value(&self, key: &str) -> Result<()> {
828        self.custom_values.write().remove(key);
829        Ok(())
830    }
831
832    async fn try_take_leased_lock(
833        &self,
834        lease_duration_ms: u32,
835        key: &str,
836        holder: &str,
837    ) -> Result<Option<CrossProcessLockGeneration>> {
838        Ok(try_take_leased_lock(&mut self.leases.write(), lease_duration_ms, key, holder))
839    }
840
841    async fn get_size(&self) -> Result<Option<usize>> {
842        Ok(None)
843    }
844}
845
846#[cfg(test)]
847mod tests {
848    use std::collections::HashMap;
849
850    use matrix_sdk_test::async_test;
851    use ruma::{RoomId, room_id, user_id};
852    use vodozemac::{Curve25519PublicKey, Ed25519PublicKey};
853
854    use super::SessionId;
855    use crate::{
856        DeviceData,
857        identities::device::testing::get_device,
858        olm::{
859            Account, InboundGroupSession, OlmMessageHash, PrivateCrossSigningIdentity, SenderData,
860            tests::get_account_and_session_test_helper,
861        },
862        store::{
863            CryptoStore,
864            memorystore::MemoryStore,
865            types::{Changes, DeviceChanges, PendingChanges},
866        },
867    };
868
869    #[async_test]
870    async fn test_session_store() {
871        let (account, session) = get_account_and_session_test_helper();
872        let own_device = DeviceData::from_account(&account);
873        let store = MemoryStore::new();
874
875        assert!(store.load_account().await.unwrap().is_none());
876
877        store
878            .save_changes(Changes {
879                devices: DeviceChanges { new: vec![own_device], ..Default::default() },
880                ..Default::default()
881            })
882            .await
883            .unwrap();
884        store.save_pending_changes(PendingChanges { account: Some(account) }).await.unwrap();
885
886        store
887            .save_changes(Changes { sessions: (vec![session.clone()]), ..Default::default() })
888            .await
889            .unwrap();
890
891        let sessions = store.get_sessions(&session.sender_key.to_base64()).await.unwrap().unwrap();
892
893        let loaded_session = &sessions[0];
894
895        assert_eq!(&session, loaded_session);
896    }
897
898    #[async_test]
899    async fn test_inbound_group_session_store() {
900        let (account, _) = get_account_and_session_test_helper();
901        let room_id = room_id!("!test:localhost");
902        let curve_key = "Nn0L2hkcCMFKqynTjyGsJbth7QrVmX3lbrksMkrGOAw";
903
904        let (outbound, _) = account.create_group_session_pair_with_defaults(room_id).await;
905        let inbound = InboundGroupSession::new(
906            Curve25519PublicKey::from_base64(curve_key).unwrap(),
907            Ed25519PublicKey::from_base64("ee3Ek+J2LkkPmjGPGLhMxiKnhiX//xcqaVL4RP6EypE").unwrap(),
908            room_id,
909            &outbound.session_key().await,
910            SenderData::unknown(),
911            None,
912            outbound.settings().algorithm.to_owned(),
913            None,
914            false,
915        )
916        .unwrap();
917
918        let store = MemoryStore::new();
919        store.save_inbound_group_sessions(vec![inbound.clone()], None).await.unwrap();
920
921        let loaded_session =
922            store.get_inbound_group_session(room_id, outbound.session_id()).await.unwrap().unwrap();
923        assert_eq!(inbound, loaded_session);
924    }
925
926    #[async_test]
927    async fn test_backing_up_marks_sessions_as_backed_up() {
928        // Given there are 2 sessions
929        let room_id = room_id!("!test:localhost");
930        let (store, sessions) = store_with_sessions(2, room_id).await;
931
932        // When I mark them as backed up
933        mark_backed_up(&store, room_id, "bkp1", &sessions).await;
934
935        // Then their backed_up_to field is set
936        let but = backed_up_tos(&store).await;
937        assert_eq!(but[sessions[0].session_id()], "bkp1");
938        assert_eq!(but[sessions[1].session_id()], "bkp1");
939    }
940
941    #[async_test]
942    async fn test_backing_up_a_second_set_of_sessions_updates_their_backup_order() {
943        // Given there are 3 sessions
944        let room_id = room_id!("!test:localhost");
945        let (store, sessions) = store_with_sessions(3, room_id).await;
946
947        // When I mark 0 and 1 as backed up in bkp1
948        mark_backed_up(&store, room_id, "bkp1", &sessions[..2]).await;
949
950        // And 1 and 2 as backed up in bkp2
951        mark_backed_up(&store, room_id, "bkp2", &sessions[1..]).await;
952
953        // Then 0 is backed up in bkp1 and the 1 and 2 are backed up in bkp2
954        let but = backed_up_tos(&store).await;
955        assert_eq!(but[sessions[0].session_id()], "bkp1");
956        assert_eq!(but[sessions[1].session_id()], "bkp2");
957        assert_eq!(but[sessions[2].session_id()], "bkp2");
958    }
959
960    #[async_test]
961    async fn test_backing_up_again_to_the_same_version_has_no_effect() {
962        // Given there are 3 sessions
963        let room_id = room_id!("!test:localhost");
964        let (store, sessions) = store_with_sessions(3, room_id).await;
965
966        // When I mark the first two as backed up in the first backup
967        mark_backed_up(&store, room_id, "bkp1", &sessions[..2]).await;
968
969        // And the last 2 as backed up in the same backup version
970        mark_backed_up(&store, room_id, "bkp1", &sessions[1..]).await;
971
972        // Then they all get the same backed_up_to value
973        let but = backed_up_tos(&store).await;
974        assert_eq!(but[sessions[0].session_id()], "bkp1");
975        assert_eq!(but[sessions[1].session_id()], "bkp1");
976        assert_eq!(but[sessions[2].session_id()], "bkp1");
977    }
978
979    #[async_test]
980    async fn test_backing_up_to_an_old_backup_version_can_increase_backed_up_to() {
981        // Given we have backed up some sessions to 2 backup versions, an older
982        // and a newer
983        let room_id = room_id!("!test:localhost");
984        let (store, sessions) = store_with_sessions(4, room_id).await;
985        mark_backed_up(&store, room_id, "older_bkp", &sessions[..2]).await;
986        mark_backed_up(&store, room_id, "newer_bkp", &sessions[1..2]).await;
987
988        // When I ask to back up the un-backed-up ones to the older backup
989        mark_backed_up(&store, room_id, "older_bkp", &sessions[2..]).await;
990
991        // Then each session lists the backup it was most recently included in
992        let but = backed_up_tos(&store).await;
993        assert_eq!(but[sessions[0].session_id()], "older_bkp");
994        assert_eq!(but[sessions[1].session_id()], "newer_bkp");
995        assert_eq!(but[sessions[2].session_id()], "older_bkp");
996        assert_eq!(but[sessions[3].session_id()], "older_bkp");
997    }
998
999    #[async_test]
1000    async fn test_backing_up_to_an_old_backup_version_overwrites_a_newer_one() {
1001        // Given we have backed up to 2 backup versions, an older and a newer
1002        let room_id = room_id!("!test:localhost");
1003        let (store, sessions) = store_with_sessions(4, room_id).await;
1004        mark_backed_up(&store, room_id, "older_bkp", &sessions).await;
1005        // Sanity: they are backed up in order number 1
1006        assert_eq!(backed_up_tos(&store).await[sessions[0].session_id()], "older_bkp");
1007        mark_backed_up(&store, room_id, "newer_bkp", &sessions).await;
1008        // Sanity: they are backed up in order number 2
1009        assert_eq!(backed_up_tos(&store).await[sessions[0].session_id()], "newer_bkp");
1010
1011        // When I ask to back up some to the older version
1012        mark_backed_up(&store, room_id, "older_bkp", &sessions[..2]).await;
1013
1014        // Then older backup overwrites: we don't consider the order here at all
1015        let but = backed_up_tos(&store).await;
1016        assert_eq!(but[sessions[0].session_id()], "older_bkp");
1017        assert_eq!(but[sessions[1].session_id()], "older_bkp");
1018        assert_eq!(but[sessions[2].session_id()], "newer_bkp");
1019        assert_eq!(but[sessions[3].session_id()], "newer_bkp");
1020    }
1021
1022    #[async_test]
1023    async fn test_not_backed_up_sessions_are_eligible_for_backup() {
1024        // Given there are 4 sessions, 2 of which are already backed up
1025        let room_id = room_id!("!test:localhost");
1026        let (store, sessions) = store_with_sessions(4, room_id).await;
1027        mark_backed_up(&store, room_id, "bkp1", &sessions[..2]).await;
1028
1029        // When I ask which to back up
1030        let mut to_backup = store
1031            .inbound_group_sessions_for_backup("bkp1", 10)
1032            .await
1033            .expect("Failed to ask for sessions to backup");
1034        to_backup.sort_by_key(|s| s.session_id().to_owned());
1035
1036        // Then I am told the last 2 only
1037        assert_eq!(to_backup, &[sessions[2].clone(), sessions[3].clone()]);
1038    }
1039
1040    #[async_test]
1041    async fn test_all_sessions_are_eligible_for_backup_if_version_is_unknown() {
1042        // Given there are 4 sessions, 2 of which are already backed up in bkp1
1043        let room_id = room_id!("!test:localhost");
1044        let (store, sessions) = store_with_sessions(4, room_id).await;
1045        mark_backed_up(&store, room_id, "bkp1", &sessions[..2]).await;
1046
1047        // When I ask which to back up in an unknown version
1048        let mut to_backup = store
1049            .inbound_group_sessions_for_backup("unknown_bkp", 10)
1050            .await
1051            .expect("Failed to ask for sessions to backup");
1052        to_backup.sort_by_key(|s| s.session_id().to_owned());
1053
1054        // Then I am told to back up all of them
1055        assert_eq!(
1056            to_backup,
1057            &[sessions[0].clone(), sessions[1].clone(), sessions[2].clone(), sessions[3].clone()]
1058        );
1059    }
1060
1061    #[async_test]
1062    async fn test_sessions_backed_up_to_a_later_version_are_eligible_for_backup() {
1063        // Given there are 4 sessions, some backed up to three different
1064        // versions
1065        let room_id = room_id!("!test:localhost");
1066        let (store, sessions) = store_with_sessions(4, room_id).await;
1067        mark_backed_up(&store, room_id, "bkp0", &sessions[..1]).await;
1068        mark_backed_up(&store, room_id, "bkp1", &sessions[1..2]).await;
1069        mark_backed_up(&store, room_id, "bkp2", &sessions[2..3]).await;
1070
1071        // When I ask which to back up in the middle version
1072        let mut to_backup = store
1073            .inbound_group_sessions_for_backup("bkp1", 10)
1074            .await
1075            .expect("Failed to ask for sessions to backup");
1076        to_backup.sort_by_key(|s| s.session_id().to_owned());
1077
1078        // Then I am told to back up everything not in the version I asked about
1079        assert_eq!(
1080            to_backup,
1081            &[
1082                sessions[0].clone(), // Backed up in bkp0
1083                // sessions[1] is backed up in bkp1 already, which we asked
1084                // about
1085                sessions[2].clone(), // Backed up in bkp2
1086                sessions[3].clone(), // Not backed up
1087            ]
1088        );
1089    }
1090
1091    #[async_test]
1092    async fn test_outbound_group_session_store() {
1093        // Given an outbound session
1094        let (account, _) = get_account_and_session_test_helper();
1095        let room_id = room_id!("!test:localhost");
1096        let (outbound, _) = account.create_group_session_pair_with_defaults(room_id).await;
1097
1098        // When we save it to the store
1099        let store = MemoryStore::new();
1100        store.save_outbound_group_sessions(vec![outbound.clone()]);
1101
1102        // Then we can get it out again
1103        let loaded_session = store.get_outbound_group_session(room_id).await.unwrap().unwrap();
1104        assert_eq!(
1105            serde_json::to_string(&outbound.pickle().await).unwrap(),
1106            serde_json::to_string(&loaded_session.pickle().await).unwrap()
1107        );
1108    }
1109
1110    #[async_test]
1111    async fn test_tracked_users_are_stored_once_per_user_id() {
1112        // Given a store containing 2 tracked users, both dirty
1113        let user1 = user_id!("@user1:s");
1114        let user2 = user_id!("@user2:s");
1115        let user3 = user_id!("@user3:s");
1116        let store = MemoryStore::new();
1117        store.save_tracked_users(&[(user1, true), (user2, true)]).await.unwrap();
1118
1119        // When we mark one as clean and add another
1120        store.save_tracked_users(&[(user2, false), (user3, false)]).await.unwrap();
1121
1122        // Then we can get them out again and their dirty flags are correct
1123        let loaded_tracked_users =
1124            store.load_tracked_users().await.expect("failed to load tracked users");
1125
1126        let tracked_contains = |user_id, dirty| {
1127            loaded_tracked_users.iter().any(|u| u.user_id == user_id && u.dirty == dirty)
1128        };
1129
1130        assert!(tracked_contains(user1, true));
1131        assert!(tracked_contains(user2, false));
1132        assert!(tracked_contains(user3, false));
1133        assert_eq!(loaded_tracked_users.len(), 3);
1134    }
1135
1136    #[async_test]
1137    async fn test_private_identity_store() {
1138        // Given a private identity
1139        let private_identity = PrivateCrossSigningIdentity::empty(user_id!("@u:s"));
1140
1141        // When we save it to the store
1142        let store = MemoryStore::new();
1143        store.save_private_identity(Some(private_identity.clone()));
1144
1145        // Then we can get it out again
1146        let loaded_identity =
1147            store.load_identity().await.expect("failed to load private identity").unwrap();
1148
1149        assert_eq!(loaded_identity.user_id(), "@u:s");
1150    }
1151
1152    #[async_test]
1153    async fn test_device_store() {
1154        let device = get_device();
1155        let store = MemoryStore::new();
1156
1157        store.save_devices(vec![device.clone()]);
1158
1159        let loaded_device =
1160            store.get_device(device.user_id(), device.device_id()).await.unwrap().unwrap();
1161
1162        assert_eq!(device, loaded_device);
1163
1164        let user_devices = store.get_user_devices(device.user_id()).await.unwrap();
1165
1166        assert_eq!(&**user_devices.keys().next().unwrap(), device.device_id());
1167        assert_eq!(user_devices.values().next().unwrap(), &device);
1168
1169        let loaded_device = user_devices.get(device.device_id()).unwrap();
1170
1171        assert_eq!(&device, loaded_device);
1172
1173        store.delete_devices(vec![device.clone()]);
1174        assert!(store.get_device(device.user_id(), device.device_id()).await.unwrap().is_none());
1175    }
1176
1177    #[async_test]
1178    async fn test_message_hash() {
1179        let store = MemoryStore::new();
1180
1181        let hash =
1182            OlmMessageHash { sender_key: "test_sender".to_owned(), hash: "test_hash".to_owned() };
1183
1184        let mut changes = Changes::default();
1185        changes.message_hashes.push(hash.clone());
1186
1187        assert!(!store.is_message_known(&hash).await.unwrap());
1188        store.save_changes(changes).await.unwrap();
1189        assert!(store.is_message_known(&hash).await.unwrap());
1190    }
1191
1192    #[async_test]
1193    async fn test_key_counts_of_empty_store_are_zero() {
1194        // Given an empty store
1195        let store = MemoryStore::new();
1196
1197        // When we count keys
1198        let key_counts = store.inbound_group_session_counts(Some("")).await.unwrap();
1199
1200        // Then the answer is zero
1201        assert_eq!(key_counts.total, 0);
1202        assert_eq!(key_counts.backed_up, 0);
1203    }
1204
1205    #[async_test]
1206    async fn test_counting_sessions_reports_the_number_of_sessions() {
1207        // Given a store with sessions
1208        let room_id = room_id!("!test:localhost");
1209        let (store, _) = store_with_sessions(4, room_id).await;
1210
1211        // When we count keys
1212        let key_counts = store.inbound_group_session_counts(Some("bkp")).await.unwrap();
1213
1214        // Then the answer equals the number of sessions we created
1215        assert_eq!(key_counts.total, 4);
1216        // And none are backed up
1217        assert_eq!(key_counts.backed_up, 0);
1218    }
1219
1220    #[async_test]
1221    async fn test_counting_backed_up_sessions_reports_the_number_backed_up_in_this_backup() {
1222        // Given a store with sessions, some backed up
1223        let room_id = room_id!("!test:localhost");
1224        let (store, sessions) = store_with_sessions(5, room_id).await;
1225        mark_backed_up(&store, room_id, "bkp", &sessions[..2]).await;
1226
1227        // When we count keys
1228        let key_counts = store.inbound_group_session_counts(Some("bkp")).await.unwrap();
1229
1230        // Then the answer equals the number of sessions we created
1231        assert_eq!(key_counts.total, 5);
1232        // And the backed_up count matches how many were backed up
1233        assert_eq!(key_counts.backed_up, 2);
1234    }
1235
1236    #[async_test]
1237    async fn test_counting_backed_up_sessions_for_null_backup_reports_zero() {
1238        // Given a store with sessions, some backed up
1239        let room_id = room_id!("!test:localhost");
1240        let (store, sessions) = store_with_sessions(4, room_id).await;
1241        mark_backed_up(&store, room_id, "bkp", &sessions[..2]).await;
1242
1243        // When we count keys, providing None as the backup version
1244        let key_counts = store.inbound_group_session_counts(None).await.unwrap();
1245
1246        // Then we ignore everything and just say zero
1247        assert_eq!(key_counts.backed_up, 0);
1248    }
1249
1250    #[async_test]
1251    async fn test_counting_backed_up_sessions_only_reports_sessions_in_the_version_specified() {
1252        // Given a store with sessions, backed up in several versions
1253        let room_id = room_id!("!test:localhost");
1254        let (store, sessions) = store_with_sessions(4, room_id).await;
1255        mark_backed_up(&store, room_id, "bkp1", &sessions[..2]).await;
1256        mark_backed_up(&store, room_id, "bkp2", &sessions[3..]).await;
1257
1258        // When we count keys for bkp2
1259        let key_counts = store.inbound_group_session_counts(Some("bkp2")).await.unwrap();
1260
1261        // Then the backed_up count reflects how many were backed up in bkp2
1262        // only
1263        assert_eq!(key_counts.backed_up, 1);
1264    }
1265
1266    /// Mark the supplied sessions as backed up in the supplied backup version
1267    async fn mark_backed_up(
1268        store: &MemoryStore,
1269        room_id: &RoomId,
1270        backup_version: &str,
1271        sessions: &[InboundGroupSession],
1272    ) {
1273        let rooms_and_ids: Vec<_> = sessions.iter().map(|s| (room_id, s.session_id())).collect();
1274
1275        store
1276            .mark_inbound_group_sessions_as_backed_up(backup_version, &rooms_and_ids)
1277            .await
1278            .expect("Failed to mark sessions as backed up");
1279    }
1280
1281    // Create a MemoryStore containing the supplied number of sessions.
1282    //
1283    // Sessions are returned in alphabetical order of session id.
1284    async fn store_with_sessions(
1285        num_sessions: usize,
1286        room_id: &RoomId,
1287    ) -> (MemoryStore, Vec<InboundGroupSession>) {
1288        let (account, _) = get_account_and_session_test_helper();
1289
1290        let mut sessions = Vec::with_capacity(num_sessions);
1291        for _ in 0..num_sessions {
1292            sessions.push(new_session(&account, room_id).await);
1293        }
1294        sessions.sort_by_key(|s| s.session_id().to_owned());
1295
1296        let store = MemoryStore::new();
1297        store.save_inbound_group_sessions(sessions.clone(), None).await.unwrap();
1298
1299        (store, sessions)
1300    }
1301
1302    // Create a new InboundGroupSession
1303    async fn new_session(account: &Account, room_id: &RoomId) -> InboundGroupSession {
1304        let curve_key = "Nn0L2hkcCMFKqynTjyGsJbth7QrVmX3lbrksMkrGOAw";
1305        let (outbound, _) = account.create_group_session_pair_with_defaults(room_id).await;
1306
1307        InboundGroupSession::new(
1308            Curve25519PublicKey::from_base64(curve_key).unwrap(),
1309            Ed25519PublicKey::from_base64("ee3Ek+J2LkkPmjGPGLhMxiKnhiX//xcqaVL4RP6EypE").unwrap(),
1310            room_id,
1311            &outbound.session_key().await,
1312            SenderData::unknown(),
1313            None,
1314            outbound.settings().algorithm.to_owned(),
1315            None,
1316            false,
1317        )
1318        .unwrap()
1319    }
1320
1321    /// Find the session_id and backed_up_to value for each of the sessions in
1322    /// the store.
1323    async fn backed_up_tos(store: &MemoryStore) -> HashMap<SessionId, String> {
1324        store
1325            .get_inbound_group_sessions_and_backed_up_to()
1326            .await
1327            .expect("Unable to get inbound group sessions and backup order")
1328            .iter()
1329            .map(|(s, o)| {
1330                (
1331                    s.session_id().to_owned(),
1332                    o.as_ref().map(|v| v.as_str().to_owned()).unwrap_or("".to_owned()),
1333                )
1334            })
1335            .collect()
1336    }
1337}
1338
1339#[cfg(test)]
1340mod integration_tests {
1341    use std::{
1342        collections::HashMap,
1343        sync::{Arc, Mutex, OnceLock},
1344    };
1345
1346    use async_trait::async_trait;
1347    use matrix_sdk_common::cross_process_lock::CrossProcessLockGeneration;
1348    use ruma::{
1349        DeviceId, OwnedDeviceId, RoomId, TransactionId, UserId, events::secret::request::SecretName,
1350    };
1351    use vodozemac::Curve25519PublicKey;
1352    use zeroize::Zeroizing;
1353
1354    use super::MemoryStore;
1355    use crate::{
1356        Account, DeviceData, GossipRequest, SecretInfo, Session, UserIdentityData,
1357        cryptostore_integration_tests, cryptostore_integration_tests_time,
1358        olm::{
1359            InboundGroupSession, OlmMessageHash, OutboundGroupSession, PrivateCrossSigningIdentity,
1360            SenderDataType, StaticAccountData,
1361        },
1362        store::{
1363            CryptoStore,
1364            types::{
1365                BackupKeys, Changes, DehydratedDeviceKey, PendingChanges, RoomKeyCounts,
1366                RoomKeyWithheldEntry, RoomPendingKeyBundleDetails, RoomSettings,
1367                StoredRoomKeyBundleData, TrackedUser,
1368            },
1369        },
1370    };
1371
1372    /// Holds on to a MemoryStore during a test, and moves it back into STORES
1373    /// when this is dropped
1374    #[derive(Clone, Debug)]
1375    struct PersistentMemoryStore(Arc<MemoryStore>);
1376
1377    impl PersistentMemoryStore {
1378        fn new() -> Self {
1379            Self(Arc::new(MemoryStore::new()))
1380        }
1381
1382        fn get_static_account(&self) -> Option<StaticAccountData> {
1383            self.0.get_static_account()
1384        }
1385    }
1386
1387    /// Return a clone of the store for the test with the supplied name. Note:
1388    /// dropping this store won't destroy its data, since
1389    /// [PersistentMemoryStore] is a reference-counted smart pointer to an
1390    /// underlying [MemoryStore].
1391    async fn get_store(
1392        name: &str,
1393        _passphrase: Option<&str>,
1394        clear_data: bool,
1395    ) -> PersistentMemoryStore {
1396        // Holds on to one [PersistentMemoryStore] per test, so even if the test
1397        // drops the store, we keep its data alive. This simulates the behaviour
1398        // of the other stores, which keep their data in a real DB, allowing us
1399        // to test MemoryStore using the same code.
1400        static STORES: OnceLock<Mutex<HashMap<String, PersistentMemoryStore>>> = OnceLock::new();
1401        let stores = STORES.get_or_init(|| Mutex::new(HashMap::new()));
1402
1403        let mut stores = stores.lock().unwrap();
1404
1405        if clear_data {
1406            // Create a new PersistentMemoryStore
1407            let new_store = PersistentMemoryStore::new();
1408            stores.insert(name.to_owned(), new_store.clone());
1409            new_store
1410        } else {
1411            stores.entry(name.to_owned()).or_insert_with(PersistentMemoryStore::new).clone()
1412        }
1413    }
1414
1415    /// Forwards all methods to the underlying [MemoryStore].
1416    #[cfg_attr(target_family = "wasm", async_trait(?Send))]
1417    #[cfg_attr(not(target_family = "wasm"), async_trait)]
1418    impl CryptoStore for PersistentMemoryStore {
1419        type Error = <MemoryStore as CryptoStore>::Error;
1420
1421        async fn close(&self) -> Result<(), Self::Error> {
1422            self.0.close().await
1423        }
1424
1425        async fn reopen(&self) -> Result<(), Self::Error> {
1426            self.0.reopen().await
1427        }
1428
1429        async fn load_account(&self) -> Result<Option<Account>, Self::Error> {
1430            self.0.load_account().await
1431        }
1432
1433        async fn load_identity(&self) -> Result<Option<PrivateCrossSigningIdentity>, Self::Error> {
1434            self.0.load_identity().await
1435        }
1436
1437        async fn save_changes(&self, changes: Changes) -> Result<(), Self::Error> {
1438            self.0.save_changes(changes).await
1439        }
1440
1441        async fn save_pending_changes(&self, changes: PendingChanges) -> Result<(), Self::Error> {
1442            self.0.save_pending_changes(changes).await
1443        }
1444
1445        async fn save_inbound_group_sessions(
1446            &self,
1447            sessions: Vec<InboundGroupSession>,
1448            backed_up_to_version: Option<&str>,
1449        ) -> Result<(), Self::Error> {
1450            self.0.save_inbound_group_sessions(sessions, backed_up_to_version).await
1451        }
1452
1453        async fn get_sessions(
1454            &self,
1455            sender_key: &str,
1456        ) -> Result<Option<Vec<Session>>, Self::Error> {
1457            self.0.get_sessions(sender_key).await
1458        }
1459
1460        async fn get_inbound_group_session(
1461            &self,
1462            room_id: &RoomId,
1463            session_id: &str,
1464        ) -> Result<Option<InboundGroupSession>, Self::Error> {
1465            self.0.get_inbound_group_session(room_id, session_id).await
1466        }
1467
1468        async fn get_withheld_info(
1469            &self,
1470            room_id: &RoomId,
1471            session_id: &str,
1472        ) -> Result<Option<RoomKeyWithheldEntry>, Self::Error> {
1473            self.0.get_withheld_info(room_id, session_id).await
1474        }
1475
1476        async fn get_withheld_sessions_by_room_id(
1477            &self,
1478            room_id: &RoomId,
1479        ) -> Result<Vec<RoomKeyWithheldEntry>, Self::Error> {
1480            self.0.get_withheld_sessions_by_room_id(room_id).await
1481        }
1482
1483        async fn get_inbound_group_sessions(
1484            &self,
1485        ) -> Result<Vec<InboundGroupSession>, Self::Error> {
1486            self.0.get_inbound_group_sessions().await
1487        }
1488
1489        async fn inbound_group_session_counts(
1490            &self,
1491            backup_version: Option<&str>,
1492        ) -> Result<RoomKeyCounts, Self::Error> {
1493            self.0.inbound_group_session_counts(backup_version).await
1494        }
1495
1496        async fn get_inbound_group_sessions_by_room_id(
1497            &self,
1498            room_id: &RoomId,
1499        ) -> Result<Vec<InboundGroupSession>, Self::Error> {
1500            self.0.get_inbound_group_sessions_by_room_id(room_id).await
1501        }
1502
1503        async fn get_inbound_group_sessions_for_device_batch(
1504            &self,
1505            sender_key: Curve25519PublicKey,
1506            sender_data_type: SenderDataType,
1507            after_session_id: Option<String>,
1508            limit: usize,
1509        ) -> Result<Vec<InboundGroupSession>, Self::Error> {
1510            self.0
1511                .get_inbound_group_sessions_for_device_batch(
1512                    sender_key,
1513                    sender_data_type,
1514                    after_session_id,
1515                    limit,
1516                )
1517                .await
1518        }
1519
1520        async fn inbound_group_sessions_for_backup(
1521            &self,
1522            backup_version: &str,
1523            limit: usize,
1524        ) -> Result<Vec<InboundGroupSession>, Self::Error> {
1525            self.0.inbound_group_sessions_for_backup(backup_version, limit).await
1526        }
1527
1528        async fn mark_inbound_group_sessions_as_backed_up(
1529            &self,
1530            backup_version: &str,
1531            room_and_session_ids: &[(&RoomId, &str)],
1532        ) -> Result<(), Self::Error> {
1533            self.0
1534                .mark_inbound_group_sessions_as_backed_up(backup_version, room_and_session_ids)
1535                .await
1536        }
1537
1538        async fn reset_backup_state(&self) -> Result<(), Self::Error> {
1539            self.0.reset_backup_state().await
1540        }
1541
1542        async fn load_backup_keys(&self) -> Result<BackupKeys, Self::Error> {
1543            self.0.load_backup_keys().await
1544        }
1545
1546        async fn load_dehydrated_device_pickle_key(
1547            &self,
1548        ) -> Result<Option<DehydratedDeviceKey>, Self::Error> {
1549            self.0.load_dehydrated_device_pickle_key().await
1550        }
1551
1552        async fn delete_dehydrated_device_pickle_key(&self) -> Result<(), Self::Error> {
1553            self.0.delete_dehydrated_device_pickle_key().await
1554        }
1555
1556        async fn get_outbound_group_session(
1557            &self,
1558            room_id: &RoomId,
1559        ) -> Result<Option<OutboundGroupSession>, Self::Error> {
1560            self.0.get_outbound_group_session(room_id).await
1561        }
1562
1563        async fn load_tracked_users(&self) -> Result<Vec<TrackedUser>, Self::Error> {
1564            self.0.load_tracked_users().await
1565        }
1566
1567        async fn save_tracked_users(&self, users: &[(&UserId, bool)]) -> Result<(), Self::Error> {
1568            self.0.save_tracked_users(users).await
1569        }
1570
1571        async fn get_device(
1572            &self,
1573            user_id: &UserId,
1574            device_id: &DeviceId,
1575        ) -> Result<Option<DeviceData>, Self::Error> {
1576            self.0.get_device(user_id, device_id).await
1577        }
1578
1579        async fn get_user_devices(
1580            &self,
1581            user_id: &UserId,
1582        ) -> Result<HashMap<OwnedDeviceId, DeviceData>, Self::Error> {
1583            self.0.get_user_devices(user_id).await
1584        }
1585
1586        async fn get_own_device(&self) -> Result<DeviceData, Self::Error> {
1587            self.0.get_own_device().await
1588        }
1589
1590        async fn get_user_identity(
1591            &self,
1592            user_id: &UserId,
1593        ) -> Result<Option<UserIdentityData>, Self::Error> {
1594            self.0.get_user_identity(user_id).await
1595        }
1596
1597        async fn is_message_known(
1598            &self,
1599            message_hash: &OlmMessageHash,
1600        ) -> Result<bool, Self::Error> {
1601            self.0.is_message_known(message_hash).await
1602        }
1603
1604        async fn get_outgoing_secret_requests(
1605            &self,
1606            request_id: &TransactionId,
1607        ) -> Result<Option<GossipRequest>, Self::Error> {
1608            self.0.get_outgoing_secret_requests(request_id).await
1609        }
1610
1611        async fn get_secret_request_by_info(
1612            &self,
1613            secret_info: &SecretInfo,
1614        ) -> Result<Option<GossipRequest>, Self::Error> {
1615            self.0.get_secret_request_by_info(secret_info).await
1616        }
1617
1618        async fn get_unsent_secret_requests(&self) -> Result<Vec<GossipRequest>, Self::Error> {
1619            self.0.get_unsent_secret_requests().await
1620        }
1621
1622        async fn delete_outgoing_secret_requests(
1623            &self,
1624            request_id: &TransactionId,
1625        ) -> Result<(), Self::Error> {
1626            self.0.delete_outgoing_secret_requests(request_id).await
1627        }
1628
1629        async fn get_secrets_from_inbox(
1630            &self,
1631            secret_name: &SecretName,
1632        ) -> Result<Vec<Zeroizing<String>>, Self::Error> {
1633            self.0.get_secrets_from_inbox(secret_name).await
1634        }
1635
1636        async fn delete_secrets_from_inbox(
1637            &self,
1638            secret_name: &SecretName,
1639        ) -> Result<(), Self::Error> {
1640            self.0.delete_secrets_from_inbox(secret_name).await
1641        }
1642
1643        async fn get_room_settings(
1644            &self,
1645            room_id: &RoomId,
1646        ) -> Result<Option<RoomSettings>, Self::Error> {
1647            self.0.get_room_settings(room_id).await
1648        }
1649
1650        async fn get_received_room_key_bundle_data(
1651            &self,
1652            room_id: &RoomId,
1653            user_id: &UserId,
1654        ) -> crate::store::Result<Option<StoredRoomKeyBundleData>, Self::Error> {
1655            self.0.get_received_room_key_bundle_data(room_id, user_id).await
1656        }
1657
1658        async fn has_downloaded_all_room_keys(
1659            &self,
1660            room_id: &RoomId,
1661        ) -> Result<bool, Self::Error> {
1662            self.0.has_downloaded_all_room_keys(room_id).await
1663        }
1664
1665        async fn get_pending_key_bundle_details_for_room(
1666            &self,
1667            room_id: &RoomId,
1668        ) -> Result<Option<RoomPendingKeyBundleDetails>, Self::Error> {
1669            self.0.get_pending_key_bundle_details_for_room(room_id).await
1670        }
1671
1672        async fn get_all_rooms_pending_key_bundles(
1673            &self,
1674        ) -> Result<Vec<RoomPendingKeyBundleDetails>, Self::Error> {
1675            self.0.get_all_rooms_pending_key_bundles().await
1676        }
1677
1678        async fn get_custom_value(&self, key: &str) -> Result<Option<Vec<u8>>, Self::Error> {
1679            self.0.get_custom_value(key).await
1680        }
1681
1682        async fn set_custom_value(&self, key: &str, value: Vec<u8>) -> Result<(), Self::Error> {
1683            self.0.set_custom_value(key, value).await
1684        }
1685
1686        async fn remove_custom_value(&self, key: &str) -> Result<(), Self::Error> {
1687            self.0.remove_custom_value(key).await
1688        }
1689
1690        async fn try_take_leased_lock(
1691            &self,
1692            lease_duration_ms: u32,
1693            key: &str,
1694            holder: &str,
1695        ) -> Result<Option<CrossProcessLockGeneration>, Self::Error> {
1696            self.0.try_take_leased_lock(lease_duration_ms, key, holder).await
1697        }
1698
1699        async fn next_batch_token(&self) -> Result<Option<String>, Self::Error> {
1700            self.0.next_batch_token().await
1701        }
1702
1703        async fn delete_next_batch_token(&self) -> Result<(), Self::Error> {
1704            self.0.delete_next_batch_token().await
1705        }
1706
1707        async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
1708            self.0.get_size().await
1709        }
1710    }
1711
1712    cryptostore_integration_tests!();
1713    cryptostore_integration_tests_time!();
1714}