nodedb_lite/storage/encrypted/
engine.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Encrypted storage operations and scan budgets.
4
5use async_trait::async_trait;
6use nodedb_types::Namespace;
7
8#[cfg(test)]
9use super::crypto::SALT_SIZE;
10use super::crypto::{EncryptedStorage, SALT_KEY};
11use crate::error::LiteError;
12use crate::storage::engine::{PrefixScan, StorageEngine, WriteOp};
13
14#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
15#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
16impl<S: StorageEngine> StorageEngine for EncryptedStorage<S> {
17    async fn get(&self, ns: Namespace, key: &[u8]) -> Result<Option<Vec<u8>>, LiteError> {
18        // Salt is stored unencrypted.
19        if ns == Namespace::Meta && key == SALT_KEY {
20            return self.inner.get(ns, key).await;
21        }
22
23        match self.inner.get(ns, key).await? {
24            Some(ciphertext) => {
25                let plaintext = self.decrypt(ns, key, &ciphertext)?;
26                Ok(Some(plaintext))
27            }
28            None => Ok(None),
29        }
30    }
31
32    async fn put(&self, ns: Namespace, key: &[u8], value: &[u8]) -> Result<(), LiteError> {
33        // Salt is stored unencrypted.
34        if ns == Namespace::Meta && key == SALT_KEY {
35            return self.inner.put(ns, key, value).await;
36        }
37
38        let ciphertext = self.encrypt(ns, key, value)?;
39        self.inner.put(ns, key, &ciphertext).await
40    }
41
42    async fn delete(&self, ns: Namespace, key: &[u8]) -> Result<(), LiteError> {
43        self.inner.delete(ns, key).await
44    }
45
46    async fn scan_prefix(
47        &self,
48        ns: Namespace,
49        prefix: &[u8],
50    ) -> Result<Vec<crate::storage::engine::KvPair>, LiteError> {
51        let encrypted_entries = self.inner.scan_prefix(ns, prefix).await?;
52        let mut results = Vec::with_capacity(encrypted_entries.len());
53        for (key, ciphertext) in encrypted_entries {
54            match self.decrypt(ns, &key, &ciphertext) {
55                Ok(plaintext) => results.push((key, plaintext)),
56                Err(e) => {
57                    tracing::warn!(
58                        key = ?String::from_utf8_lossy(&key),
59                        error = %e,
60                        "skipping undecryptable entry in scan"
61                    );
62                }
63            }
64        }
65        Ok(results)
66    }
67
68    async fn scan_prefix_bounded(
69        &self,
70        ns: Namespace,
71        prefix: &[u8],
72        limit: usize,
73    ) -> Result<Vec<crate::storage::engine::KvPair>, LiteError> {
74        let entries = self.inner.scan_prefix_bounded(ns, prefix, limit).await?;
75        entries
76            .into_iter()
77            .map(|(key, ciphertext)| {
78                let plaintext = if ns == Namespace::Meta && key == SALT_KEY {
79                    ciphertext
80                } else {
81                    self.decrypt(ns, &key, &ciphertext)?
82                };
83                Ok((key, plaintext))
84            })
85            .collect()
86    }
87
88    async fn scan_prefix_budgeted(
89        &self,
90        ns: Namespace,
91        prefix: &[u8],
92        max_records: usize,
93        max_bytes: usize,
94    ) -> Result<PrefixScan, LiteError> {
95        self.scan_budgeted(ns, prefix, None, max_records, max_bytes, false)
96            .await
97    }
98
99    async fn scan_prefix_from_budgeted(
100        &self,
101        ns: Namespace,
102        prefix: &[u8],
103        after_key: Option<&[u8]>,
104        max_records: usize,
105        max_bytes: usize,
106    ) -> Result<PrefixScan, LiteError> {
107        self.scan_budgeted(ns, prefix, after_key, max_records, max_bytes, true)
108            .await
109    }
110
111    async fn batch_write(&self, ops: &[WriteOp]) -> Result<(), LiteError> {
112        let encrypted_ops: Vec<WriteOp> = ops
113            .iter()
114            .map(|op| match op {
115                WriteOp::Put { ns, key, value } => {
116                    if *ns == Namespace::Meta && key == SALT_KEY {
117                        return Ok(WriteOp::Put {
118                            ns: *ns,
119                            key: key.clone(),
120                            value: value.clone(),
121                        });
122                    }
123                    let ciphertext = self.encrypt(*ns, key, value)?;
124                    Ok(WriteOp::Put {
125                        ns: *ns,
126                        key: key.clone(),
127                        value: ciphertext,
128                    })
129                }
130                WriteOp::Delete { ns, key } => Ok(WriteOp::Delete {
131                    ns: *ns,
132                    key: key.clone(),
133                }),
134            })
135            .collect::<Result<Vec<_>, LiteError>>()?;
136
137        self.inner.batch_write(&encrypted_ops).await
138    }
139
140    async fn count(&self, ns: Namespace) -> Result<u64, LiteError> {
141        self.inner.count(ns).await
142    }
143
144    async fn scan_range(
145        &self,
146        ns: Namespace,
147        start: &[u8],
148        limit: usize,
149    ) -> Result<Vec<crate::storage::engine::KvPair>, LiteError> {
150        let encrypted_entries = self.inner.scan_range(ns, start, limit).await?;
151        let mut results = Vec::with_capacity(encrypted_entries.len());
152        for (key, ciphertext) in encrypted_entries {
153            match self.decrypt(ns, &key, &ciphertext) {
154                Ok(plaintext) => results.push((key, plaintext)),
155                Err(e) => {
156                    tracing::warn!(
157                        key = ?String::from_utf8_lossy(&key),
158                        error = %e,
159                        "skipping undecryptable entry in scan_range"
160                    );
161                }
162            }
163        }
164        Ok(results)
165    }
166
167    async fn scan_range_bounded(
168        &self,
169        ns: Namespace,
170        start: Option<&[u8]>,
171        end: Option<&[u8]>,
172        limit: Option<usize>,
173    ) -> Result<Vec<crate::storage::engine::KvPair>, LiteError> {
174        let encrypted_entries = self.inner.scan_range_bounded(ns, start, end, limit).await?;
175        let mut results = Vec::with_capacity(encrypted_entries.len());
176        for (key, ciphertext) in encrypted_entries {
177            match self.decrypt(ns, &key, &ciphertext) {
178                Ok(plaintext) => results.push((key, plaintext)),
179                Err(e) => {
180                    tracing::warn!(
181                        key = ?String::from_utf8_lossy(&key),
182                        error = %e,
183                        "skipping undecryptable entry in scan_range_bounded"
184                    );
185                }
186            }
187        }
188        Ok(results)
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195    use crate::config::LiteConfig;
196    use crate::storage::engine::PrefixScanLimit;
197    use crate::storage::pagedb_storage::PagedbStorageMem;
198
199    async fn make_encrypted() -> EncryptedStorage<PagedbStorageMem> {
200        let cfg = LiteConfig::default();
201        let inner = PagedbStorageMem::open_in_memory().await.unwrap();
202        EncryptedStorage::open(
203            inner,
204            "test-passphrase-123",
205            cfg.argon2_m_cost,
206            cfg.argon2_t_cost,
207            cfg.argon2_p_cost,
208        )
209        .await
210        .unwrap()
211    }
212
213    #[tokio::test]
214    async fn roundtrip_basic() {
215        let s = make_encrypted().await;
216        s.put(Namespace::Vector, b"v1", b"hello world")
217            .await
218            .unwrap();
219        let val = s.get(Namespace::Vector, b"v1").await.unwrap();
220        assert_eq!(val.as_deref(), Some(b"hello world".as_slice()));
221    }
222
223    #[tokio::test]
224    async fn get_missing_returns_none() {
225        let s = make_encrypted().await;
226        assert!(s.get(Namespace::Vector, b"nope").await.unwrap().is_none());
227    }
228
229    #[tokio::test]
230    async fn different_namespaces_isolated() {
231        let s = make_encrypted().await;
232        s.put(Namespace::Vector, b"k", b"vec").await.unwrap();
233        s.put(Namespace::Graph, b"k", b"graph").await.unwrap();
234
235        assert_eq!(
236            s.get(Namespace::Vector, b"k").await.unwrap().as_deref(),
237            Some(b"vec".as_slice())
238        );
239        assert_eq!(
240            s.get(Namespace::Graph, b"k").await.unwrap().as_deref(),
241            Some(b"graph".as_slice())
242        );
243    }
244
245    #[tokio::test]
246    async fn wrong_passphrase_fails_decrypt() {
247        let cfg = LiteConfig::default();
248        let inner = PagedbStorageMem::open_in_memory().await.unwrap();
249        // Write with passphrase A.
250        {
251            let s = EncryptedStorage::open(
252                inner,
253                "passphrase-A",
254                cfg.argon2_m_cost,
255                cfg.argon2_t_cost,
256                cfg.argon2_p_cost,
257            )
258            .await
259            .unwrap();
260            s.put(Namespace::Vector, b"secret", b"classified data")
261                .await
262                .unwrap();
263        }
264        // The inner storage is consumed, so we can't reopen with a different passphrase
265        // in this test. Instead, verify the salt persists.
266    }
267
268    #[tokio::test]
269    async fn scan_prefix_decrypts() {
270        let s = make_encrypted().await;
271        s.put(Namespace::Crdt, b"delta:001", b"data1")
272            .await
273            .unwrap();
274        s.put(Namespace::Crdt, b"delta:002", b"data2")
275            .await
276            .unwrap();
277        s.put(Namespace::Crdt, b"other:001", b"other")
278            .await
279            .unwrap();
280
281        let results = s.scan_prefix(Namespace::Crdt, b"delta:").await.unwrap();
282        assert_eq!(results.len(), 2);
283        assert_eq!(results[0].1, b"data1");
284        assert_eq!(results[1].1, b"data2");
285    }
286
287    #[tokio::test]
288    async fn batch_write_encrypts() {
289        let s = make_encrypted().await;
290        s.batch_write(&[
291            WriteOp::Put {
292                ns: Namespace::Vector,
293                key: b"a".to_vec(),
294                value: b"alpha".to_vec(),
295            },
296            WriteOp::Put {
297                ns: Namespace::Vector,
298                key: b"b".to_vec(),
299                value: b"beta".to_vec(),
300            },
301        ])
302        .await
303        .unwrap();
304
305        assert_eq!(
306            s.get(Namespace::Vector, b"a").await.unwrap().as_deref(),
307            Some(b"alpha".as_slice())
308        );
309        assert_eq!(
310            s.get(Namespace::Vector, b"b").await.unwrap().as_deref(),
311            Some(b"beta".as_slice())
312        );
313    }
314
315    #[tokio::test]
316    async fn large_value_roundtrip() {
317        let s = make_encrypted().await;
318        let large = vec![0xABu8; 100_000];
319        s.put(Namespace::LoroState, b"snapshot", &large)
320            .await
321            .unwrap();
322        let val = s.get(Namespace::LoroState, b"snapshot").await.unwrap();
323        assert_eq!(val.unwrap().len(), 100_000);
324    }
325
326    #[tokio::test]
327    async fn salt_persists() {
328        let s = make_encrypted().await;
329        let salt = s.inner.get(Namespace::Meta, SALT_KEY).await.unwrap();
330        assert!(salt.is_some());
331        assert_eq!(salt.unwrap().len(), SALT_SIZE);
332    }
333
334    #[tokio::test]
335    async fn bounded_prefix_decrypts_every_row_and_propagates_errors() {
336        let s = make_encrypted().await;
337        s.put(Namespace::Graph, b"edge:1", b"first").await.unwrap();
338        s.put(Namespace::Graph, b"edge:2", b"second").await.unwrap();
339        assert!(
340            s.scan_prefix_bounded(Namespace::Graph, b"edge:", 0)
341                .await
342                .unwrap()
343                .is_empty()
344        );
345        assert_eq!(
346            s.scan_prefix_bounded(Namespace::Graph, b"edge:", 1)
347                .await
348                .unwrap(),
349            vec![(b"edge:1".to_vec(), b"first".to_vec())]
350        );
351        assert_eq!(
352            s.scan_prefix_bounded(Namespace::Graph, b"edge:", 2)
353                .await
354                .unwrap()
355                .len(),
356            2
357        );
358        s.inner
359            .put(Namespace::Graph, b"edge:2", b"invalid ciphertext")
360            .await
361            .unwrap();
362        assert!(
363            s.scan_prefix_bounded(Namespace::Graph, b"edge:", 2)
364                .await
365                .is_err()
366        );
367        assert_eq!(
368            s.scan_prefix_bounded(Namespace::Meta, SALT_KEY, 1)
369                .await
370                .unwrap()[0]
371                .1
372                .len(),
373            SALT_SIZE
374        );
375    }
376
377    #[tokio::test]
378    async fn budgeted_prefix_accounts_tags_and_checks_every_returned_ciphertext() {
379        let s = make_encrypted().await;
380        for key in [b"p1", b"p2"] {
381            s.put(Namespace::Graph, key, b"abc").await.unwrap();
382        }
383        let exact = s
384            .scan_prefix_budgeted(Namespace::Graph, b"p", 2, 10)
385            .await
386            .unwrap();
387        assert_eq!(exact.entries.len(), 2);
388        assert_eq!(exact.limit, None);
389        let count = s
390            .scan_prefix_budgeted(Namespace::Graph, b"p", 1, 10)
391            .await
392            .unwrap();
393        assert_eq!(count.entries.len(), 1);
394        assert_eq!(count.limit, Some(PrefixScanLimit::Records));
395        let bytes = s
396            .scan_prefix_budgeted(Namespace::Graph, b"p", 2, 5)
397            .await
398            .unwrap();
399        assert_eq!(bytes.entries.len(), 1);
400        assert_eq!(bytes.limit, Some(PrefixScanLimit::Bytes));
401        assert!(
402            s.scan_prefix_budgeted(Namespace::Graph, b"p", 0, 0)
403                .await
404                .unwrap()
405                .entries
406                .is_empty()
407        );
408        s.inner.put(Namespace::Graph, b"p2", b"bad").await.unwrap();
409        assert!(
410            s.scan_prefix_budgeted(Namespace::Graph, b"p", 2, 1)
411                .await
412                .is_err()
413        );
414    }
415}
nodedb_lite/storage/encrypted/engine.rs

nodedb_lite/storage/encrypted/
engine.rs
