core_crypto_keystore/connection/
mod.rs

1use std::{fmt, ops::Deref};
2
3use sha2::{Digest as _, Sha256};
4use zeroize::{Zeroize, ZeroizeOnDrop};
5
6pub mod platform {
7    cfg_if::cfg_if! {
8        if #[cfg(target_family = "wasm")] {
9            mod wasm;
10            pub use self::wasm::WasmConnection as KeystoreDatabaseConnection;
11            pub use wasm::storage;
12            pub use self::wasm::storage::WasmStorageTransaction as TransactionWrapper;
13        } else {
14            mod generic;
15            pub use self::generic::SqlCipherConnection as KeystoreDatabaseConnection;
16            pub use self::generic::TransactionWrapper;
17        }
18    }
19}
20
21use std::{ops::DerefMut, sync::Arc};
22
23use async_lock::{Mutex, MutexGuard, Semaphore};
24
25pub use self::platform::*;
26use crate::{
27    CryptoKeystoreError, CryptoKeystoreResult,
28    entities::{Entity, EntityFindParams, EntityTransactionExt, MlsPendingMessage, StringEntityId, UniqueEntity},
29    transaction::KeystoreTransaction,
30};
31
32/// Limit on the length of a blob to be stored in the database.
33///
34/// This limit applies to both SQLCipher-backed stores and WASM.
35/// This limit is conservative on purpose when targeting WASM, as the lower bound that exists is Safari with a limit of 1GB per origin.
36///
37/// See: [SQLite limits](https://www.sqlite.org/limits.html)
38/// See: [IndexedDB limits](https://stackoverflow.com/a/63019999/1934177)
39pub const MAX_BLOB_LEN: usize = 1_000_000_000;
40
41#[cfg(not(target_family = "wasm"))]
42// ? Because of UniFFI async requirements, we need our keystore to be Send as well now
43pub trait DatabaseConnectionRequirements: Sized + Send {}
44#[cfg(target_family = "wasm")]
45// ? On the other hand, things cannot be Send on WASM because of platform restrictions (all things are copied across the FFI)
46pub trait DatabaseConnectionRequirements: Sized {}
47
48/// The key used to encrypt the database.
49#[derive(Clone, Zeroize, ZeroizeOnDrop, derive_more::From, PartialEq, Eq)]
50pub struct DatabaseKey([u8; Self::LEN]);
51
52impl DatabaseKey {
53    pub const LEN: usize = 32;
54
55    pub fn generate() -> DatabaseKey {
56        DatabaseKey(rand::random::<[u8; Self::LEN]>())
57    }
58}
59
60impl fmt::Debug for DatabaseKey {
61    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
62        f.write_str("DatabaseKey(hash=")?;
63        for x in Sha256::digest(self).as_slice().iter().take(10) {
64            fmt::LowerHex::fmt(x, f)?
65        }
66        f.write_str("...)")
67    }
68}
69
70impl AsRef<[u8]> for DatabaseKey {
71    fn as_ref(&self) -> &[u8] {
72        &self.0
73    }
74}
75
76impl Deref for DatabaseKey {
77    type Target = [u8];
78
79    fn deref(&self) -> &Self::Target {
80        &self.0
81    }
82}
83
84impl TryFrom<&[u8]> for DatabaseKey {
85    type Error = CryptoKeystoreError;
86
87    fn try_from(buf: &[u8]) -> Result<Self, Self::Error> {
88        if buf.len() != Self::LEN {
89            Err(CryptoKeystoreError::InvalidDbKeySize {
90                expected: Self::LEN,
91                actual: buf.len(),
92            })
93        } else {
94            Ok(Self(buf.try_into().unwrap()))
95        }
96    }
97}
98
99#[cfg_attr(target_family = "wasm", async_trait::async_trait(?Send))]
100#[cfg_attr(not(target_family = "wasm"), async_trait::async_trait)]
101pub trait DatabaseConnection<'a>: DatabaseConnectionRequirements {
102    type Connection: 'a;
103
104    async fn open(name: &str, key: &DatabaseKey) -> CryptoKeystoreResult<Self>;
105
106    async fn open_in_memory(key: &DatabaseKey) -> CryptoKeystoreResult<Self>;
107
108    async fn update_key(&mut self, new_key: &DatabaseKey) -> CryptoKeystoreResult<()>;
109
110    /// Clear all data from the database and close it.
111    async fn wipe(self) -> CryptoKeystoreResult<()>;
112
113    fn check_buffer_size(size: usize) -> CryptoKeystoreResult<()> {
114        #[cfg(not(target_family = "wasm"))]
115        if size > i32::MAX as usize {
116            return Err(CryptoKeystoreError::BlobTooBig);
117        }
118
119        if size >= MAX_BLOB_LEN {
120            return Err(CryptoKeystoreError::BlobTooBig);
121        }
122
123        Ok(())
124    }
125}
126
127#[derive(Debug, Clone)]
128pub struct Database {
129    pub(crate) conn: Arc<Mutex<Option<KeystoreDatabaseConnection>>>,
130    pub(crate) transaction: Arc<Mutex<Option<KeystoreTransaction>>>,
131    transaction_semaphore: Arc<Semaphore>,
132}
133
134const ALLOWED_CONCURRENT_TRANSACTIONS_COUNT: usize = 1;
135
136/// Interface to fetch from the database either from the connection directly or through a
137/// transaaction
138#[cfg_attr(target_family = "wasm", async_trait::async_trait(?Send))]
139#[cfg_attr(not(target_family = "wasm"), async_trait::async_trait)]
140pub trait FetchFromDatabase: Send + Sync {
141    async fn find<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
142        &self,
143        id: impl AsRef<[u8]> + Send,
144    ) -> CryptoKeystoreResult<Option<E>>;
145
146    async fn find_unique<U: UniqueEntity<ConnectionType = KeystoreDatabaseConnection>>(
147        &self,
148    ) -> CryptoKeystoreResult<U>;
149
150    async fn find_all<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
151        &self,
152        params: EntityFindParams,
153    ) -> CryptoKeystoreResult<Vec<E>>;
154
155    async fn find_many<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
156        &self,
157        ids: &[Vec<u8>],
158    ) -> CryptoKeystoreResult<Vec<E>>;
159    async fn count<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(&self) -> CryptoKeystoreResult<usize>;
160}
161
162// SAFETY: this has mutexes and atomics protecting underlying data so this is safe to share between threads
163unsafe impl Send for Database {}
164// SAFETY: this has mutexes and atomics protecting underlying data so this is safe to share between threads
165unsafe impl Sync for Database {}
166
167/// Where to open a connection
168#[derive(Debug, Clone)]
169pub enum ConnectionType<'a> {
170    /// This connection is persistent at the provided path
171    Persistent(&'a str),
172    /// This connection is transient and lives in memory
173    InMemory,
174}
175
176/// Exclusive access to the database connection
177///
178/// Note that this is only ever constructed when we already hold exclusive access,
179/// and the connection has already been tested to ensure that it is non-empty.
180pub struct ConnectionGuard<'a> {
181    guard: MutexGuard<'a, Option<KeystoreDatabaseConnection>>,
182}
183
184impl<'a> TryFrom<MutexGuard<'a, Option<KeystoreDatabaseConnection>>> for ConnectionGuard<'a> {
185    type Error = CryptoKeystoreError;
186
187    fn try_from(guard: MutexGuard<'a, Option<KeystoreDatabaseConnection>>) -> Result<Self, Self::Error> {
188        guard
189            .is_some()
190            .then_some(Self { guard })
191            .ok_or(CryptoKeystoreError::Closed)
192    }
193}
194
195impl Deref for ConnectionGuard<'_> {
196    type Target = KeystoreDatabaseConnection;
197
198    fn deref(&self) -> &Self::Target {
199        self.guard
200            .as_ref()
201            .expect("we have exclusive access and already checked that the connection exists")
202    }
203}
204
205impl DerefMut for ConnectionGuard<'_> {
206    fn deref_mut(&mut self) -> &mut Self::Target {
207        self.guard
208            .as_mut()
209            .expect("we have exclusive access and already checked that the connection exists")
210    }
211}
212
213// Only the functions in this impl block directly mess with `self.conn`
214impl Database {
215    pub async fn open(location: ConnectionType<'_>, key: &DatabaseKey) -> CryptoKeystoreResult<Self> {
216        let conn = match location {
217            ConnectionType::Persistent(name) => KeystoreDatabaseConnection::open(name, key).await?,
218            ConnectionType::InMemory => KeystoreDatabaseConnection::open_in_memory(key).await?,
219        };
220        let conn = Mutex::new(Some(conn));
221        #[allow(clippy::arc_with_non_send_sync)] // see https://github.com/rustwasm/wasm-bindgen/pull/955
222        let conn = Arc::new(conn);
223        Ok(Self {
224            conn,
225            transaction: Default::default(),
226            transaction_semaphore: Arc::new(Semaphore::new(ALLOWED_CONCURRENT_TRANSACTIONS_COUNT)),
227        })
228    }
229
230    /// Get direct exclusive access to the connection.
231    pub async fn conn(&self) -> CryptoKeystoreResult<ConnectionGuard<'_>> {
232        self.conn.lock().await.try_into()
233    }
234
235    /// Wait for any running transaction to finish, then take the connection out of this database,
236    /// preventing this database from being used again.
237    async fn take(&self) -> CryptoKeystoreResult<KeystoreDatabaseConnection> {
238        let _semaphore = self.transaction_semaphore.acquire_arc().await;
239
240        let mut guard = self.conn.lock().await;
241        guard.take().ok_or(CryptoKeystoreError::Closed)
242    }
243
244    // Close this database connection
245    pub async fn close(&self) -> CryptoKeystoreResult<()> {
246        #[cfg(not(target_family = "wasm"))]
247        self.take().await?;
248
249        #[cfg(target_family = "wasm")]
250        {
251            let conn = self.take().await?;
252            conn.close().await?;
253        }
254        Ok(())
255    }
256}
257
258impl Database {
259    /// Close this database and delete its contents.
260    pub async fn wipe(&self) -> CryptoKeystoreResult<()> {
261        self.take().await?.wipe().await
262    }
263
264    pub async fn migrate_db_key_type_to_bytes(
265        name: &str,
266        old_key: &str,
267        new_key: &DatabaseKey,
268    ) -> CryptoKeystoreResult<()> {
269        KeystoreDatabaseConnection::migrate_db_key_type_to_bytes(name, old_key, new_key).await
270    }
271
272    pub async fn update_key(&mut self, new_key: &DatabaseKey) -> CryptoKeystoreResult<()> {
273        self.conn().await?.update_key(new_key).await
274    }
275
276    /// Waits for the current transaction to be committed or rolled back, then starts a new one.
277    pub async fn new_transaction(&self) -> CryptoKeystoreResult<()> {
278        let semaphore = self.transaction_semaphore.acquire_arc().await;
279        let mut transaction_guard = self.transaction.lock().await;
280        *transaction_guard = Some(KeystoreTransaction::new(semaphore).await?);
281        Ok(())
282    }
283
284    pub async fn commit_transaction(&self) -> CryptoKeystoreResult<()> {
285        let mut transaction_guard = self.transaction.lock().await;
286        let Some(transaction) = transaction_guard.as_ref() else {
287            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
288        };
289        transaction.commit(self).await?;
290        *transaction_guard = None;
291        Ok(())
292    }
293
294    pub async fn rollback_transaction(&self) -> CryptoKeystoreResult<()> {
295        let mut transaction_guard = self.transaction.lock().await;
296        if transaction_guard.is_none() {
297            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
298        };
299        *transaction_guard = None;
300        Ok(())
301    }
302
303    pub async fn child_groups<
304        E: Entity<ConnectionType = KeystoreDatabaseConnection> + crate::entities::PersistedMlsGroupExt + Sync,
305    >(
306        &self,
307        entity: E,
308    ) -> CryptoKeystoreResult<Vec<E>> {
309        let mut conn = self.conn().await?;
310        let persisted_records = entity.child_groups(conn.deref_mut()).await?;
311
312        let transaction_guard = self.transaction.lock().await;
313        let Some(transaction) = transaction_guard.as_ref() else {
314            return Ok(persisted_records);
315        };
316        transaction.child_groups(entity, persisted_records).await
317    }
318
319    pub async fn save<E: Entity<ConnectionType = KeystoreDatabaseConnection> + Sync + EntityTransactionExt>(
320        &self,
321        entity: E,
322    ) -> CryptoKeystoreResult<E> {
323        let transaction_guard = self.transaction.lock().await;
324        let Some(transaction) = transaction_guard.as_ref() else {
325            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
326        };
327        transaction.save_mut(entity).await
328    }
329
330    pub async fn remove<
331        E: Entity<ConnectionType = KeystoreDatabaseConnection> + EntityTransactionExt,
332        S: AsRef<[u8]>,
333    >(
334        &self,
335        id: S,
336    ) -> CryptoKeystoreResult<()> {
337        let transaction_guard = self.transaction.lock().await;
338        let Some(transaction) = transaction_guard.as_ref() else {
339            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
340        };
341        transaction.remove::<E, S>(id).await
342    }
343
344    pub async fn find_pending_messages_by_conversation_id(
345        &self,
346        conversation_id: &[u8],
347    ) -> CryptoKeystoreResult<Vec<MlsPendingMessage>> {
348        let mut conn = self.conn().await?;
349        let persisted_records =
350            MlsPendingMessage::find_all_by_conversation_id(&mut conn, conversation_id, Default::default()).await?;
351
352        let transaction_guard = self.transaction.lock().await;
353        let Some(transaction) = transaction_guard.as_ref() else {
354            return Ok(persisted_records);
355        };
356        transaction
357            .find_pending_messages_by_conversation_id(conversation_id, persisted_records)
358            .await
359    }
360
361    pub async fn remove_pending_messages_by_conversation_id(
362        &self,
363        conversation_id: impl AsRef<[u8]> + Send,
364    ) -> CryptoKeystoreResult<()> {
365        let transaction_guard = self.transaction.lock().await;
366        let Some(transaction) = transaction_guard.as_ref() else {
367            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
368        };
369        transaction
370            .remove_pending_messages_by_conversation_id(conversation_id)
371            .await
372    }
373
374    pub async fn cred_delete_by_credential(&self, cred: Vec<u8>) -> CryptoKeystoreResult<()> {
375        let transaction_guard = self.transaction.lock().await;
376        let Some(transaction) = transaction_guard.as_ref() else {
377            return Err(CryptoKeystoreError::MutatingOperationWithoutTransaction);
378        };
379        transaction.cred_delete_by_credential(cred).await
380    }
381}
382
383#[cfg_attr(target_family = "wasm", async_trait::async_trait(?Send))]
384#[cfg_attr(not(target_family = "wasm"), async_trait::async_trait)]
385impl FetchFromDatabase for Database {
386    async fn find<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
387        &self,
388        id: impl AsRef<[u8]> + Send,
389    ) -> CryptoKeystoreResult<Option<E>> {
390        // If a transaction is in progress...
391        if let Some(transaction) = self.transaction.lock().await.as_ref()
392            //... and it has information about this entity, ...
393            && let Some(cached_record) = transaction.find::<E>(id.as_ref()).await?
394        {
395            // ... return that result
396            return Ok(cached_record);
397        }
398
399        // Otherwise get it from the database
400        let mut conn = self.conn().await?;
401        E::find_one(&mut conn, &id.as_ref().into()).await
402    }
403
404    async fn find_unique<U: UniqueEntity>(&self) -> CryptoKeystoreResult<U> {
405        // If a transaction is in progress...
406        if let Some(transaction) = self.transaction.lock().await.as_ref()
407            //... and it has information about this entity, ...
408            && let Some(cached_record) = transaction.find_unique::<U>().await?
409        {
410            // ... return that result
411            return Ok(cached_record);
412        }
413        // Otherwise get it from the database
414        let mut conn = self.conn().await?;
415        U::find_unique(&mut conn).await
416    }
417
418    async fn find_all<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
419        &self,
420        params: EntityFindParams,
421    ) -> CryptoKeystoreResult<Vec<E>> {
422        let mut conn = self.conn().await?;
423        let persisted_records = E::find_all(&mut conn, params.clone()).await?;
424
425        let transaction_guard = self.transaction.lock().await;
426        let Some(transaction) = transaction_guard.as_ref() else {
427            return Ok(persisted_records);
428        };
429        transaction.find_all(persisted_records, params).await
430    }
431
432    async fn find_many<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(
433        &self,
434        ids: &[Vec<u8>],
435    ) -> CryptoKeystoreResult<Vec<E>> {
436        let entity_ids: Vec<StringEntityId> = ids.iter().map(|id| id.as_slice().into()).collect();
437        let mut conn = self.conn().await?;
438        let persisted_records = E::find_many(&mut conn, &entity_ids).await?;
439
440        let transaction_guard = self.transaction.lock().await;
441        let Some(transaction) = transaction_guard.as_ref() else {
442            return Ok(persisted_records);
443        };
444        transaction.find_many(persisted_records, ids).await
445    }
446
447    async fn count<E: Entity<ConnectionType = KeystoreDatabaseConnection>>(&self) -> CryptoKeystoreResult<usize> {
448        if self.transaction.lock().await.is_some() {
449            // Unfortunately, we have to do this because of possible record id overlap
450            // between cache and db.
451            return Ok(self.find_all::<E>(Default::default()).await?.len());
452        };
453        let mut conn = self.conn().await?;
454        E::count(&mut conn).await
455    }
456}
core_crypto_keystore/connection/mod.rs

core_crypto_keystore/connection/
mod.rs
