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

core_crypto/transaction_context/
mod.rs
