core_crypto/transaction_context/
mod.rs

1//! This module contains the primitives to enable transactional support on a higher level within the
2//! [Session]. All mutating operations need to be done through a [TransactionContext].
3
4use std::sync::Arc;
5
6use async_lock::{Mutex, MutexGuardArc, RwLock};
7use core_crypto_keystore::{
8    CryptoKeystoreError, UniqueArc,
9    entities::ConsumerData,
10    traits::{EntityDatabaseMutation as _, FetchFromDatabase as _},
11};
12pub use error::{Error, Result};
13use openmls_traits::OpenMlsCryptoProvider as _;
14use wire_e2e_identity::pki_env::PkiEnvironment;
15
16use crate::{
17    ClientId, ConversationId, CoreCrypto, KeystoreError, MlsTransport, OpenMlsError, RecursiveError, Session,
18    mls::{self, conversation_cache::ConversationCache},
19    mls_provider::{CryptoProvider, Database},
20};
21pub mod conversation;
22mod credential;
23pub mod e2e_identity;
24mod error;
25mod inner_guard;
26pub mod key_package;
27#[cfg(feature = "proteus")]
28pub mod proteus;
29#[cfg(test)]
30pub mod test_utils;
31
32/// This struct provides transactional support for Core Crypto.
33///
34/// This struct provides mutable access to the internals of Core Crypto. Every operation that
35/// causes data to be persisted needs to be done through this struct. This struct will buffer all
36/// operations in memory and when [TransactionContext::finish] is called, it will persist the data into
37/// the keystore.
38///
39/// Due to uniffi's design, we can't force the context to be dropped after the transaction is
40/// committed. To work around that keep everything important in `TransactionContextInner`;
41/// see `inner` and `take_inner`.
42#[derive(Debug, Clone)]
43pub struct TransactionContext {
44    inner: Arc<RwLock<Option<TransactionContextInner>>>,
45}
46
47#[derive(derive_more::Debug)]
48pub(crate) struct TransactionContextInner {
49    core_crypto: Arc<CoreCrypto>,
50    pending_epoch_changes: Arc<Mutex<Vec<(ConversationId, u64)>>>,
51    #[debug(skip)]
52    transaction: UniqueArc<core_crypto_keystore::Transaction>,
53}
54
55impl CoreCrypto {
56    /// Creates a new transaction. All operations that persist data will be
57    /// buffered in memory and when [TransactionContext::finish] is called, the data will be persisted
58    /// in a single database transaction.
59    pub async fn new_transaction(self: &Arc<Self>) -> Result<TransactionContext> {
60        TransactionContext::new(self.clone()).await
61    }
62}
63
64impl TransactionContext {
65    async fn new(core_crypto: Arc<CoreCrypto>) -> Result<Self> {
66        let transaction = core_crypto
67            .database
68            .new_transaction()
69            .await
70            .map_err(OpenMlsError::wrap("creating new transaction"))?;
71        Ok(Self {
72            inner: Arc::new(RwLock::new(Some(TransactionContextInner {
73                core_crypto,
74                pending_epoch_changes: Default::default(),
75                transaction,
76            }))),
77        })
78    }
79
80    pub(crate) async fn session(&self) -> Result<Session> {
81        let inner = self.inner().await?;
82        inner.core_crypto.mls.read().await.as_ref().cloned().ok_or(
83            RecursiveError::context("Getting mls session from transaction context")(
84                mls::session::Error::MlsNotInitialized,
85            )
86            .into(),
87        )
88    }
89
90    #[cfg(test)]
91    pub(crate) async fn set_session_if_exists(&self, new_session: Session) {
92        let Ok(inner) = self.inner().await else {
93            return;
94        };
95        let mut guard = inner.core_crypto.mls.write().await;
96        if guard.as_ref().is_some() {
97            *guard = Some(new_session)
98        }
99    }
100
101    pub(crate) async fn mls_transport(&self) -> Result<Arc<dyn MlsTransport + 'static>> {
102        let inner = self.inner().await?;
103        inner
104            .core_crypto
105            .mls
106            .read()
107            .await
108            .as_ref()
109            .map(|s| s.transport.clone())
110            .ok_or(
111                RecursiveError::context("Getting mls session from transaction context")(
112                    mls::session::Error::MlsNotInitialized,
113                )
114                .into(),
115            )
116    }
117
118    /// Clones the [CryptoProvider].
119    pub async fn crypto_provider(&self) -> Result<CryptoProvider> {
120        let inner = self.inner().await?;
121        inner
122            .core_crypto
123            .mls
124            .read()
125            .await
126            .as_ref()
127            .map(|s| s.crypto_provider.clone())
128            .ok_or(
129                RecursiveError::context("Getting mls session from transaction context")(
130                    mls::session::Error::MlsNotInitialized,
131                )
132                .into(),
133            )
134    }
135
136    pub(crate) async fn database(&self) -> Result<Arc<Database>> {
137        let inner = self.inner().await?;
138        Ok(inner.core_crypto.database.clone())
139    }
140
141    pub(crate) async fn pki_environment(&self) -> Result<Arc<PkiEnvironment>> {
142        let inner = self.inner().await?;
143        inner
144            .core_crypto
145            .pki_environment
146            .read()
147            .await
148            .as_ref()
149            .map(Clone::clone)
150            .ok_or(Error::PkiEnvironmentUnset)
151    }
152
153    pub(crate) async fn mls_groups(&self) -> Result<MutexGuardArc<ConversationCache>> {
154        let inner = self.inner().await?;
155        let cache = inner
156            .core_crypto
157            .mls
158            .read()
159            .await
160            .as_ref()
161            .map(|session| session.conversation_cache.clone())
162            .ok_or_else(|| {
163                RecursiveError::context("getting mls session from transaction context")(
164                    mls::session::Error::MlsNotInitialized,
165                )
166            })?;
167
168        Ok(cache.lock_arc().await)
169    }
170
171    pub(crate) async fn queue_epoch_changed(&self, conversation_id: ConversationId, epoch: u64) -> Result<()> {
172        let inner = self.inner().await?;
173        inner.pending_epoch_changes.lock().await.push((conversation_id, epoch));
174        Ok(())
175    }
176
177    /// Commits the transaction, meaning it takes all the enqueued operations and persist them into
178    /// the keystore. After that the internal state is switched to invalid, causing errors if
179    /// something is called from this object.
180    pub async fn finish(&self) -> Result<()> {
181        let TransactionContextInner {
182            core_crypto,
183            pending_epoch_changes,
184            transaction: tx,
185        } = self.take_inner().await?;
186
187        let commit_result = tx
188            .commit()
189            .await
190            .map_err(KeystoreError::wrap("commiting transaction"))
191            .map_err(Into::into);
192
193        if let Some(session) = core_crypto.mls.read().await.as_ref() {
194            if commit_result.is_ok() {
195                // We need owned values, so we could just clone the conversation ids, but we don't need the events
196                // anymore, so draining the vector works, too.
197                let mut epoch_changes = pending_epoch_changes.lock().await;
198                for (conversation_id, epoch) in epoch_changes.drain(..) {
199                    session.notify_epoch_changed(conversation_id, epoch).await;
200                }
201            } else {
202                // Commit failed: the keystore is back to its pre-transaction state, but the in-memory
203                // conversation cache may have absorbed mutations that never made it to disk. Clear them
204                // so subsequent reads load fresh state from the keystore.
205                session.conversation_cache.lock().await.clear();
206            }
207        }
208
209        commit_result
210    }
211
212    /// Aborts the transaction, meaning it discards all the enqueued operations.
213    /// After that the internal state is switched to invalid, causing errors if
214    /// something is called from this object.
215    pub async fn abort(&self) -> Result<()> {
216        let inner = self.take_inner().await?;
217
218        // Drop any in-memory conversation state mutated during this transaction; it never reached
219        // the keystore and would otherwise diverge from disk after rollback.
220        if let Some(session) = inner.core_crypto.mls.read().await.as_ref() {
221            session.conversation_cache.lock().await.clear();
222        }
223
224        Ok(())
225    }
226
227    /// Initializes the MLS client of [super::CoreCrypto].
228    pub async fn mls_init(&self, session_id: ClientId, transport: Arc<dyn MlsTransport>) -> Result<()> {
229        let database = self.database().await?;
230        let pki_env = self.pki_environment().await.ok();
231        let crypto_provider = CryptoProvider::new_with_pki_env(database.clone(), pki_env);
232        let session = Session::new(session_id.clone(), crypto_provider, database.into(), transport);
233        self.set_mls_session(session).await?;
234
235        Ok(())
236    }
237
238    /// Set the `mls_session` Arc (also sets it on the transaction's CoreCrypto instance)
239    pub(crate) async fn set_mls_session(&self, session: Session) -> Result<()> {
240        let inner = self.inner().await?;
241        let mut guard = inner.core_crypto.mls.write().await;
242        *guard = Some(session);
243        Ok(())
244    }
245
246    /// see [Session::id]
247    pub async fn client_id(&self) -> Result<ClientId> {
248        let session = self.session().await?;
249        Ok(session.id())
250    }
251
252    /// Generates a random byte array of the specified size
253    pub async fn random_bytes(&self, len: usize) -> Result<Vec<u8>> {
254        use openmls_traits::random::OpenMlsRand as _;
255        self.crypto_provider()
256            .await?
257            .rand()
258            .random_vec(len)
259            .map_err(OpenMlsError::wrap("generating random vector"))
260            .map_err(Into::into)
261    }
262
263    /// Set arbitrary data to be retrieved by [TransactionContext::get_data].
264    /// This is meant to be used as a check point at the end of a transaction.
265    /// The data should be limited to a reasonable size.
266    pub async fn set_data(&self, data: Vec<u8>) -> Result<()> {
267        let inner = self.inner().await?;
268        ConsumerData::from(data)
269            .save(inner.transaction())
270            .map_err(KeystoreError::wrap("saving consumer data"))?;
271        Ok(())
272    }
273
274    /// Get the data that has previously been set by [TransactionContext::set_data].
275    /// This is meant to be used as a check point at the end of a transaction.
276    pub async fn get_data(&self) -> Result<Option<Vec<u8>>> {
277        let inner = self.inner().await?;
278        match inner.transaction.get_unique::<ConsumerData>().await {
279            Ok(maybe_data) => Ok(maybe_data.map(Arc::unwrap_or_clone).map(Into::into)),
280            Err(CryptoKeystoreError::NotFound(..)) => Ok(None),
281            Err(err) => Err(KeystoreError::wrap("finding unique consumer data")(err).into()),
282        }
283    }
284}
285
286impl TransactionContextInner {
287    pub(crate) fn transaction(&self) -> &core_crypto_keystore::Transaction {
288        &self.transaction
289    }
290}
core_crypto/transaction_context/mod.rs

core_crypto/transaction_context/
mod.rs
