From dac2c5aa9bc059784f82931dc6e8300289c01779 Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Tue, 18 Aug 2026 06:26:53 -0700 Subject: [PATCH 1/2] feat: encrypt durable cell database objects --- crates/celld/durability_encryption.rs | 408 ++++++++++++++++++++++++++ crates/celld/env_vars.rs | 2 + crates/celld/lib.rs | 1 + crates/celld/ltx_repl.rs | 317 +++++++++++++++++++- crates/ltx/src/client/object_store.rs | 61 +++- crates/ltx/src/lib.rs | 4 + docs/README.md | 3 + docs/security.md | 40 +++ scripts/cell-archive-minio.sh | 11 +- 9 files changed, 835 insertions(+), 12 deletions(-) create mode 100644 crates/celld/durability_encryption.rs diff --git a/crates/celld/durability_encryption.rs b/crates/celld/durability_encryption.rs new file mode 100644 index 000000000..78d313c01 --- /dev/null +++ b/crates/celld/durability_encryption.rs @@ -0,0 +1,408 @@ +// Copyright 2026 Deno Land Inc. Apache-2.0 license. + +//! Application-level encryption for cell database bytes in object storage. +//! +//! Coordination objects intentionally remain plaintext: ownership and epoch +//! CAS must be operable without decrypting customer data. LTX bodies and +//! checkpoint/fork SQLite images use this codec, authenticated to their exact +//! bucket key so copying ciphertext across cells, epochs, or checkpoints fails. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use aes_gcm::aead::{Aead, KeyInit, Payload}; +use aes_gcm::{Aes256Gcm, Nonce}; +use anyhow::{bail, Context}; +use base64::Engine; +use celld_ltx::{Error as LtxError, ReplicaObjectCodec}; +use rand::RngCore; +use serde::Deserialize; + +const MAGIC: &[u8; 8] = b"CRCELD01"; +const NONCE_BYTES: usize = 12; +const DATA_KEY_BYTES: usize = 32; +const WRAPPED_DATA_KEY_BYTES: usize = DATA_KEY_BYTES + 16; +const FIXED_HEADER_BYTES: usize = MAGIC.len() + 2 + 8 + NONCE_BYTES + NONCE_BYTES + 2; +const MAXIMUM_KEYS: usize = 16; +const MAXIMUM_KEYRING_BYTES: usize = 16 * 1024; + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct SerializedKeyring { + active_key_id: String, + keys: BTreeMap, +} + +pub struct Aes256GcmDurabilityCodec { + active_key_id: String, + keys: BTreeMap, + allow_plaintext_reads: bool, +} + +impl std::fmt::Debug for Aes256GcmDurabilityCodec { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("Aes256GcmDurabilityCodec") + .field("active_key_id", &self.active_key_id) + .field("key_count", &self.keys.len()) + .field("allow_plaintext_reads", &self.allow_plaintext_reads) + .finish() + } +} + +fn valid_key_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) +} + +impl Aes256GcmDurabilityCodec { + pub fn parse(serialized: &str, allow_plaintext_reads: bool) -> anyhow::Result { + anyhow::ensure!( + serialized.len() <= MAXIMUM_KEYRING_BYTES, + "CELLD_DATA_ENCRYPTION_KEYRING exceeds {MAXIMUM_KEYRING_BYTES} bytes" + ); + let parsed: SerializedKeyring = + serde_json::from_str(serialized).context("decode CELLD_DATA_ENCRYPTION_KEYRING")?; + anyhow::ensure!( + valid_key_id(&parsed.active_key_id), + "CELLD_DATA_ENCRYPTION_KEYRING has an invalid active key ID" + ); + anyhow::ensure!( + !parsed.keys.is_empty() && parsed.keys.len() <= MAXIMUM_KEYS, + "CELLD_DATA_ENCRYPTION_KEYRING must contain 1 through {MAXIMUM_KEYS} keys" + ); + anyhow::ensure!( + parsed.keys.contains_key(&parsed.active_key_id), + "CELLD_DATA_ENCRYPTION_KEYRING does not contain its active key" + ); + let mut keys = BTreeMap::new(); + for (key_id, encoded) in parsed.keys { + anyhow::ensure!( + valid_key_id(&key_id), + "CELLD_DATA_ENCRYPTION_KEYRING has an invalid key ID" + ); + let encoded = encoded.trim(); + anyhow::ensure!(encoded.len() <= 64, "durability key {key_id} is too large"); + let key = base64::engine::general_purpose::STANDARD + .decode(encoded) + .with_context(|| format!("decode durability key {key_id}"))?; + anyhow::ensure!( + key.len() == 32, + "durability key {key_id} must decode to exactly 32 bytes" + ); + keys.insert( + key_id, + Aes256Gcm::new_from_slice(&key).expect("validated AES-256 key length"), + ); + } + Ok(Self { + active_key_id: parsed.active_key_id, + keys, + allow_plaintext_reads, + }) + } + + fn aad(object_key: &str, header: &[u8], domain: &[u8]) -> Vec { + let mut aad = Vec::with_capacity(header.len() + object_key.len() + domain.len() + 2); + aad.extend_from_slice(header); + aad.push(0); + aad.extend_from_slice(object_key.as_bytes()); + aad.push(0); + aad.extend_from_slice(domain); + aad + } + + fn codec_error(message: impl Into) -> LtxError { + LtxError::Other(message.into().into()) + } +} + +impl ReplicaObjectCodec for Aes256GcmDurabilityCodec { + fn name(&self) -> &'static str { + "aes-256-gcm-v1" + } + + fn encode(&self, object_key: &str, plaintext: &[u8]) -> celld_ltx::Result> { + let key_id = self.active_key_id.as_bytes(); + let key_id_length = u16::try_from(key_id.len()) + .map_err(|_| Self::codec_error("durability key ID is too long"))?; + let plaintext_length = u64::try_from(plaintext.len()) + .map_err(|_| Self::codec_error("durability object is too large"))?; + let mut wrapping_nonce = [0_u8; NONCE_BYTES]; + let mut data_nonce = [0_u8; NONCE_BYTES]; + let mut data_key = [0_u8; DATA_KEY_BYTES]; + rand::rngs::OsRng.fill_bytes(&mut wrapping_nonce); + rand::rngs::OsRng.fill_bytes(&mut data_nonce); + rand::rngs::OsRng.fill_bytes(&mut data_key); + + let mut prefix = Vec::with_capacity(FIXED_HEADER_BYTES + key_id.len()); + prefix.extend_from_slice(MAGIC); + prefix.extend_from_slice(&key_id_length.to_be_bytes()); + prefix.extend_from_slice(&plaintext_length.to_be_bytes()); + prefix.extend_from_slice(&wrapping_nonce); + prefix.extend_from_slice(&data_nonce); + prefix.extend_from_slice(&(WRAPPED_DATA_KEY_BYTES as u16).to_be_bytes()); + prefix.extend_from_slice(key_id); + let wrapping_aad = Self::aad(object_key, &prefix, b"data-key"); + let wrapping_key = self + .keys + .get(&self.active_key_id) + .expect("active durability key was validated"); + let wrapped_data_key = wrapping_key + .encrypt( + Nonce::from_slice(&wrapping_nonce), + Payload { + msg: &data_key, + aad: &wrapping_aad, + }, + ) + .map_err(|_| Self::codec_error("durability data-key wrapping failed"))?; + debug_assert_eq!(wrapped_data_key.len(), WRAPPED_DATA_KEY_BYTES); + + let data_cipher = Aes256Gcm::new_from_slice(&data_key) + .expect("generated AES-256 data key has the required length"); + data_key.fill(0); + let mut header = prefix; + header.extend_from_slice(&wrapped_data_key); + let data_aad = Self::aad(object_key, &header, b"database-bytes"); + let ciphertext = data_cipher + .encrypt( + Nonce::from_slice(&data_nonce), + Payload { + msg: plaintext, + aad: &data_aad, + }, + ) + .map_err(|_| Self::codec_error("durability object encryption failed"))?; + header.extend_from_slice(&ciphertext); + Ok(header) + } + + fn decode(&self, object_key: &str, encoded: &[u8]) -> celld_ltx::Result> { + if !encoded.starts_with(MAGIC) { + return if self.allow_plaintext_reads { + Ok(encoded.to_vec()) + } else { + Err(Self::codec_error("durability object is not encrypted")) + }; + } + if encoded.len() < FIXED_HEADER_BYTES { + return Err(Self::codec_error( + "encrypted durability object is truncated", + )); + } + let key_id_length = u16::from_be_bytes([encoded[8], encoded[9]]) as usize; + let prefix_length = FIXED_HEADER_BYTES + .checked_add(key_id_length) + .ok_or_else(|| Self::codec_error("encrypted durability header overflow"))?; + if encoded.len() < prefix_length + WRAPPED_DATA_KEY_BYTES + 16 { + return Err(Self::codec_error( + "encrypted durability object is truncated", + )); + } + let plaintext_length = u64::from_be_bytes( + encoded[10..18] + .try_into() + .expect("fixed plaintext length field"), + ); + let wrapping_nonce = &encoded[18..30]; + let data_nonce = &encoded[30..42]; + let wrapped_data_key_length = u16::from_be_bytes([encoded[42], encoded[43]]) as usize; + if wrapped_data_key_length != WRAPPED_DATA_KEY_BYTES { + return Err(Self::codec_error( + "encrypted durability wrapped data-key length is invalid", + )); + } + let key_id = std::str::from_utf8(&encoded[FIXED_HEADER_BYTES..prefix_length]) + .map_err(|_| Self::codec_error("encrypted durability key ID is invalid UTF-8"))?; + if !valid_key_id(key_id) { + return Err(Self::codec_error("encrypted durability key ID is invalid")); + } + let wrapping_key = self.keys.get(key_id).ok_or_else(|| { + Self::codec_error(format!( + "encrypted durability object requires unknown key {key_id}" + )) + })?; + let wrapped_end = prefix_length + wrapped_data_key_length; + let wrapping_aad = Self::aad(object_key, &encoded[..prefix_length], b"data-key"); + let mut data_key = wrapping_key + .decrypt( + Nonce::from_slice(wrapping_nonce), + Payload { + msg: &encoded[prefix_length..wrapped_end], + aad: &wrapping_aad, + }, + ) + .map_err(|_| Self::codec_error("durability data-key authentication failed"))?; + if data_key.len() != DATA_KEY_BYTES { + data_key.fill(0); + return Err(Self::codec_error( + "encrypted durability data key has an invalid length", + )); + } + let data_cipher = Aes256Gcm::new_from_slice(&data_key) + .expect("unwrapped AES-256 data key has the required length"); + data_key.fill(0); + let data_aad = Self::aad(object_key, &encoded[..wrapped_end], b"database-bytes"); + let plaintext = data_cipher + .decrypt( + Nonce::from_slice(data_nonce), + Payload { + msg: &encoded[wrapped_end..], + aad: &data_aad, + }, + ) + .map_err(|_| Self::codec_error("durability object authentication failed"))?; + if plaintext.len() as u64 != plaintext_length { + return Err(Self::codec_error( + "durability object plaintext length mismatch", + )); + } + Ok(plaintext) + } +} + +pub fn codec_from_env() -> anyhow::Result> { + let required = crate::env_vars::flag("CELLD_DATA_ENCRYPTION_REQUIRED", false)?; + let allow_plaintext_reads = + crate::env_vars::flag("CELLD_DATA_ENCRYPTION_ALLOW_PLAINTEXT_READS", false)?; + let Some(serialized) = crate::env_vars::value("CELLD_DATA_ENCRYPTION_KEYRING")? else { + if required { + bail!("CELLD_DATA_ENCRYPTION_REQUIRED=1 requires CELLD_DATA_ENCRYPTION_KEYRING"); + } + anyhow::ensure!( + !allow_plaintext_reads, + "CELLD_DATA_ENCRYPTION_ALLOW_PLAINTEXT_READS requires an encryption keyring" + ); + return Ok(Arc::new(celld_ltx::PlaintextReplicaObjectCodec)); + }; + anyhow::ensure!( + !serialized.trim().is_empty(), + "CELLD_DATA_ENCRYPTION_KEYRING cannot be empty" + ); + Ok(Arc::new(Aes256GcmDurabilityCodec::parse( + &serialized, + allow_plaintext_reads, + )?)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(byte: u8) -> String { + base64::engine::general_purpose::STANDARD.encode([byte; 32]) + } + + fn codec( + active: &str, + keys: &[(&str, u8)], + allow_plaintext_reads: bool, + ) -> Aes256GcmDurabilityCodec { + let keys = keys + .iter() + .map(|(id, byte)| (id.to_string(), key(*byte))) + .collect::>(); + Aes256GcmDurabilityCodec::parse( + &serde_json::json!({ "active_key_id": active, "keys": keys }).to_string(), + allow_plaintext_reads, + ) + .expect("valid keyring") + } + + #[test] + fn encrypts_and_authenticates_the_exact_object_key() { + let codec = codec("v2", &[("v1", 1), ("v2", 2)], false); + let plaintext = b"SQLite format 3\0private"; + let encoded = codec + .encode("cells/a/ltx/e1/0000/x.ltx", plaintext) + .unwrap(); + assert!(!encoded + .windows(plaintext.len()) + .any(|bytes| bytes == plaintext)); + assert_eq!( + codec.decode("cells/a/ltx/e1/0000/x.ltx", &encoded).unwrap(), + plaintext + ); + assert!(codec + .decode("cells/b/ltx/e1/0000/x.ltx", &encoded) + .unwrap_err() + .to_string() + .contains("authentication failed")); + let mut tampered = encoded; + *tampered.last_mut().unwrap() ^= 1; + assert!(codec + .decode("cells/a/ltx/e1/0000/x.ltx", &tampered) + .is_err()); + } + + #[test] + fn reads_retained_keys_during_rotation_and_rejects_unknown_keys() { + let old = codec("v1", &[("v1", 1)], false); + let encoded = old.encode("object", b"state").unwrap(); + let rotated = codec("v2", &[("v1", 1), ("v2", 2)], false); + assert_eq!(rotated.decode("object", &encoded).unwrap(), b"state"); + let missing = codec("v2", &[("v2", 2)], false); + assert!(missing + .decode("object", &encoded) + .unwrap_err() + .to_string() + .contains("unknown key v1")); + } + + #[test] + fn plaintext_migration_is_explicit() { + assert!(codec("v1", &[("v1", 1)], false) + .decode("object", b"plaintext") + .is_err()); + assert_eq!( + codec("v1", &[("v1", 1)], true) + .decode("object", b"plaintext") + .unwrap(), + b"plaintext" + ); + } + + #[test] + fn parser_rejects_missing_active_short_and_unknown_fields() { + assert!(Aes256GcmDurabilityCodec::parse( + &serde_json::json!({ "active_key_id": "v2", "keys": { "v1": key(1) } }).to_string(), + false, + ) + .is_err()); + assert!(Aes256GcmDurabilityCodec::parse( + &serde_json::json!({ "active_key_id": "v1", "keys": { "v1": "AA==" } }).to_string(), + false, + ) + .is_err()); + assert!(Aes256GcmDurabilityCodec::parse( + &serde_json::json!({ "active_key_id": "v1", "keys": { "v1": key(1) }, "extra": true }) + .to_string(), + false, + ) + .is_err()); + } + + #[test] + fn adversarial_envelopes_never_panic_or_return_plaintext() { + let codec = codec("v1", &[("v1", 1)], false); + for length in 0..512 { + let mut encoded = vec![0_u8; length]; + rand::rngs::OsRng.fill_bytes(&mut encoded); + if length >= MAGIC.len() { + encoded[..MAGIC.len()].copy_from_slice(MAGIC); + } + assert!(codec.decode("object", &encoded).is_err()); + } + + let valid = codec.encode("object", b"private database bytes").unwrap(); + for index in 0..valid.len() { + let mut tampered = valid.clone(); + tampered[index] ^= 1; + assert!(codec.decode("object", &tampered).is_err()); + } + } +} diff --git a/crates/celld/env_vars.rs b/crates/celld/env_vars.rs index 5769696f9..fbb8f235b 100644 --- a/crates/celld/env_vars.rs +++ b/crates/celld/env_vars.rs @@ -17,6 +17,8 @@ pub fn validate() -> anyhow::Result<()> { for name in [ "CELLD_CLOUD", "CELLD_CLOUD_RESTART_ON_DEPLOY", + "CELLD_DATA_ENCRYPTION_ALLOW_PLAINTEXT_READS", + "CELLD_DATA_ENCRYPTION_REQUIRED", "CELLD_LTX_COMPACTION", "CELLD_OUTPUT_GATE", "CELLD_PRESENCE_SHADOW", diff --git a/crates/celld/lib.rs b/crates/celld/lib.rs index 51e37b846..f23da5ffa 100644 --- a/crates/celld/lib.rs +++ b/crates/celld/lib.rs @@ -14,6 +14,7 @@ pub mod control_plane; pub mod dead_node_gc; pub mod deploy; pub mod deployment_auth; +pub mod durability_encryption; pub mod env_vars; /// Test-only SQLite fault injection, ported from celld unchanged. /// diff --git a/crates/celld/ltx_repl.rs b/crates/celld/ltx_repl.rs index 099f4b28f..f5be0168b 100644 --- a/crates/celld/ltx_repl.rs +++ b/crates/celld/ltx_repl.rs @@ -36,6 +36,7 @@ use celld_ltx::Db; use celld_ltx::ObjectStoreClient; use celld_ltx::ObjectStoreConfig; use celld_ltx::Replica; +use celld_ltx::ReplicaObjectCodec; use celld_ltx::TXID; use sha2::Digest; use sha2::Sha256; @@ -142,6 +143,10 @@ pub struct LtxRepl { credentials: Option, /// One connection pool for the whole node, shared by every cell client. store: Arc, + /// Encodes only customer database bytes. Ownership, epoch seals, and other + /// coordination metadata deliberately bypass it so bucket CAS remains + /// independently operable. + durability_codec: Arc, cells: Arc>>, /// Woken when a cell's `committed` advances, so the background loop syncs /// without polling; a slow tick backstops any missed notification. @@ -180,6 +185,19 @@ impl LtxRepl { /// protocol runs against an in-memory bucket instead of S3. #[cfg(test)] pub fn start_with_store_for_test(watch: &Path, store: Arc) -> Self { + Self::start_with_store_and_codec_for_test( + watch, + store, + Arc::new(celld_ltx::PlaintextReplicaObjectCodec), + ) + } + + #[cfg(test)] + fn start_with_store_and_codec_for_test( + watch: &Path, + store: Arc, + durability_codec: Arc, + ) -> Self { let cells: Arc>> = Arc::default(); let dirty = Arc::new(Notify::new()); let slots = Arc::new(Semaphore::new(SYNC_CONCURRENCY)); @@ -192,6 +210,7 @@ impl LtxRepl { region: "auto".into(), credentials: None, store, + durability_codec, cells, dirty, restore_slots: Arc::new(Semaphore::new(RESTORE_DOWNLOAD_CONCURRENCY)), @@ -225,6 +244,7 @@ impl LtxRepl { region: "auto".into(), credentials: None, store, + durability_codec: Arc::new(celld_ltx::PlaintextReplicaObjectCodec), cells, dirty, restore_slots: Arc::new(Semaphore::new(RESTORE_DOWNLOAD_CONCURRENCY)), @@ -243,6 +263,7 @@ impl LtxRepl { credentials: Option, ) -> anyhow::Result { let compaction = compaction_config_from_env()?; + let durability_codec = crate::durability_encryption::codec_from_env()?; // Everything downstream of the store is backend-agnostic already, // so the dialect decides construction and nothing else. let store = match backend { @@ -268,6 +289,7 @@ impl LtxRepl { region, credentials, store, + durability_codec, cells, dirty, restore_slots: Arc::new(Semaphore::new(RESTORE_DOWNLOAD_CONCURRENCY)), @@ -295,7 +317,11 @@ impl LtxRepl { self.credentials.as_ref(), ); config.path = format!("{}cells/{cell}/ltx/e{epoch}", self.prefix); - ObjectStoreClient::with_store(config, self.store.clone()) + ObjectStoreClient::with_store_and_codec( + config, + self.store.clone(), + self.durability_codec.clone(), + ) } /// Highest epoch under `cells//ltx/` that holds any LTX — the newest @@ -346,20 +372,34 @@ impl LtxRepl { use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; let key = ObjPath::from(self.fork_seed_key(cell, name)); + let stored = if name == "database.sqlite" { + self.durability_codec + .encode(key.as_ref(), &bytes) + .map_err(|error| anyhow!("encrypt fork seed {cell}: {error}"))? + } else { + bytes.clone() + }; let create = PutOptions { mode: PutMode::Create, ..Default::default() }; match self .store - .put_opts(&key, PutPayload::from(bytes.clone()), create) + .put_opts(&key, PutPayload::from(stored), create) .await { Ok(_) => Ok(()), Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { let existing = self.store.get(&key).await?.bytes().await?; + let existing = if name == "database.sqlite" { + self.durability_codec + .decode(key.as_ref(), &existing) + .map_err(|error| anyhow!("decrypt existing fork seed {cell}: {error}"))? + } else { + existing.to_vec() + }; anyhow::ensure!( - existing.as_ref() == bytes, + existing == bytes, "fork seed target {cell} already contains a different {name}" ); Ok(()) @@ -423,20 +463,36 @@ impl LtxRepl { use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; let key = ObjPath::from(self.checkpoint_key(cell, checkpoint, name)); + let stored = if name == "database.sqlite" { + self.durability_codec + .encode(key.as_ref(), &bytes) + .map_err(|error| anyhow!("encrypt checkpoint {cell}/{checkpoint}: {error}"))? + } else { + bytes.clone() + }; let create = PutOptions { mode: PutMode::Create, ..Default::default() }; match self .store - .put_opts(&key, PutPayload::from(bytes.clone()), create) + .put_opts(&key, PutPayload::from(stored), create) .await { Ok(_) => Ok(()), Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { let existing = self.store.get(&key).await?.bytes().await?; + let existing = if name == "database.sqlite" { + self.durability_codec + .decode(key.as_ref(), &existing) + .map_err(|error| { + anyhow!("decrypt existing checkpoint {cell}/{checkpoint}: {error}") + })? + } else { + existing.to_vec() + }; anyhow::ensure!( - existing.as_ref() == bytes, + existing == bytes, "checkpoint {cell}/{checkpoint} already contains a different {name}" ); Ok(()) @@ -468,7 +524,13 @@ impl LtxRepl { ); let database_key = ObjPath::from(self.checkpoint_key(source_cell, checkpoint_id, "database.sqlite")); - let sqlite = self.store.get(&database_key).await?.bytes().await?.to_vec(); + let encoded = self.store.get(&database_key).await?.bytes().await?; + let sqlite = self + .durability_codec + .decode(database_key.as_ref(), &encoded) + .map_err(|error| { + anyhow!("decrypt checkpoint {source_cell}/{checkpoint_id}: {error}") + })?; anyhow::ensure!( sqlite.len() as u64 == manifest.sqlite_bytes, "checkpoint byte count mismatch" @@ -565,7 +627,11 @@ impl LtxRepl { "unsupported fork seed format" ); let database = ObjPath::from(self.fork_seed_key(cell, "database.sqlite")); - let sqlite = self.store.get(&database).await?.bytes().await?; + let encoded = self.store.get(&database).await?.bytes().await?; + let sqlite = self + .durability_codec + .decode(database.as_ref(), &encoded) + .map_err(|error| anyhow!("decrypt fork seed for {cell}: {error}"))?; anyhow::ensure!( sqlite.len() as u64 == manifest.sqlite_bytes, "fork seed byte count mismatch" @@ -1541,9 +1607,48 @@ fn node_config( #[cfg(test)] mod fork_seed_tests { use super::*; + use base64::Engine; use celld_ltx::object_store::memory::InMemory; use celld_ltx::object_store::path::Path as ObjPath; use celld_ltx::object_store::PutPayload; + use futures_util::TryStreamExt; + + fn encrypted_codec(active: &str, keys: &[(&str, u8)]) -> Arc { + let keys = keys + .iter() + .map(|(id, byte)| { + ( + id.to_string(), + base64::engine::general_purpose::STANDARD.encode([*byte; 32]), + ) + }) + .collect::>(); + Arc::new( + crate::durability_encryption::Aes256GcmDurabilityCodec::parse( + &serde_json::json!({ "active_key_id": active, "keys": keys }).to_string(), + false, + ) + .unwrap(), + ) + } + + async fn raw_objects(store: &Arc, prefix: &str) -> Vec<(String, Vec)> { + let prefix = ObjPath::from(prefix); + let mut listed = store.list(Some(&prefix)); + let mut objects = Vec::new(); + while let Some(meta) = listed.try_next().await.unwrap() { + let bytes = store + .get(&meta.location) + .await + .unwrap() + .bytes() + .await + .unwrap(); + objects.push((meta.location.to_string(), bytes.to_vec())); + } + objects.sort_by(|left, right| left.0.cmp(&right.0)); + objects + } fn activation<'a>(cell: &'a str, fresh: bool) -> ActivationOptions<'a> { ActivationOptions { @@ -1685,6 +1790,204 @@ mod fork_seed_tests { assert_eq!(source_value, "source-advanced"); } + #[tokio::test] + async fn encrypted_ltx_checkpoint_and_fork_survive_rotation_and_restore() { + let store: Arc = Arc::new(InMemory::new()); + let old_directory = tempfile::tempdir().unwrap(); + let old = LtxRepl::start_with_store_and_codec_for_test( + old_directory.path(), + store.clone(), + encrypted_codec("old", &[("old", 1)]), + ); + let source = old.activate(activation("source", true)).await.unwrap(); + { + let connection = rusqlite::Connection::open(&source.path).unwrap(); + connection + .execute_batch( + "CREATE TABLE state(key TEXT PRIMARY KEY, value TEXT NOT NULL);\n\ + INSERT INTO state VALUES ('phase', 'encrypted-old');", + ) + .unwrap(); + } + old.await_durable("source", 1, 1).await.unwrap(); + old.publish_checkpoint("source", 1, "checkpoint-1") + .await + .unwrap(); + // Create-or-verify compares decrypted bytes. A randomized nonce must + // not make an exact checkpoint retry look like conflicting content. + old.publish_checkpoint("source", 1, "checkpoint-1") + .await + .unwrap(); + + let old_ltx = raw_objects(&store, "cells/source/ltx/e1/").await; + assert!(!old_ltx.is_empty()); + assert!(old_ltx + .iter() + .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"old" })); + let checkpoint = store + .get(&ObjPath::from( + "cells/source/checkpoints/checkpoint-1/database.sqlite", + )) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert!(checkpoint.starts_with(b"CRCELD01")); + assert!(!checkpoint + .windows(b"encrypted-old".len()) + .any(|window| window == b"encrypted-old")); + + let missing_directory = tempfile::tempdir().unwrap(); + let missing = LtxRepl::start_with_store_and_codec_for_test( + missing_directory.path(), + store.clone(), + encrypted_codec("new", &[("new", 2)]), + ); + let error = match missing + .activate(ActivationOptions { + cell: "source", + epoch: 2, + fresh: false, + took_over: true, + resume_local: false, + }) + .await + { + Ok(_) => panic!("restore succeeded without the required old key"), + Err(error) => error, + }; + assert!(error.to_string().contains("unknown key old")); + + let rotated_directory = tempfile::tempdir().unwrap(); + let rotated = LtxRepl::start_with_store_and_codec_for_test( + rotated_directory.path(), + store.clone(), + encrypted_codec("new", &[("old", 1), ("new", 2)]), + ); + let restored = rotated + .activate(ActivationOptions { + cell: "source", + epoch: 2, + fresh: false, + took_over: true, + resume_local: false, + }) + .await + .unwrap(); + assert!(restored.restored); + let connection = rusqlite::Connection::open(&restored.path).unwrap(); + let value: String = connection + .query_row("SELECT value FROM state WHERE key = 'phase'", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(value, "encrypted-old"); + connection + .execute( + "UPDATE state SET value = 'encrypted-new' WHERE key = 'phase'", + [], + ) + .unwrap(); + drop(connection); + rotated.await_durable("source", 2, 1).await.unwrap(); + let new_ltx = raw_objects(&store, "cells/source/ltx/e2/").await; + assert!(!new_ltx.is_empty()); + assert!(new_ltx + .iter() + .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"new" })); + + rotated + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", false) + .await + .unwrap(); + rotated + .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", false) + .await + .unwrap(); + let fork_object = store + .get(&ObjPath::from("cells/fork/fork-seed/database.sqlite")) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert!(fork_object.starts_with(b"CRCELD01")); + assert_eq!(&fork_object[44..47], b"new"); + + let fork = rotated.activate(activation("fork", true)).await.unwrap(); + assert!(fork.restored); + let fork_connection = rusqlite::Connection::open(&fork.path).unwrap(); + let fork_value: String = fork_connection + .query_row("SELECT value FROM state WHERE key = 'phase'", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(fork_value, "encrypted-old"); + + // The supported offline export path restores through the same codec. + let exported = rotated.restore_snapshot("source").await.unwrap().unwrap(); + let exported_connection = rusqlite::Connection::open(exported.path()).unwrap(); + let exported_value: String = exported_connection + .query_row("SELECT value FROM state WHERE key = 'phase'", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(exported_value, "encrypted-new"); + + // The supported offline import path encrypts its new LTX lineage and + // verifies it by immediately restoring through the configured codec. + let import_directory = tempfile::tempdir().unwrap(); + let import_database = import_directory.path().join("import.sqlite"); + let import_connection = rusqlite::Connection::open(&import_database).unwrap(); + import_connection + .execute_batch( + "CREATE TABLE imported(value TEXT NOT NULL);\n\ + INSERT INTO imported VALUES ('encrypted-import');", + ) + .unwrap(); + drop(import_connection); + rotated + .seed_import_epoch("imported", 1, &import_database) + .await + .unwrap(); + let imported_ltx = raw_objects(&store, "cells/imported/ltx/e1/").await; + assert!(!imported_ltx.is_empty()); + assert!(imported_ltx + .iter() + .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"new" })); + + let (tampered_key, mut tampered_bytes) = new_ltx.into_iter().next().unwrap(); + *tampered_bytes.last_mut().unwrap() ^= 1; + store + .put( + &ObjPath::from(tampered_key), + PutPayload::from(tampered_bytes), + ) + .await + .unwrap(); + let corrupt_directory = tempfile::tempdir().unwrap(); + let corrupt = LtxRepl::start_with_store_and_codec_for_test( + corrupt_directory.path(), + store.clone(), + encrypted_codec("new", &[("old", 1), ("new", 2)]), + ); + let error = match corrupt + .activate(ActivationOptions { + cell: "source", + epoch: 3, + fresh: false, + took_over: true, + resume_local: false, + }) + .await + { + Ok(_) => panic!("restore accepted tampered ciphertext"), + Err(error) => error, + }; + assert!(error.to_string().contains("authentication failed")); + } + #[tokio::test] async fn active_snapshots_use_independent_temporary_directories() { let directory = tempfile::tempdir().unwrap(); diff --git a/crates/ltx/src/client/object_store.rs b/crates/ltx/src/client/object_store.rs index 538fc053f..693a455c6 100644 --- a/crates/ltx/src/client/object_store.rs +++ b/crates/ltx/src/client/object_store.rs @@ -46,6 +46,36 @@ use crate::TXID; use super::ReplicaClient; +/// Transforms durable LTX object bytes without changing their bucket keys. +/// +/// Hosts use this boundary for application-level encryption. Listings, +/// ownership metadata, and epoch seals remain ordinary object-store data, while +/// every LTX body is encoded before upload and decoded after download. The +/// object key is part of the codec input so authenticated codecs can prevent a +/// ciphertext from being copied into another cell or epoch. +pub trait ReplicaObjectCodec: Send + Sync { + fn name(&self) -> &'static str; + fn encode(&self, object_key: &str, plaintext: &[u8]) -> Result>; + fn decode(&self, object_key: &str, encoded: &[u8]) -> Result>; +} + +#[derive(Debug, Default)] +pub struct PlaintextReplicaObjectCodec; + +impl ReplicaObjectCodec for PlaintextReplicaObjectCodec { + fn name(&self) -> &'static str { + "plaintext" + } + + fn encode(&self, _object_key: &str, plaintext: &[u8]) -> Result> { + Ok(plaintext.to_vec()) + } + + fn decode(&self, _object_key: &str, encoded: &[u8]) -> Result> { + Ok(encoded.to_vec()) + } +} + /// The standard Litestream S3 metadata key for an LTX header timestamp. const METADATA_KEY_TIMESTAMP: &str = "litestream-timestamp"; @@ -440,6 +470,7 @@ fn match_filebase(host: &str) -> Option { pub struct ObjectStoreClient { store: tokio::sync::OnceCell>, config: ObjectStoreConfig, + codec: Arc, } impl std::fmt::Debug for ObjectStoreClient { @@ -447,6 +478,7 @@ impl std::fmt::Debug for ObjectStoreClient { f.debug_struct("ObjectStoreClient") .field("config", &self.config) .field("initialized", &self.store.initialized()) + .field("codec", &self.codec.name()) .finish() } } @@ -457,6 +489,7 @@ impl ObjectStoreClient { ObjectStoreClient { store: tokio::sync::OnceCell::new(), config, + codec: Arc::new(PlaintextReplicaObjectCodec), } } @@ -468,6 +501,25 @@ impl ObjectStoreClient { ObjectStoreClient { store: cell, config, + codec: Arc::new(PlaintextReplicaObjectCodec), + } + } + + /// Create a client over a shared store with a host-owned durable-object + /// codec. The codec is intentionally applied only to LTX bodies; callers + /// retain ordinary access to coordination metadata through the shared + /// store. + pub fn with_store_and_codec( + config: ObjectStoreConfig, + store: Arc, + codec: Arc, + ) -> Self { + let cell = tokio::sync::OnceCell::new(); + cell.set(store).ok(); + ObjectStoreClient { + store: cell, + config, + codec, } } @@ -581,7 +633,7 @@ impl ReplicaClient for ObjectStoreClient { }; let bytes = result.bytes().await.map_err(map_os_error)?; - Ok(bytes.to_vec()) + self.codec.decode(key.as_ref(), &bytes) } async fn write_ltx_file( @@ -605,13 +657,14 @@ impl ReplicaClient for ObjectStoreClient { AttributeValue::from(format_rfc3339_nano(header.timestamp)?), ); let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid)); + let encoded = self.codec.encode(key.as_ref(), data)?; // Multipart threshold: < 5 MiB → single PUT; ≥ 5 MiB → multipart with // fixed-size parts. Ported from the Go uploader's 5 MiB PartSize default // (s3/replica_client.go:99, brief §5.1). let part_size = self.config.effective_part_size(); - if data.len() < MULTIPART_THRESHOLD { - let payload = PutPayload::from(data.to_vec()); + if encoded.len() < MULTIPART_THRESHOLD { + let payload = PutPayload::from(encoded); let options = PutOptions { attributes, ..Default::default() @@ -631,7 +684,7 @@ impl ReplicaClient for ObjectStoreClient { .map_err(|e| Error::Other(format!("replica: upload to {key}: {e}").into()))?; // Upload in fixed-size parts (each ≥ 5 MiB except possibly the last, // matching object_store's part-size requirement). - for chunk in data.chunks(part_size.max(MULTIPART_THRESHOLD)) { + for chunk in encoded.chunks(part_size.max(MULTIPART_THRESHOLD)) { upload .put_part(PutPayload::from(chunk.to_vec())) .await diff --git a/crates/ltx/src/lib.rs b/crates/ltx/src/lib.rs index 4540618a5..236843632 100644 --- a/crates/ltx/src/lib.rs +++ b/crates/ltx/src/lib.rs @@ -70,6 +70,10 @@ pub use client::file::FileReplicaClient; #[cfg(feature = "s3")] pub use client::object_store::ObjectStoreClient; +/// Codec boundary for durable LTX bodies, behind the object-store feature. +#[cfg(feature = "s3")] +pub use client::object_store::{PlaintextReplicaObjectCodec, ReplicaObjectCodec}; + /// Configuration for the S3/R2/MinIO backend, behind the `s3` feature. /// Re-exported from [`crate::client::object_store::ObjectStoreConfig`]. #[cfg(feature = "s3")] diff --git a/docs/README.md b/docs/README.md index 55f5002f0..e103eb7cc 100644 --- a/docs/README.md +++ b/docs/README.md @@ -336,6 +336,9 @@ For the full list, run `celld -h`. This table shows the primary settings: | `CELLD_MAX_RESIDENT_CELLS` | The hard limit for resident cells, enforced at admission | | `CELLD_MAX_RSS_MB` | The memory threshold for pressure shedding, applied to the memory that the cells hold (default: 80% of the available memory; 0 disables the threshold and the absolute cap) | | `CELLD_OUTPUT_GATE` | The default is `1`, so celld proves each write durable before it acknowledges the write. Set `0` to remove the replication wait and accept possible loss of an acknowledged write | +| `CELLD_DATA_ENCRYPTION_KEYRING` | Secret JSON containing `active_key_id` and a map of versioned key IDs to base64 32-byte AES key-encryption keys. When set, each LTX body and SQLite checkpoint/fork image gets a random data key wrapped by the active key-encryption key; both AES-256-GCM layers are authenticated to the exact object key | +| `CELLD_DATA_ENCRYPTION_REQUIRED` | Set to `1` on fleets that must never write plaintext customer database bytes. Startup fails unless a valid keyring is present | +| `CELLD_DATA_ENCRYPTION_ALLOW_PLAINTEXT_READS` | Temporary migration-only switch. Set to `1` with a keyring to read legacy plaintext LTX while all new durable objects are encrypted. Do not enable it on a new fleet | | `CELLD_DEPLOYMENT_VERIFY_KEYS_FILE` | JSON map of release-key IDs to base64 Ed25519 public keys. When set, every bucket deployment pointer must carry a valid signature | | `CELLD_LTX_COMPACTION` | The default is `1`: celld creates additive L1 objects, and a takeover reads tens of objects instead of thousands. Set `0` on every node of a mixed fleet until all nodes can read v0.5.2 block objects, because an old reader cannot take over a cell after its first L1 publication | | `CELLD_LTX_COMPACTION_MIN_TXIDS` | The durable TXID distance that queues an L1 attempt (default: 256) | diff --git a/docs/security.md b/docs/security.md index 4770bab94..6380e050c 100644 --- a/docs/security.md +++ b/docs/security.md @@ -97,6 +97,46 @@ A person who holds the bucket credentials controls the fleet. Give each credential access to one fleet bucket only, and replace a credential after a suspected disclosure. +### Encrypt customer database objects + +Set `CELLD_DATA_ENCRYPTION_KEYRING` to a secret JSON object containing one +active key and any retained read keys: + +```json +{ + "active_key_id": "2026-08", + "keys": { + "2026-07": "BASE64_32_BYTE_KEY", + "2026-08": "BASE64_32_BYTE_KEY" + } +} +``` + +Celld generates a random 256-bit data key for every LTX body and SQLite +checkpoint/fork image, encrypts the database bytes with AES-256-GCM, and wraps +that data key with the active versioned AES-256-GCM key-encryption key. Both +authenticated-data domains bind the envelope header and exact bucket key, so a +copied or modified ciphertext fails restore. Ownership records, epoch seals, +deployment pointers, and checkpoint manifests remain plaintext because they +are coordination and integrity metadata rather than customer database bytes. + +Set `CELLD_DATA_ENCRYPTION_REQUIRED=1` on a fleet that carries customer data. +Startup then fails without a valid keyring. Rotate without downtime by adding a +new key-encryption key, making it active, and retaining every old key until no +LTX, checkpoint, fork seed, rollback deployment, or retained object references +it. Removing a still-referenced key makes restore fail closed. + +`CELLD_DATA_ENCRYPTION_ALLOW_PLAINTEXT_READS=1` is only for a reviewed migration +of an existing plaintext fleet. New writes are still encrypted, but old objects +may be read. New fleets must leave it disabled. The keyring itself must come +from a secret manager or KMS-protected deployment channel; never place it in a +Worker bundle, bucket object, command line, repository, or log. + +Nodes that predate this envelope format cannot read encrypted LTX. Upgrade the +complete fleet before enabling encryption, then enable the same retained +keyring on every node as one coordinated configuration change. Do not roll back +to a pre-encryption binary after the first encrypted object is published. + ### Require signed deployments An operator can make bucket deployment pointers fail closed by setting diff --git a/scripts/cell-archive-minio.sh b/scripts/cell-archive-minio.sh index 79fc1d1cb..7e0fb4402 100755 --- a/scripts/cell-archive-minio.sh +++ b/scripts/cell-archive-minio.sh @@ -22,6 +22,13 @@ readonly ACCESS_KEY SECRET_KEY="$(printf '%s' "$RUN_ID-secret" | shasum -a 256 | cut -c1-32)" readonly SECRET_KEY export TEST_ROOT +# Test-only key material: exercise the production image's required encrypted +# import/export path against the real object-store API. Docker receives only +# the environment variable name, matching secret-injected deployments. +TEST_KEY="$(printf '00000000000000000000000000000000' | base64 | tr -d '\n')" +readonly TEST_KEY +export CELLD_DATA_ENCRYPTION_KEYRING="{\"active_key_id\":\"archive-test\",\"keys\":{\"archive-test\":\"$TEST_KEY\"}}" +export CELLD_DATA_ENCRYPTION_REQUIRED=1 if [[ "$BACKEND" != 'minio' && "$BACKEND" != 'gcs' ]]; then echo 'CELLD_ARCHIVE_BACKEND must be minio or gcs' >&2 @@ -165,7 +172,8 @@ celld() { docker run --rm --network "$NETWORK" --user "$(id -u):$(id -g)" \ -v "$TEST_ROOT:/archive" \ -e "AWS_ACCESS_KEY_ID=$ACCESS_KEY" -e "AWS_SECRET_ACCESS_KEY=$SECRET_KEY" \ - -e AWS_REGION=us-east-1 "$CELLD_IMAGE" "$@" + -e AWS_REGION=us-east-1 -e CELLD_DATA_ENCRYPTION_KEYRING \ + -e CELLD_DATA_ENCRYPTION_REQUIRED "$CELLD_IMAGE" "$@" else "$CELLD_ARCHIVE_BINARY" "$@" fi @@ -181,6 +189,7 @@ fi celld cell import Knowledge:test --input "$archive_root/source.sqlite" \ "${storage_args[@]}" --offline +test "$(object_cat 'cells/Knowledge:test/ltx/e1/0000/0000000000000001-0000000000000001.ltx' | head -c 8)" = 'CRCELD01' celld cell export Knowledge:test --output "$archive_root/export.sqlite" \ "${storage_args[@]}" From 2c3d01148e1d5f2d00ce0106242bd86614b35b1e Mon Sep 17 00:00:00 2001 From: Harjot Gill Date: Tue, 18 Aug 2026 06:46:44 -0700 Subject: [PATCH 2/2] fix: align encrypted object accounting --- crates/celld/durability_encryption.rs | 18 ++++++++ crates/celld/ltx_repl.rs | 44 ++++++++++-------- crates/ltx/src/client/object_store.rs | 66 ++++++++++++++++++++++++++- docs/security.md | 3 ++ 4 files changed, 110 insertions(+), 21 deletions(-) diff --git a/crates/celld/durability_encryption.rs b/crates/celld/durability_encryption.rs index 78d313c01..6f6d31c50 100644 --- a/crates/celld/durability_encryption.rs +++ b/crates/celld/durability_encryption.rs @@ -265,6 +265,22 @@ impl ReplicaObjectCodec for Aes256GcmDurabilityCodec { } } +#[cfg(test)] +pub(crate) fn envelope_key_id_for_test(encoded: &[u8]) -> anyhow::Result<&str> { + anyhow::ensure!( + encoded.starts_with(MAGIC) && encoded.len() >= FIXED_HEADER_BYTES, + "not a complete encrypted durability header" + ); + let key_id_length = u16::from_be_bytes([encoded[8], encoded[9]]) as usize; + let end = FIXED_HEADER_BYTES + .checked_add(key_id_length) + .context("encrypted durability key ID length overflow")?; + let key_id = encoded + .get(FIXED_HEADER_BYTES..end) + .context("encrypted durability key ID is truncated")?; + std::str::from_utf8(key_id).context("encrypted durability key ID is invalid UTF-8") +} + pub fn codec_from_env() -> anyhow::Result> { let required = crate::env_vars::flag("CELLD_DATA_ENCRYPTION_REQUIRED", false)?; let allow_plaintext_reads = @@ -320,6 +336,8 @@ mod tests { let encoded = codec .encode("cells/a/ltx/e1/0000/x.ltx", plaintext) .unwrap(); + assert_eq!(envelope_key_id_for_test(&encoded).unwrap(), "v2"); + assert!(envelope_key_id_for_test(b"CRCELD01").is_err()); assert!(!encoded .windows(plaintext.len()) .any(|bytes| bytes == plaintext)); diff --git a/crates/celld/ltx_repl.rs b/crates/celld/ltx_repl.rs index f5be0168b..120f38578 100644 --- a/crates/celld/ltx_repl.rs +++ b/crates/celld/ltx_repl.rs @@ -74,6 +74,7 @@ const COMPACTION_MAX_FILES: usize = 256; const COMPACTION_MAX_INPUT_BYTES: u64 = 64 * 1024 * 1024; const FORK_SEED_FORMAT: &str = "celld-sqlite-fork-seed-v1"; +const DATABASE_OBJECT_NAME: &str = "database.sqlite"; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] pub struct ForkSeedManifest { @@ -372,7 +373,7 @@ impl LtxRepl { use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; let key = ObjPath::from(self.fork_seed_key(cell, name)); - let stored = if name == "database.sqlite" { + let stored = if name == DATABASE_OBJECT_NAME { self.durability_codec .encode(key.as_ref(), &bytes) .map_err(|error| anyhow!("encrypt fork seed {cell}: {error}"))? @@ -391,7 +392,7 @@ impl LtxRepl { Ok(_) => Ok(()), Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { let existing = self.store.get(&key).await?.bytes().await?; - let existing = if name == "database.sqlite" { + let existing = if name == DATABASE_OBJECT_NAME { self.durability_codec .decode(key.as_ref(), &existing) .map_err(|error| anyhow!("decrypt existing fork seed {cell}: {error}"))? @@ -440,7 +441,7 @@ impl LtxRepl { sqlite_bytes: sqlite.len() as u64, }; let encoded_manifest = serde_json::to_vec(&manifest)?; - self.put_checkpoint_object(source_cell, checkpoint_id, "database.sqlite", sqlite) + self.put_checkpoint_object(source_cell, checkpoint_id, DATABASE_OBJECT_NAME, sqlite) .await?; self.put_checkpoint_object( source_cell, @@ -463,7 +464,7 @@ impl LtxRepl { use celld_ltx::object_store::{PutMode, PutOptions, PutPayload}; let key = ObjPath::from(self.checkpoint_key(cell, checkpoint, name)); - let stored = if name == "database.sqlite" { + let stored = if name == DATABASE_OBJECT_NAME { self.durability_codec .encode(key.as_ref(), &bytes) .map_err(|error| anyhow!("encrypt checkpoint {cell}/{checkpoint}: {error}"))? @@ -482,7 +483,7 @@ impl LtxRepl { Ok(_) => Ok(()), Err(celld_ltx::object_store::Error::AlreadyExists { .. }) => { let existing = self.store.get(&key).await?.bytes().await?; - let existing = if name == "database.sqlite" { + let existing = if name == DATABASE_OBJECT_NAME { self.durability_codec .decode(key.as_ref(), &existing) .map_err(|error| { @@ -523,7 +524,7 @@ impl LtxRepl { "checkpoint coordinates do not match its manifest" ); let database_key = - ObjPath::from(self.checkpoint_key(source_cell, checkpoint_id, "database.sqlite")); + ObjPath::from(self.checkpoint_key(source_cell, checkpoint_id, DATABASE_OBJECT_NAME)); let encoded = self.store.get(&database_key).await?.bytes().await?; let sqlite = self .durability_codec @@ -582,7 +583,7 @@ impl LtxRepl { if exact_retry { self.put_fork_seed_object(target_cell, "reserved.json", encoded_manifest.clone()) .await?; - self.put_fork_seed_object(target_cell, "database.sqlite", sqlite) + self.put_fork_seed_object(target_cell, DATABASE_OBJECT_NAME, sqlite) .await?; self.put_fork_seed_object(target_cell, "ready.json", encoded_manifest) .await?; @@ -598,7 +599,7 @@ impl LtxRepl { ); self.put_fork_seed_object(target_cell, "reserved.json", encoded_manifest.clone()) .await?; - self.put_fork_seed_object(target_cell, "database.sqlite", sqlite) + self.put_fork_seed_object(target_cell, DATABASE_OBJECT_NAME, sqlite) .await?; self.put_fork_seed_object(target_cell, "ready.json", encoded_manifest) .await?; @@ -626,7 +627,7 @@ impl LtxRepl { manifest.format == FORK_SEED_FORMAT, "unsupported fork seed format" ); - let database = ObjPath::from(self.fork_seed_key(cell, "database.sqlite")); + let database = ObjPath::from(self.fork_seed_key(cell, DATABASE_OBJECT_NAME)); let encoded = self.store.get(&database).await?.bytes().await?; let sqlite = self .durability_codec @@ -1669,7 +1670,7 @@ mod fork_seed_tests { let encoded = serde_json::to_vec(manifest).unwrap(); for (name, bytes) in [ ("reserved.json", encoded.clone()), - ("database.sqlite", sqlite), + (DATABASE_OBJECT_NAME, sqlite), ("ready.json", encoded), ] { store @@ -1821,9 +1822,9 @@ mod fork_seed_tests { let old_ltx = raw_objects(&store, "cells/source/ltx/e1/").await; assert!(!old_ltx.is_empty()); - assert!(old_ltx - .iter() - .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"old" })); + assert!(old_ltx.iter().all(|(_, bytes)| { + crate::durability_encryption::envelope_key_id_for_test(bytes).unwrap() == "old" + })); let checkpoint = store .get(&ObjPath::from( "cells/source/checkpoints/checkpoint-1/database.sqlite", @@ -1893,9 +1894,9 @@ mod fork_seed_tests { rotated.await_durable("source", 2, 1).await.unwrap(); let new_ltx = raw_objects(&store, "cells/source/ltx/e2/").await; assert!(!new_ltx.is_empty()); - assert!(new_ltx - .iter() - .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"new" })); + assert!(new_ltx.iter().all(|(_, bytes)| { + crate::durability_encryption::envelope_key_id_for_test(bytes).unwrap() == "new" + })); rotated .publish_fork_seed_from_checkpoint("source", "checkpoint-1", "fork", false) @@ -1913,7 +1914,10 @@ mod fork_seed_tests { .await .unwrap(); assert!(fork_object.starts_with(b"CRCELD01")); - assert_eq!(&fork_object[44..47], b"new"); + assert_eq!( + crate::durability_encryption::envelope_key_id_for_test(&fork_object).unwrap(), + "new" + ); let fork = rotated.activate(activation("fork", true)).await.unwrap(); assert!(fork.restored); @@ -1953,9 +1957,9 @@ mod fork_seed_tests { .unwrap(); let imported_ltx = raw_objects(&store, "cells/imported/ltx/e1/").await; assert!(!imported_ltx.is_empty()); - assert!(imported_ltx - .iter() - .all(|(_, bytes)| { bytes.starts_with(b"CRCELD01") && &bytes[44..47] == b"new" })); + assert!(imported_ltx.iter().all(|(_, bytes)| { + crate::durability_encryption::envelope_key_id_for_test(bytes).unwrap() == "new" + })); let (tampered_key, mut tampered_bytes) = new_ltx.into_iter().next().unwrap(); *tampered_bytes.last_mut().unwrap() ^= 1; diff --git a/crates/ltx/src/client/object_store.rs b/crates/ltx/src/client/object_store.rs index 693a455c6..afa7f822e 100644 --- a/crates/ltx/src/client/object_store.rs +++ b/crates/ltx/src/client/object_store.rs @@ -658,6 +658,7 @@ impl ReplicaClient for ObjectStoreClient { ); let key = ObjPath::from(self.ltx_key(level, min_txid, max_txid)); let encoded = self.codec.encode(key.as_ref(), data)?; + let encoded_size = encoded.len(); // Multipart threshold: < 5 MiB → single PUT; ≥ 5 MiB → multipart with // fixed-size parts. Ported from the Go uploader's 5 MiB PartSize default @@ -701,7 +702,7 @@ impl ReplicaClient for ObjectStoreClient { level, min_txid, max_txid, - size: data.len() as i64, + size: encoded_size as i64, created_at: Some(created_at), ..Default::default() }) @@ -798,6 +799,28 @@ mod tests { use super::*; use crate::replica_url::parse_replica_url_with_query; + #[derive(Debug)] + struct PrefixCodec; + + impl ReplicaObjectCodec for PrefixCodec { + fn name(&self) -> &'static str { + "test-prefix" + } + + fn encode(&self, _object_key: &str, plaintext: &[u8]) -> Result> { + let mut encoded = b"prefix".to_vec(); + encoded.extend_from_slice(plaintext); + Ok(encoded) + } + + fn decode(&self, _object_key: &str, encoded: &[u8]) -> Result> { + encoded + .strip_prefix(b"prefix") + .map(|bytes| bytes.to_vec()) + .ok_or_else(|| Error::Other("missing test prefix".into())) + } + } + #[test] fn timestamp_metadata_matches_go_rfc3339_nano() { assert_eq!( @@ -865,6 +888,47 @@ mod tests { assert_eq!(value.as_ref(), "2021-01-01T00:00:00.123Z"); } + #[tokio::test] + async fn encoded_size_matches_write_result_and_listing() { + let store = Arc::new(object_store::memory::InMemory::new()); + let client = ObjectStoreClient::with_store_and_codec( + ObjectStoreConfig { + bucket: "bucket".into(), + path: "replica".into(), + ..Default::default() + }, + store, + Arc::new(PrefixCodec), + ); + let data = ltx::Header { + version: ltx::VERSION, + flags: ltx::HEADER_FLAG_NO_CHECKSUM, + page_size: 512, + commit: 1, + min_txid: TXID(1), + max_txid: TXID(1), + timestamp: 1_609_459_200_123, + pre_apply_checksum: 0, + wal_offset: 0, + wal_size: 0, + wal_salt1: 0, + wal_salt2: 0, + node_id: 0, + } + .marshal(); + let written = client + .write_ltx_file(0, TXID(1), TXID(1), &data) + .await + .unwrap(); + let listed = client.ltx_files(0, TXID(0)).await.unwrap(); + assert_eq!(written.size, (data.len() + 6) as i64); + assert_eq!(listed[0].size, written.size); + assert_eq!( + client.open_ltx_file(0, TXID(1), TXID(1)).await.unwrap(), + data + ); + } + // ── ParseHost (port of TestParseHost, s3/replica_client_test.go:1071) ────── #[test] fn parse_host_table() { diff --git a/docs/security.md b/docs/security.md index 6380e050c..77c7e7dd2 100644 --- a/docs/security.md +++ b/docs/security.md @@ -112,6 +112,9 @@ active key and any retained read keys: } ``` +Key IDs may contain only ASCII letters, digits, periods, underscores, and +hyphens, and may be at most 64 characters long. + Celld generates a random 256-bit data key for every LTX body and SQLite checkpoint/fork image, encrypts the database bytes with AES-256-GCM, and wraps that data key with the active versioned AES-256-GCM key-encryption key. Both