core_crypto/mls/session/
mod.rs

1mod credential;
2pub(crate) mod e2e_identity;
3mod epoch_observer;
4mod error;
5mod history_observer;
6pub(crate) mod id;
7pub(crate) mod key_package;
8pub(crate) mod user_id;
9
10use std::sync::Arc;
11
12use async_lock::{Mutex, RwLock};
13pub use epoch_observer::EpochObserver;
14pub(crate) use error::{Error, Result};
15pub use history_observer::HistoryObserver;
16use openmls_traits::OpenMlsCryptoProvider;
17
18use crate::{
19    ClientId, Credential, HistorySecret, ImmutableDatabase, MlsTransport, OpenMlsError, RecursiveError,
20    mls::{
21        conversation::{Conversation, ConversationIdRef},
22        conversation_cache::ConversationCache,
23    },
24    mls_provider::{CryptoProvider, EntropySeed},
25};
26
27/// An MLS Session enables a user device to communicate via the MLS protocol.
28///
29/// This closely maps to the `Client` term in [RFC 9720], but we avoid that term to avoid ambiguity;
30/// `Client` is very overloaded with distinct meanings.
31///
32/// There is one `Session` per user per device. A session can contain many MLS groups/conversations.
33///
34/// [RFC 9720]: https://www.rfc-editor.org/rfc/rfc9420.html
35#[derive(derive_more::Debug)]
36pub struct Session {
37    id: ClientId,
38    pub(crate) crypto_provider: CryptoProvider,
39    pub(crate) transport: Arc<dyn MlsTransport + 'static>,
40    database: ImmutableDatabase,
41    #[debug("EpochObserver")]
42    pub(crate) epoch_observer: RwLock<Option<Arc<dyn EpochObserver + 'static>>>,
43    #[debug("HistoryObserver")]
44    pub(crate) history_observer: RwLock<Option<Arc<dyn HistoryObserver + 'static>>>,
45    /// LRU cache of live MLS conversations.
46    ///
47    /// Shared across transactions for cache reuse;
48    /// cleared on transaction rollback to avoid serving stale state.
49    pub(crate) conversation_cache: Arc<Mutex<ConversationCache>>,
50}
51
52impl Session {
53    /// Create a new `Session`
54    pub fn new(
55        id: ClientId,
56        crypto_provider: CryptoProvider,
57        database: ImmutableDatabase,
58        transport: Arc<dyn MlsTransport>,
59    ) -> Arc<Self> {
60        Arc::new(Self {
61            id,
62            crypto_provider,
63            transport,
64            database,
65            epoch_observer: RwLock::new(None),
66            history_observer: RwLock::new(None),
67            conversation_cache: Arc::new(Mutex::new(ConversationCache::new())),
68        })
69    }
70
71    /// Get an immutable view of an MLS conversation.
72    ///
73    /// This may be faster than
74    /// [crate::transaction_context::TransactionContext::conversation].
75    pub async fn get_raw_conversation(self: &Arc<Self>, id: &ConversationIdRef) -> Result<Conversation> {
76        Conversation::load(self, id)
77            .await
78            .map_err(RecursiveError::context("getting raw conversation by id"))?
79            .ok_or_else(|| Error::ConversationNotFound(id.to_owned()))
80    }
81
82    /// Checks if a given conversation id exists locally
83    pub async fn conversation_exists(self: &Arc<Self>, id: &ConversationIdRef) -> Result<bool> {
84        match self.get_raw_conversation(id).await {
85            Ok(_) => Ok(true),
86            Err(Error::ConversationNotFound(_)) => Ok(false),
87            Err(e) => Err(e),
88        }
89    }
90
91    /// Generates a random byte array of the specified size
92    pub fn random_bytes(&self, len: usize) -> crate::Result<Vec<u8>> {
93        use openmls_traits::random::OpenMlsRand as _;
94        self.crypto_provider
95            .rand()
96            .random_vec(len)
97            .map_err(OpenMlsError::wrap("generating random vector"))
98            .map_err(Into::into)
99    }
100
101    /// Get read-only access to the database.
102    pub fn database(&self) -> &ImmutableDatabase {
103        &self.database
104    }
105
106    /// see [crate::mls_provider::CryptoProvider::reseed]
107    pub fn reseed(&self, seed: Option<EntropySeed>) -> crate::Result<()> {
108        self.crypto_provider
109            .reseed(seed)
110            .map_err(OpenMlsError::wrap("reseeding mls backend"))
111            .map_err(Into::into)
112    }
113
114    /// Restore from an external [`HistorySecret`].
115    ///
116    /// Returns the public half of the credential this history client was created with. Persisting it
117    /// is the caller's job, and has to happen before this client joins a conversation: a persisted
118    /// conversation refers to the credential of the client which persisted it.
119    pub(crate) async fn restore_from_history_secret(&self, history_secret: HistorySecret) -> Result<Credential> {
120        // store the key package
121        let key_package = history_secret
122            .key_package
123            .store(&self.crypto_provider)
124            .await
125            .map_err(OpenMlsError::wrap("storing key package encapsulation"))?;
126
127        Credential::public_only(&key_package)
128            .map_err(RecursiveError::context("recovering credential from key package"))
129            .map_err(Into::into)
130    }
131
132    /// Retrieves the client's client id. This is free-form and not inspected.
133    pub fn id(&self) -> ClientId {
134        self.id.clone()
135    }
136}
137
138#[cfg(test)]
139mod tests {
140    use core_crypto_keystore::{entities::*, traits::FetchFromDatabase};
141
142    use super::*;
143    use crate::transaction_context::test_utils::EntitiesCount;
144
145    impl Session {
146        /// Count the entities
147        pub async fn count_entities(&self) -> EntitiesCount {
148            let keystore = &self.database;
149            let buffered_commits = keystore.count::<StoredBufferedCommit>().await.unwrap();
150            let credential = keystore.count::<StoredCredential>().await.unwrap();
151            let encryption_keypair = keystore.count::<StoredEncryptionKeyPair>().await.unwrap();
152            let epoch_encryption_keypair = keystore.count::<StoredEpochEncryptionKeypair>().await.unwrap();
153            // `mls_groups` now holds both established and pending rows, distinguished by `is_pending`,
154            // so `group` and `pending_group` come from one `load_all` rather than two separate counts.
155            let all_groups = keystore.load_all::<PersistedMlsGroup>().await.unwrap();
156            let group = all_groups.iter().filter(|group| !group.is_pending).count() as u32;
157            let pending_group = all_groups.iter().filter(|group| group.is_pending).count() as u32;
158            let hpke_private_key = keystore.count::<StoredHpkePrivateKey>().await.unwrap();
159            let key_package = keystore.count::<StoredKeyPackage>().await.unwrap();
160            let pending_messages = keystore.count::<MlsPendingMessage>().await.unwrap();
161            let psk_bundle = keystore.count::<StoredPskBundle>().await.unwrap();
162            EntitiesCount {
163                buffered_commits,
164                credential,
165                encryption_keypair,
166                epoch_encryption_keypair,
167                group,
168                hpke_private_key,
169                key_package,
170                pending_group,
171                pending_messages,
172                psk_bundle,
173            }
174        }
175    }
176
177    mod release_database {
178        use std::sync::Weak;
179
180        use core_crypto_keystore::Database;
181
182        use crate::{
183            CoreCrypto,
184            test_utils::{CoreCryptoTransportSuccessProvider, TestContext, conversation_id},
185        };
186
187        /// Initialize MLS on a fresh `CoreCrypto`, optionally create a conversation, then drop every handle.
188        ///
189        /// Returns a weak reference to the database, which should be dead by then.
190        async fn database_after_dropping_everything(create_conversation: bool) -> Weak<Database> {
191            let database = Database::open_in_memory().unwrap();
192            let weak_database = std::sync::Arc::downgrade(&database);
193            {
194                let case = TestContext::default();
195                let cc = CoreCrypto::new(database);
196                let context = cc.new_transaction().await.unwrap();
197                let credential = case.generate_credential().await;
198                context
199                    .mls_init(
200                        credential.client_id().to_owned(),
201                        std::sync::Arc::new(CoreCryptoTransportSuccessProvider::default()),
202                    )
203                    .await
204                    .unwrap();
205                let credential_ref = context.add_credential(credential).await.unwrap();
206                if create_conversation {
207                    context
208                        .new_conversation(&conversation_id(), &credential_ref, case.cfg.clone())
209                        .await
210                        .unwrap();
211                }
212                context.finish().await.unwrap();
213            }
214            weak_database
215        }
216
217        #[macro_rules_attribute::apply(smol_macros::test)]
218        async fn database_is_released_after_mls_init() {
219            let database = database_after_dropping_everything(false).await;
220            assert_eq!(database.strong_count(), 0);
221        }
222
223        #[macro_rules_attribute::apply(smol_macros::test)]
224        async fn database_is_released_after_creating_a_conversation() {
225            let database = database_after_dropping_everything(true).await;
226            assert_eq!(database.strong_count(), 0);
227        }
228    }
229}
core_crypto/mls/session/mod.rs

core_crypto/mls/session/
mod.rs
