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

core_crypto/transaction_context/
mod.rs
