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::mls_client("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::mls_client("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::mls_client("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(
151                RecursiveError::transaction("getting PKI environment from transaction context")(
152                    e2e_identity::Error::PkiEnvironmentUnset,
153                )
154                .into(),
155            )
156    }
157
158    pub(crate) async fn mls_groups(&self) -> Result<MutexGuardArc<ConversationCache>> {
159        let inner = self.inner().await?;
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::mls_client("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().await?;
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        } = self.take_inner().await?;
191
192        let commit_result = tx
193            .commit()
194            .await
195            .map_err(KeystoreError::wrap("commiting transaction"))
196            .map_err(Into::into);
197
198        if let Some(session) = core_crypto.mls.read().await.as_ref() {
199            if commit_result.is_ok() {
200                // We need owned values, so we could just clone the conversation ids, but we don't need the events
201                // anymore, so draining the vector works, too.
202                let mut epoch_changes = pending_epoch_changes.lock().await;
203                for (conversation_id, epoch) in epoch_changes.drain(..) {
204                    session.notify_epoch_changed(conversation_id, epoch).await;
205                }
206            } else {
207                // Commit failed: the keystore is back to its pre-transaction state, but the in-memory
208                // conversation cache may have absorbed mutations that never made it to disk. Clear them
209                // so subsequent reads load fresh state from the keystore.
210                session.conversation_cache.lock().await.clear();
211            }
212        }
213
214        commit_result
215    }
216
217    /// Aborts the transaction, meaning it discards all the enqueued operations.
218    /// After that the internal state is switched to invalid, causing errors if
219    /// something is called from this object.
220    pub async fn abort(&self) -> Result<()> {
221        let inner = self.take_inner().await?;
222
223        // Drop any in-memory conversation state mutated during this transaction; it never reached
224        // the keystore and would otherwise diverge from disk after rollback.
225        if let Some(session) = inner.core_crypto.mls.read().await.as_ref() {
226            session.conversation_cache.lock().await.clear();
227        }
228
229        Ok(())
230    }
231
232    /// Initializes the MLS client of [super::CoreCrypto].
233    pub async fn mls_init(&self, session_id: ClientId, transport: Arc<dyn MlsTransport>) -> Result<()> {
234        let database = self.database().await?;
235        let pki_env = self.pki_environment().await.ok();
236        let crypto_provider = CryptoProvider::new_with_pki_env(database.clone(), pki_env);
237        let session = Session::new(session_id.clone(), crypto_provider, database.into(), transport);
238        self.set_mls_session(session).await?;
239
240        Ok(())
241    }
242
243    /// Set the `mls_session` Arc (also sets it on the transaction's CoreCrypto instance)
244    pub(crate) async fn set_mls_session(&self, session: Session) -> Result<()> {
245        let inner = self.inner().await?;
246        let mut guard = inner.core_crypto.mls.write().await;
247        *guard = Some(session);
248        Ok(())
249    }
250
251    /// see [Session::id]
252    pub async fn client_id(&self) -> Result<ClientId> {
253        let session = self.session().await?;
254        Ok(session.id())
255    }
256
257    /// Generates a random byte array of the specified size
258    pub async fn random_bytes(&self, len: usize) -> Result<Vec<u8>> {
259        use openmls_traits::random::OpenMlsRand as _;
260        self.crypto_provider()
261            .await?
262            .rand()
263            .random_vec(len)
264            .map_err(OpenMlsError::wrap("generating random vector"))
265            .map_err(Into::into)
266    }
267
268    /// Set arbitrary data to be retrieved by [TransactionContext::get_data].
269    /// This is meant to be used as a check point at the end of a transaction.
270    /// The data should be limited to a reasonable size.
271    pub async fn set_data(&self, data: Vec<u8>) -> Result<()> {
272        let inner = self.inner().await?;
273        ConsumerData::from(data)
274            .save(inner.transaction())
275            .map_err(KeystoreError::wrap("saving consumer data"))?;
276        Ok(())
277    }
278
279    /// Get the data that has previously been set by [TransactionContext::set_data].
280    /// This is meant to be used as a check point at the end of a transaction.
281    pub async fn get_data(&self) -> Result<Option<Vec<u8>>> {
282        let inner = self.inner().await?;
283        match inner.transaction.get_unique::<ConsumerData>().await {
284            Ok(maybe_data) => Ok(maybe_data.map(Arc::unwrap_or_clone).map(Into::into)),
285            Err(CryptoKeystoreError::NotFound(..)) => Ok(None),
286            Err(err) => Err(KeystoreError::wrap("finding unique consumer data")(err).into()),
287        }
288    }
289}
290
291impl TransactionContextInner {
292    pub(crate) fn transaction(&self) -> &core_crypto_keystore::Transaction {
293        &self.transaction
294    }
295}
core_crypto/transaction_context/mod.rs

core_crypto/transaction_context/
mod.rs
