chronix-engine 0.4.0

Storage engine for Chronix — WAL, memtable, columnar segments, indexes, compaction, and caches
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
//! Encrypting storage backend — transparent AES-256-GCM encryption layer.
//!
//! Wraps any [`StorageBackend`] to provide transparent encryption and
//! decryption of segment data at rest, using the `EncryptionService`
//! from `chronix_security::auth`.
//!
//! # Envelope Format
//!
//! ```text
//! [4-byte LE key_id_len][key_id bytes][12-byte nonce][AES-256-GCM ciphertext + tag]
//! ```
//!
//! The key ID is embedded in each encrypted segment so that key rotation
//! is seamless — segments encrypted with an older key can still be
//! decrypted as long as the key provider retains the retired key.
//!
//! # Thread Safety
//!
//! `EncryptingBackend<B>` is `Send + Sync` whenever `B` is, making it
//! usable as a drop-in replacement in async contexts.

use std::sync::Arc;

use tracing::debug;

use chronix_core::{NamespaceId, ShardId};

use crate::storage::backend::{SegmentPath, StorageBackend};
use crate::storage::error::{Result, StorageError};

/// Transparent encryption wrapper around any [`StorageBackend`].
///
/// Encrypts segment data before writing and decrypts after reading,
/// using AES-256-GCM via [`chronix_security::auth::EncryptionService`].
#[derive(Debug, Clone)]
pub struct EncryptingBackend<B> {
    inner: B,
    encryption: Arc<chronix_security::auth::EncryptionService>,
}

impl<B> EncryptingBackend<B> {
    /// Wrap `inner` backend with transparent encryption.
    pub fn new(inner: B, encryption: Arc<chronix_security::auth::EncryptionService>) -> Self {
        Self { inner, encryption }
    }

    /// Returns a reference to the inner (unencrypted) backend.
    #[must_use]
    pub fn inner(&self) -> &B {
        &self.inner
    }
}

/// Envelope: `[4-byte LE key_id_len][key_id][encrypted_data]`
fn seal(
    encryption: &chronix_security::auth::EncryptionService,
    plaintext: &[u8],
) -> Result<Vec<u8>> {
    let (ciphertext, key_id) = encryption
        .encrypt(plaintext)
        .map_err(|e| StorageError::Io(std::io::Error::other(e.to_string())))?;

    let key_id_bytes = key_id.as_bytes();
    let key_id_len = u32::try_from(key_id_bytes.len()).map_err(|_| {
        StorageError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "key_id too long",
        ))
    })?;

    let mut envelope = Vec::with_capacity(4 + key_id_bytes.len() + ciphertext.len());
    envelope.extend_from_slice(&key_id_len.to_le_bytes());
    envelope.extend_from_slice(key_id_bytes);
    envelope.extend_from_slice(&ciphertext);
    Ok(envelope)
}

/// Parse envelope and decrypt.
fn unseal(
    encryption: &chronix_security::auth::EncryptionService,
    envelope: &[u8],
) -> Result<Vec<u8>> {
    if envelope.len() < 4 {
        return Err(StorageError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "encrypted envelope too short",
        )));
    }

    let key_id_len = u32::from_le_bytes(envelope[..4].try_into().expect("4 bytes")) as usize;

    if envelope.len() < 4 + key_id_len {
        return Err(StorageError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "encrypted envelope truncated at key_id",
        )));
    }

    let key_id = std::str::from_utf8(&envelope[4..4 + key_id_len]).map_err(|_| {
        StorageError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "key_id is not valid UTF-8",
        ))
    })?;

    let ciphertext = &envelope[4 + key_id_len..];

    encryption.decrypt(ciphertext, key_id).map_err(|e| {
        StorageError::Io(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("decryption failed: {e}"),
        ))
    })
}

impl<B: StorageBackend> StorageBackend for EncryptingBackend<B> {
    async fn put_segment(&self, path: &SegmentPath, data: &[u8]) -> Result<()> {
        let envelope = seal(&self.encryption, data)?;
        debug!(
            path = %path,
            plaintext_bytes = data.len(),
            encrypted_bytes = envelope.len(),
            "encrypting segment"
        );
        self.inner.put_segment(path, &envelope).await
    }

