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

core_crypto/transaction_context/
mod.rs
