core_crypto_keystore/connection/
mod.rs

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

core_crypto_keystore/connection/
mod.rs