    async fn get_segment(&self, path: &SegmentPath) -> Result<Vec<u8>> {
        let envelope = self.inner.get_segment(path).await?;
        let plaintext = unseal(&self.encryption, &envelope)?;
        debug!(
            path = %path,
            encrypted_bytes = envelope.len(),
            plaintext_bytes = plaintext.len(),
            "decrypting segment"
        );
        Ok(plaintext)
    }

    async fn get_range(&self, path: &SegmentPath, offset: u64, length: usize) -> Result<Vec<u8>> {
        // Range reads on encrypted data require decrypting the entire segment
        // because AES-GCM is authenticated — partial reads cannot be verified.
        // This is the correct trade-off: security > performance for range reads.
        let plaintext = self.get_segment(path).await?;
        let end = offset as usize + length;
        if end > plaintext.len() {
            return Err(StorageError::Io(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                format!(
                    "range [{offset}..{end}) exceeds decrypted segment size {}",
                    plaintext.len()
                ),
            )));
        }
        Ok(plaintext[offset as usize..end].to_vec())
    }

    async fn delete_segment(&self, path: &SegmentPath) -> Result<()> {
        self.inner.delete_segment(path).await
    }

    async fn list_segments(
        &self,
        namespace: &NamespaceId,
        shard_id: &ShardId,
    ) -> Result<Vec<SegmentPath>> {
        self.inner.list_segments(namespace, shard_id).await
    }

    async fn exists(&self, path: &SegmentPath) -> Result<bool> {
        self.inner.exists(path).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::local::LocalFsBackend;
    use chronix_security::auth::error::AuthError;
    use chronix_security::auth::{EncryptionService, KeyProvider, SecretKey};

    /// Simple in-memory key provider for tests.
    struct TestKeyProvider {
        key: Vec<u8>,
        key_id: String,
    }

    impl TestKeyProvider {
        fn new() -> Self {
            Self {
                key: vec![0xAB; 32], // 256-bit key
                key_id: "test-key-1".to_string(),
            }
        }
    }

    impl std::fmt::Debug for TestKeyProvider {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.debug_struct("TestKeyProvider")
                .field("key_id", &self.key_id)
                .finish()
        }
    }

    impl KeyProvider for TestKeyProvider {
        fn current_key(&self) -> std::result::Result<(SecretKey, String), AuthError> {
            Ok((SecretKey::new(self.key.clone()), self.key_id.clone()))
        }

        fn key_by_id(&self, key_id: &str) -> std::result::Result<SecretKey, AuthError> {
            if key_id == self.key_id {
                Ok(SecretKey::new(self.key.clone()))
            } else {
                Err(AuthError::KeyNotFound(key_id.to_string()))
            }
        }

        fn key_ids(&self) -> Vec<String> {
            vec![self.key_id.clone()]
        }
    }

    fn make_backend(dir: &std::path::Path) -> EncryptingBackend<LocalFsBackend> {
        let local = LocalFsBackend::new(dir);
        let enc = Arc::new(EncryptionService::new(Box::new(TestKeyProvider::new())));
        EncryptingBackend::new(local, enc)
    }

    fn test_path() -> SegmentPath {
        SegmentPath::new(
            NamespaceId::default_namespace(),
            ShardId(1),
            "test_segment.csx",
        )
        .unwrap()
    }

    #[tokio::test]
    async fn encrypt_decrypt_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        let data = b"Hello, encrypted segment!";
        backend.put_segment(&path, data).await.unwrap();

        // Verify on-disk data is NOT plaintext
        let raw = LocalFsBackend::new(dir.path())
            .get_segment(&path)
            .await
            .unwrap();
        assert_ne!(&raw, data, "on-disk data must be encrypted");

        // Verify decrypted data matches original
        let decrypted = backend.get_segment(&path).await.unwrap();
        assert_eq!(decrypted, data);
    }

    #[tokio::test]
    async fn range_read_after_encryption() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        let data = b"0123456789ABCDEF";
        backend.put_segment(&path, data).await.unwrap();

        let range = backend.get_range(&path, 4, 8).await.unwrap();
        assert_eq!(&range, &data[4..12]);
    }

    #[tokio::test]
    async fn range_read_out_of_bounds() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"short").await.unwrap();

        let err = backend.get_range(&path, 0, 100).await;
        assert!(err.is_err());
    }

    #[tokio::test]
    async fn delete_encrypted_segment() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"delete me").await.unwrap();
        assert!(backend.exists(&path).await.unwrap());

        backend.delete_segment(&path).await.unwrap();
        assert!(!backend.exists(&path).await.unwrap());
    }

    #[tokio::test]
    async fn list_encrypted_segments() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"data").await.unwrap();

        let segments = backend
            .list_segments(&NamespaceId::default_namespace(), &ShardId(1))
            .await
            .unwrap();
        assert_eq!(segments.len(), 1);
        assert_eq!(segments[0].segment_name(), "test_segment.csx");
    }

    #[tokio::test]
    async fn tampered_data_fails_decrypt() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"sensitive data").await.unwrap();

        // Tamper with the encrypted data on disk
        let raw_backend = LocalFsBackend::new(dir.path());
        let mut raw = raw_backend.get_segment(&path).await.unwrap();
        if let Some(last) = raw.last_mut() {
            *last ^= 0xFF; // flip bits in the auth tag
        }
        raw_backend.put_segment(&path, &raw).await.unwrap();

        // Decryption must fail due to authentication tag mismatch
        let result = backend.get_segment(&path).await;
        assert!(result.is_err(), "tampered segment must fail decryption");
    }

    #[tokio::test]
    async fn wrong_key_fails_decrypt() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"secret").await.unwrap();

        // Create a different backend with a different key
        struct OtherKey;
        impl std::fmt::Debug for OtherKey {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                write!(f, "OtherKey")
            }
        }
        impl KeyProvider for OtherKey {
            fn current_key(&self) -> std::result::Result<(SecretKey, String), AuthError> {
                Ok((SecretKey::new(vec![0xCD; 32]), "test-key-1".into()))
            }
            fn key_by_id(&self, _: &str) -> std::result::Result<SecretKey, AuthError> {
                Ok(SecretKey::new(vec![0xCD; 32]))
            }
            fn key_ids(&self) -> Vec<String> {
                vec!["test-key-1".into()]
            }
        }

        let wrong_backend = EncryptingBackend::new(
            LocalFsBackend::new(dir.path()),
            Arc::new(EncryptionService::new(Box::new(OtherKey))),
        );

        let result = wrong_backend.get_segment(&path).await;
        assert!(result.is_err(), "wrong key must fail decryption");
    }

    #[tokio::test]
    async fn large_segment_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        // 1 MiB of pseudo-random data
        let data: Vec<u8> = (0..1_048_576u32).map(|i| (i % 256) as u8).collect();
        backend.put_segment(&path, &data).await.unwrap();

        let decrypted = backend.get_segment(&path).await.unwrap();
        assert_eq!(decrypted, data);
    }

    // ── Additional edge-case tests ────────────────────────────────

    #[tokio::test]
    async fn exists_nonexistent_encrypted() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();
        assert!(!backend.exists(&path).await.unwrap());
    }

    #[tokio::test]
    async fn empty_segment_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        backend.put_segment(&path, b"").await.unwrap();
        let data = backend.get_segment(&path).await.unwrap();
        assert!(data.is_empty());
    }

    #[tokio::test]
    async fn range_read_full_segment() {
        let dir = tempfile::tempdir().unwrap();
        let backend = make_backend(dir.path());
        let path = test_path();

        let original = b"complete data";
        backend.put_segment(&path, original).await.unwrap();

        let range = backend.get_range(&path, 0, original.len()).await.unwrap();
        assert_eq!(&range, original);
    }
}