From c36b09876173a2872f91dc0e476293612b33f29a Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 11:10:40 +0000 Subject: [PATCH 1/4] feat(TC-887): seal KV blocks in the S3 store With encryption at rest on, the S3 block store writes new blocks as TCE1 objects under their sealed names (v1//) and reads them back through the authenticated decrypting reader. Range reads fetch the header, then only the segments that cover the range. Legacy objects stay readable in encrypt mode and are refused in strict mode. While keys are locked, writes and sealed reads fail with the at-rest 503. Quotas count plaintext bytes, sealed objects by space tag. Removing a block removes its sealed copy, or else the legacy object. storage::attach_at_rest attaches the runtime to both stores. The object size check uses HeadObject: GetObjectAttributes without an attribute list is rejected by S3 and MinIO. tests/at_rest_s3_store.rs runs against the at-rest harness MinIO (ignored by default; the harness CI job runs it). --- .github/workflows/rust.yml | 5 + tinycloud-node-server/src/storage/mod.rs | 18 +- tinycloud-node-server/src/storage/s3.rs | 657 ++++++++++++------ .../tests/at_rest_s3_store.rs | 273 ++++++++ 4 files changed, 742 insertions(+), 211 deletions(-) create mode 100644 tinycloud-node-server/tests/at_rest_s3_store.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 7d67f8f4..33de19f6 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -355,6 +355,11 @@ jobs: (set -a; . "$TC887_RUN_DIR/harness.env"; set +a cargo test --locked -p tinycloud-node --test s3_block_store -- --ignored) + - name: S3 block store with encryption at rest (against the harness MinIO) + run: | + (set -a; . "$TC887_RUN_DIR/harness.env"; set +a + cargo test --locked -p tinycloud-node --test at_rest_s3_store -- --ignored) + - name: Harness logs if: failure() run: | diff --git a/tinycloud-node-server/src/storage/mod.rs b/tinycloud-node-server/src/storage/mod.rs index 853e4d62..ccad5473 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -9,18 +9,14 @@ use std::sync::Arc; use crate::{at_rest::AtRestRuntime, BlockStores}; use tinycloud_core::storage::either::Either; -/// Links the opened block store to the node's at-rest state (TC-887). -/// -/// The FileSystem store seals KV blocks when the mode and keys allow, so it -/// declares a seal path. The S3 store does not seal yet and declares none, -/// so `at_rest::setup` refuses a non-`off` mode with S3 storage rather than -/// write plaintext. +/// Links the opened block store to the node's at-rest state (TC-887). Both +/// stores seal KV blocks when the mode and keys allow, so storage declares a +/// seal path. pub fn attach_at_rest(stores: &mut BlockStores, runtime: Arc) { + runtime.declare_seal_path(); + let handle = sealed::AtRestHandle::new(runtime); match stores { - Either::B(fs) => { - runtime.declare_seal_path(); - fs.attach_at_rest(sealed::AtRestHandle::new(runtime)); - } - Either::A(_) => {} + Either::A(s3) => s3.attach_at_rest(handle), + Either::B(fs) => fs.attach_at_rest(handle), } } diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 5935a343..27dbaad9 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -8,18 +8,29 @@ use aws_smithy_http::{byte_stream::Error as ByteStreamError, endpoint::Endpoint} use aws_types::sdk_config::SdkConfig; use futures::{ future::Either as AsyncEither, + io::AsyncReadExt, stream::{IntoAsyncRead, MapErr, TryStreamExt}, }; use rocket::{async_trait, http::hyper::Uri}; use serde::{Deserialize, Serialize}; use serde_with::{serde_as, DisplayFromStr}; -use std::{collections::HashMap, io::Error as IoError, ops::AddAssign, time::Instant}; +use std::{ + collections::HashMap, future::Future, io::Error as IoError, ops::AddAssign, time::Instant, +}; use tinycloud_auth::resource::SpaceId; -use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*}; +use tinycloud_core::{ + at_rest::{plaintext_len, DataKeyProvider, DecryptReader, HEADER_LEN}, + hash::Hash, + storage::*, +}; use super::{ file_system, - sealed::{self, scanned_object, ScannedObject}, + sealed::{ + self, open_sealed, plan_range, read_plan, scanned_object, seal_file, seal_vec, + space_tag_of, write_target, AtRestHandle, AtRestStoreError, AtRestView, ReadPlan, + ScannedObject, SealFileError, SealedName, WriteTarget, + }, size::SpaceSizes, }; @@ -32,6 +43,38 @@ pub struct S3BlockStore { pub client: Client, pub bucket: String, sizes: SpaceSizes, + at_rest: Option, +} + +/// An S3 object body as an `AsyncRead`. +type Body = IntoAsyncRead IoError>>; + +/// Plaintext from a legacy object, or authenticated plaintext from a sealed one. +/// The decrypting side is boxed: it carries the AES key schedule and segment +/// buffers, and `Content` values move by value through deep request paths. +pub type S3Readable = AsyncEither>>; + +fn body(stream: ByteStream) -> Body { + stream + .map_err(convert as fn(ByteStreamError) -> IoError) + .into_async_read() +} + +/// A staged block on its way into the store. +enum Staged { + File(file_system::TempFileStage), + Bytes(Vec), +} + +/// The outcome of a ranged GET. +enum RangeGet { + Missing, + InvalidRange, + Got { + range: ResolvedByteRange, + total: u64, + body: Body, + }, } #[serde_as] @@ -120,6 +163,7 @@ impl S3BlockStore { client, bucket: config.bucket.clone(), sizes, + at_rest: None, }) } @@ -131,11 +175,336 @@ impl S3BlockStore { ) } - async fn increment_size(&self, space: &SpaceId, size: u64) { - self.sizes.increment_size(space, None, size).await; + /// Links the store to the node's at-rest state (TC-887). Without it the + /// store behaves as at-rest mode `off`. + pub fn attach_at_rest(&mut self, handle: AtRestHandle) { + self.at_rest = Some(handle); } - async fn decrement_size(&self, space: &SpaceId, size: u64) { - self.sizes.decrement_size(space, None, size).await; + + fn view(&self) -> AtRestView { + AtRestView::of(self.at_rest.as_ref()) + } + + async fn head_exists(&self, key: &str) -> Result { + Ok(self.object_size(key).await?.is_some()) + } + + /// The whole object: (stored length, body), or `None` if absent. + async fn get_full(&self, key: &str) -> Result, S3StoreError> { + match self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + { + Ok(o) => Ok(Some((o.content_length().try_into()?, body(o.body)))), + Err(SdkError::ServiceError { + err: + GetObjectError { + kind: GetObjectErrorKind::NoSuchKey(_), + .. + }, + .. + }) => Ok(None), + Err(e) => Err(S3Error::from(e).into()), + } + } + + async fn get_range(&self, key: &str, range: ByteRangeSpec) -> Result { + match self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .range(range.to_http_value()) + .send() + .await + { + Ok(output) => { + let content_range = output.content_range().ok_or_else(|| { + S3StoreError::InvalidContentRange("missing header".to_string()) + })?; + let (range, total) = parse_content_range(content_range)?; + Ok(RangeGet::Got { + range, + total, + body: body(output.body), + }) + } + Err(SdkError::ServiceError { + err: + GetObjectError { + kind: GetObjectErrorKind::NoSuchKey(_), + .. + }, + .. + }) => Ok(RangeGet::Missing), + Err(SdkError::ServiceError { err, .. }) if err.code() == Some("InvalidRange") => { + Ok(RangeGet::InvalidRange) + } + Err(error) => Err(S3Error::from(error).into()), + } + } + + async fn put(&self, key: &str, body: ByteStream) -> Result<(), S3StoreError> { + self.client + .put_object() + .bucket(&self.bucket) + .key(key) + .body(body) + .send() + .await + .map_err(S3Error::from)?; + Ok(()) + } + + async fn persist_staged( + &self, + space: &SpaceId, + hash: &Hash, + staged: Staged, + ) -> Result<(), S3StoreError> { + match write_target(&self.view(), space, hash).await? { + WriteTarget::Legacy => { + let key = self.key(space, hash); + if self.head_exists(&key).await? { + return Ok(()); + } + let size = match staged { + Staged::File(t_file) => { + let size = t_file.size().await?; + let (_file, path) = t_file.into_inner(); + self.put(&key, ByteStream::from_path(&path).await?).await?; + size + } + Staged::Bytes(b) => { + let size = b.len() as u64; + self.put(&key, ByteStream::from(b)).await?; + size + } + }; + self.sizes.increment_size(space, None, size).await; + } + WriteTarget::Sealed { name, key } => { + // Only the sealed name dedups: a legacy copy never suppresses + // the sealed write. + let object = name.key(); + if self.head_exists(&object).await? { + return Ok(()); + } + let size = match staged { + Staged::File(t_file) => { + let size = t_file.size().await?; + let (_file, path) = t_file.into_inner(); + let sealed = seal_file(&key, space, hash, &path, size).await?; + self.put(&object, ByteStream::from_path(sealed.path()).await?) + .await?; + size + } + Staged::Bytes(b) => { + let size = b.len() as u64; + let ct = seal_vec(key, space, hash, b).await?; + self.put(&object, ByteStream::from(ct)).await?; + size + } + }; + self.sizes + .increment_size(space, Some(&name.space_tag), size) + .await; + } + } + Ok(()) + } + + async fn read_legacy( + &self, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + Ok(self + .get_full(&self.key(space, id)) + .await? + .map(|(len, b)| Content::new(len, AsyncEither::Left(b)))) + } + + async fn read_sealed( + &self, + name: &SealedName, + provider: &dyn DataKeyProvider, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + let Some((len, b)) = self.get_full(&name.key()).await? else { + return Ok(None); + }; + let (size, reader) = open_sealed(provider, space, id, b, len).await?.into_inner(); + Ok(Some(Content::new( + size, + AsyncEither::Right(Box::new(reader)), + ))) + } + + async fn read_block( + &self, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id)? { + ReadPlan::Legacy => self.read_legacy(space, id).await, + ReadPlan::LegacyWhileLocked => match self.read_legacy(space, id).await? { + Some(c) => Ok(Some(c)), + None => Err(AtRestStoreError::Unavailable("at-rest keys are locked").into()), + }, + ReadPlan::SealedThenLegacy { name, provider } => { + match self.read_sealed(&name, &*provider, space, id).await? { + Some(c) => Ok(Some(c)), + None => self.read_legacy(space, id).await, + } + } + ReadPlan::Sealed { name, provider } => { + match self.read_sealed(&name, &*provider, space, id).await? { + Some(c) => Ok(Some(c)), + None if self.head_exists(&self.key(space, id)).await? => { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } + } + } + + async fn range_legacy( + &self, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + let key = self.key(space, id); + match self.get_range(&key, spec).await? { + RangeGet::Got { range, total, body } => Ok(Some(RangeRead::Content { + total_size: total, + range, + content: Content::new(range.len(), AsyncEither::Left(body)), + })), + RangeGet::Missing => Ok(None), + RangeGet::InvalidRange => { + let head = self + .client + .head_object() + .bucket(&self.bucket) + .key(&key) + .send() + .await; + match head { + Ok(output) => Ok(Some(RangeRead::Unsatisfiable { + total_size: output.content_length().try_into()?, + })), + Err(SdkError::ServiceError { + err: + HeadObjectError { + kind: HeadObjectErrorKind::NotFound(_), + .. + }, + .. + }) => Ok(None), + Err(error) => Err(S3Error::from(error).into()), + } + } + } + } + + /// A range of the block's sealed object, or `None` if it is absent: one + /// GET for the header, one for the segments. Totals and the 416 decision + /// use the plaintext length. + async fn range_sealed( + &self, + name: &SealedName, + provider: &dyn DataKeyProvider, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + let object = name.key(); + let header_span = ByteRangeSpec::Inclusive { + start: 0, + end: HEADER_LEN as u64 - 1, + }; + let (total, mut head_body) = match self.get_range(&object, header_span).await? { + RangeGet::Missing => return Ok(None), + RangeGet::InvalidRange => { + return Err(AtRestStoreError::Integrity("object shorter than its header").into()) + } + RangeGet::Got { total, body, .. } => (total, body), + }; + let mut header = [0u8; HEADER_LEN]; + head_body + .read_exact(&mut header) + .await + .map_err(|_| AtRestStoreError::Integrity("object shorter than its header"))?; + let plan = plan_range(provider, space, &header, total, spec).await?; + let Some(r) = plan.range else { + return Ok(Some(RangeRead::Unsatisfiable { + total_size: plan.plaintext_len(), + })); + }; + let span = ByteRangeSpec::Inclusive { + start: r.ciphertext_start(), + end: r.ciphertext_end(), + }; + let body = match self.get_range(&object, span).await? { + RangeGet::Got { range, body, .. } + if range.start() == r.ciphertext_start() && range.end() == r.ciphertext_end() => + { + body + } + RangeGet::Missing => return Ok(None), + _ => return Err(AtRestStoreError::Integrity("object changed during a read").into()), + }; + let reader = plan.reader(space, id, r, body)?; + let range = r.plaintext_range(); + Ok(Some(RangeRead::Content { + total_size: plan.plaintext_len(), + range, + content: Content::new(range.len(), AsyncEither::Right(Box::new(reader))), + })) + } + + async fn range_block( + &self, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id)? { + ReadPlan::Legacy => self.range_legacy(space, id, spec).await, + ReadPlan::LegacyWhileLocked => match self.range_legacy(space, id, spec).await? { + Some(r) => Ok(Some(r)), + None => Err(AtRestStoreError::Unavailable("at-rest keys are locked").into()), + }, + ReadPlan::SealedThenLegacy { name, provider } => { + match self + .range_sealed(&name, &*provider, space, id, spec) + .await? + { + Some(r) => Ok(Some(r)), + None => self.range_legacy(space, id, spec).await, + } + } + ReadPlan::Sealed { name, provider } => { + match self + .range_sealed(&name, &*provider, space, id, spec) + .await? + { + Some(r) => Ok(Some(r)), + None if self.head_exists(&self.key(space, id)).await? => { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } + } } /// The stored size of an object, or `None` if absent. HeadObject, not @@ -195,29 +564,27 @@ pub enum S3StoreError { AtRest(#[from] sealed::AtRestStoreError), } +impl From for S3StoreError { + fn from(e: SealFileError) -> Self { + match e { + SealFileError::AtRest(e) => Self::AtRest(e), + SealFileError::Io(e) => Self::Io(e), + } + } +} + #[async_trait] impl ImmutableReadStore for S3BlockStore { type Error = S3StoreError; - type Readable = IntoAsyncRead IoError>>; + type Readable = S3Readable; async fn contains(&self, space: &SpaceId, id: &Hash) -> Result { - match self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await - { - Ok(_) => Ok(true), - Err(SdkError::ServiceError { - err: - HeadObjectError { - kind: HeadObjectErrorKind::NotFound(_), - .. - }, - .. - }) => Ok(false), - Err(e) => Err(S3Error::from(e).into()), + match read_plan(&self.view(), space, id)? { + ReadPlan::Legacy | ReadPlan::LegacyWhileLocked => { + self.head_exists(&self.key(space, id)).await + } + ReadPlan::SealedThenLegacy { name, .. } => Ok(self.head_exists(&name.key()).await? + || self.head_exists(&self.key(space, id)).await?), + ReadPlan::Sealed { name, .. } => self.head_exists(&name.key()).await, } } @@ -227,30 +594,9 @@ impl ImmutableReadStore for S3BlockStore { id: &Hash, ) -> Result>, Self::Error> { let start = Instant::now(); - let res = self - .client - .get_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await; - let result = match res { - Ok(o) => Ok(Some(Content::new( - o.content_length().try_into()?, - o.body - .map_err(convert as fn(ByteStreamError) -> IoError) - .into_async_read(), - ))), - Err(SdkError::ServiceError { - err: - GetObjectError { - kind: GetObjectErrorKind::NoSuchKey(_), - .. - }, - .. - }) => Ok(None), - Err(e) => Err(S3Error::from(e).into()), - }; + // Boxed: the sealed read path's future is large, and the trait method + // future is built on the caller's stack before async_trait boxes it. + let result = Box::pin(self.read_block(space, id)).await; crate::prometheus::observe_stage( crate::prometheus::InvocationStage::BlockRead, crate::prometheus::StageOutcome::from(result.is_ok()), @@ -265,67 +611,7 @@ impl ImmutableReadStore for S3BlockStore { id: &Hash, range: ByteRangeSpec, ) -> Result>, Self::Error> { - let key = self.key(space, id); - let output = self - .client - .get_object() - .bucket(&self.bucket) - .key(key) - .range(range.to_http_value()) - .send() - .await; - match output { - Ok(output) => { - let content_range = output.content_range().ok_or_else(|| { - S3StoreError::InvalidContentRange("missing header".to_string()) - })?; - let (range, total_size) = parse_content_range(content_range)?; - let content = Content::new( - range.len(), - output - .body - .map_err(convert as fn(ByteStreamError) -> IoError) - .into_async_read(), - ); - Ok(Some(RangeRead::Content { - total_size, - range, - content, - })) - } - Err(SdkError::ServiceError { - err: - GetObjectError { - kind: GetObjectErrorKind::NoSuchKey(_), - .. - }, - .. - }) => return Ok(None), - Err(SdkError::ServiceError { err, .. }) if err.code() == Some("InvalidRange") => { - let head = self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await; - match head { - Ok(output) => Ok(Some(RangeRead::Unsatisfiable { - total_size: output.content_length().try_into()?, - })), - Err(SdkError::ServiceError { - err: - HeadObjectError { - kind: HeadObjectErrorKind::NotFound(_), - .. - }, - .. - }) => Ok(None), - Err(error) => Err(S3Error::from(error).into()), - } - } - Err(error) => Err(S3Error::from(error).into()), - } + Box::pin(self.range_block(space, id, range)).await } } @@ -355,79 +641,60 @@ fn parse_content_range(value: &str) -> Result<(ResolvedByteRange, u64), S3StoreE Ok((range, total_size)) } +/// Times a block write, as before. +async fn observed_write(write: F) -> Result +where + F: Future>, +{ + let start = Instant::now(); + let result = write.await; + crate::prometheus::observe_stage( + crate::prometheus::InvocationStage::BlockWrite, + crate::prometheus::StageOutcome::from(result.is_ok()), + start.elapsed(), + ); + result +} + #[async_trait] impl ImmutableWriteStore for S3BlockStore { type Error = S3StoreError; + async fn prepare_write(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer<::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - let size = f.len() as u64; - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from(f)) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } + Box::pin(self.persist_staged(space, &hash, Staged::Bytes(f))).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } #[async_trait] impl ImmutableWriteStore for S3BlockStore { type Error = S3StoreError; + async fn prepare_write(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer<::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - let size = f.size().await?; - let (_file, path) = f.into_inner(); - - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from_path(&path).await?) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } + Box::pin(self.persist_staged(space, &hash, Staged::File(f))).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } @@ -436,54 +703,25 @@ impl ImmutableWriteStore Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer< as ImmutableStaging>::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - match f { - AsyncEither::Left(t_file) => { - let size = t_file.size().await?; - let (_file, path) = t_file.into_inner(); - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from_path(&path).await?) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } - AsyncEither::Right(b) => { - let size = b.len() as u64; - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from(b)) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } - } + let staged = match f { + AsyncEither::Left(t_file) => Staged::File(t_file), + AsyncEither::Right(b) => Staged::Bytes(b), }; + Box::pin(self.persist_staged(space, &hash, staged)).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } @@ -491,12 +729,27 @@ impl ImmutableWriteStore Result, Self::Error> { + // A sealed copy first (if keys can name it), then the legacy object. + if let Some(nk) = self.view().object_name_key() { + let name = SealedName::new(&nk, space, id); + if let Some(size) = self.object_size(&name.key()).await? { + self.delete(&name.key()).await?; + self.sizes + .decrement_size( + space, + Some(&name.space_tag), + plaintext_len(size).unwrap_or(0), + ) + .await; + return Ok(Some(())); + } + } let key = self.key(space, id); let Some(size) = self.object_size(&key).await? else { return Ok(None); }; self.delete(&key).await?; - self.decrement_size(space, size).await; + self.sizes.decrement_size(space, None, size).await; Ok(Some(())) } } @@ -505,7 +758,11 @@ impl ImmutableDeleteStore for S3BlockStore { impl StoreSize for S3BlockStore { type Error = S3StoreError; async fn total_size(&self, space: &SpaceId) -> Result, Self::Error> { - Ok(self.sizes.get_size(space, None).await) + let view = self.view(); + Ok(self + .sizes + .get_size(space, space_tag_of(&view, space).as_deref()) + .await) } } diff --git a/tinycloud-node-server/tests/at_rest_s3_store.rs b/tinycloud-node-server/tests/at_rest_s3_store.rs new file mode 100644 index 00000000..3e61c985 --- /dev/null +++ b/tinycloud-node-server/tests/at_rest_s3_store.rs @@ -0,0 +1,273 @@ +//! TC-887 S3c: the S3 block store with encryption at rest, against a real +//! S3 API (the native MinIO of `scripts/at-rest-harness`). Ignored by +//! default; run with the harness environment loaded: +//! +//! ```sh +//! (set -a; . "$RUN_DIR/harness.env"; set +a; +//! cargo test -p tinycloud-node --test at_rest_s3_store -- --ignored) +//! ``` + +use std::sync::Arc; + +use anyhow::{ensure, Context, Result}; +use futures::io::{AsyncReadExt, AsyncWriteExt}; +use parking_lot::RwLock; +use tinycloud::{ + config::AtRestMode, + storage::{ + s3::{S3BlockConfig, S3BlockStore, S3StoreError}, + sealed::{AtRestHandle, AtRestSource, AtRestStoreError, AtRestView}, + }, +}; +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::{ + at_rest::{ciphertext_len, DataKeyProvider, FixedKeyProvider, SEGMENT_LEN}, + hash::Hash, + storage::{ + memory::MemoryStaging, ByteRangeSpec, HashBuffer, ImmutableDeleteStore, ImmutableReadStore, + ImmutableWriteStore, RangeRead, StorageConfig, StorageSetup, StoreSize, + }, +}; + +const SEG: usize = SEGMENT_LEN; + +struct Switch(RwLock); + +impl AtRestSource for Switch { + fn view(&self) -> AtRestView { + self.0.read().clone() + } +} + +fn view(mode: AtRestMode, encrypting: bool, keys: Option) -> AtRestView { + AtRestView { + mode, + encrypting, + provider: keys.map(|k| Arc::new(k) as Arc), + } +} + +fn keys() -> FixedKeyProvider { + FixedKeyProvider::new([31; 32]) +} + +fn env(name: &str) -> Result { + std::env::var(name).with_context(|| format!("{name} (load the harness env)")) +} + +async fn open_store(v: AtRestView) -> Result<(S3BlockStore, Arc)> { + // The S3 client reads credentials from the AWS_* environment. + std::env::set_var("AWS_ACCESS_KEY_ID", env("TC887_S3_ACCESS_KEY_ID")?); + std::env::set_var("AWS_SECRET_ACCESS_KEY", env("TC887_S3_SECRET_ACCESS_KEY")?); + std::env::set_var("AWS_REGION", env("TC887_S3_REGION")?); + std::env::set_var("AWS_EC2_METADATA_DISABLED", "true"); + let config: S3BlockConfig = serde_json::from_value(serde_json::json!({ + "bucket": env("TC887_S3_BUCKET")?, + "endpoint": env("TC887_S3_ENDPOINT")?, + }))?; + let mut store = config.open().await?; + let switch = Arc::new(Switch(RwLock::new(v))); + store.attach_at_rest(AtRestHandle::new(switch.clone())); + Ok((store, switch)) +} + +async fn store(v: AtRestView) -> Result<(S3BlockStore, Arc, SpaceId)> { + let (store, switch) = open_store(v).await?; + let space: SpaceId = format!( + "tinycloud:key:s3store:{}", + hex::encode(rand::random::<[u8; 6]>()) + ) + .parse()?; + store.create(&space).await?; + Ok((store, switch, space)) +} + +fn value(n: usize) -> Vec { + (0..n).map(|i| (i * 13 % 251) as u8).collect() +} + +async fn put(store: &S3BlockStore, space: &SpaceId, data: &[u8]) -> Result { + ImmutableWriteStore::::prepare_write(store, space).await?; + let mut staged = HashBuffer::new(Vec::new()); + staged.write_all(data).await.unwrap(); + ImmutableWriteStore::::persist(store, space, staged).await +} + +async fn get(store: &S3BlockStore, space: &SpaceId, h: &Hash) -> Result> { + let content = store.read(space, h).await?.context("block missing")?; + let len = content.len(); + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + ensure!( + out.len() as u64 == len, + "Content::len is not the plaintext length" + ); + Ok(out) +} + +async fn raw(store: &S3BlockStore, key: &str) -> Result> { + Ok(store + .client + .get_object() + .bucket(&store.bucket) + .key(key) + .send() + .await? + .body + .collect() + .await? + .into_bytes() + .to_vec()) +} + +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn sealed_blocks_round_trip_and_range_over_s3() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + let nk = keys().object_name_key()?; + let mut total = 0u64; + for n in [0, 17, SEG, SEG + 1, 3 * SEG + 9] { + let data = value(n); + let h = put(&store, &space, &data).await?; + total += n as u64; + // Stored under the sealed name only, as TCE1. + let sealed = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + let bytes = raw(&store, &sealed).await?; + ensure!( + bytes.len() as u64 == ciphertext_len(n as u64)?, + "{n}: stored size" + ); + ensure!(bytes.starts_with(b"TCE1"), "{n}: not TCE1"); + ensure!(get(&store, &space, &h).await? == data, "{n}: read back"); + if n > 0 { + let edges = [0, 1, SEG.saturating_sub(1), SEG, SEG + 1, n - 1]; + for &a in edges.iter().filter(|&&a| a < n) { + for &b in edges.iter().filter(|&&b| b >= a && b < n) { + let spec = ByteRangeSpec::Inclusive { + start: a as u64, + end: b as u64, + }; + match store.read_range(&space, &h, spec).await?.context("range")? { + RangeRead::Content { + total_size, + range, + content, + } => { + ensure!(total_size == n as u64, "{n}: range total"); + ensure!((range.start(), range.end()) == (a as u64, b as u64)); + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + ensure!(out == data[a..=b], "{n}: range {a}..={b}"); + } + RangeRead::Unsatisfiable { .. } => anyhow::bail!("{n}: {a}..={b} 416"), + } + } + } + } + match store + .read_range(&space, &h, ByteRangeSpec::From { start: n as u64 }) + .await? + .context("range")? + { + RangeRead::Unsatisfiable { total_size } => { + ensure!( + total_size == n as u64, + "{n}: 416 total is the plaintext length" + ) + } + RangeRead::Content { .. } => anyhow::bail!("{n}: range past the end was served"), + } + } + ensure!( + store.total_size(&space).await? == Some(total), + "plaintext quota" + ); + // A fresh store over the same bucket counts the sealed objects' plaintext + // bytes at boot, by space tag, with no legacy objects for the space. + let (reopened, _) = open_store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + ensure!( + reopened.total_size(&space).await? == Some(total), + "boot-scan plaintext quota" + ); + Ok(()) +} + +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn legacy_locked_strict_and_tamper_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let legacy = value(5000); + let hl = put(&store, &space, &legacy).await?; + + // encrypt: legacy still readable; a new block is sealed. + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + ensure!( + get(&store, &space, &hl).await? == legacy, + "legacy in encrypt" + ); + let sealed = value(SEG + 7); + let hs = put(&store, &space, &sealed).await?; + + // Locked: writes and sealed reads are 503; legacy is served. + *switch.0.write() = view(AtRestMode::Encrypt, false, Some(keys().locked())); + let unavailable = |e: &S3StoreError| matches!(e, S3StoreError::AtRest(e) if e.status() == rocket::http::Status::ServiceUnavailable); + ensure!( + put(&store, &space, &value(9)) + .await + .as_ref() + .is_err_and(unavailable), + "locked write" + ); + ensure!( + store + .read(&space, &hs) + .await + .as_ref() + .is_err_and(unavailable), + "locked sealed read" + ); + ensure!( + get(&store, &space, &hl).await? == legacy, + "legacy while locked" + ); + + // strict: legacy refused, sealed served. + *switch.0.write() = view(AtRestMode::Strict, true, Some(keys())); + ensure!( + matches!( + store.read(&space, &hl).await, + Err(S3StoreError::AtRest(AtRestStoreError::LegacyRefused)) + ), + "strict refuses legacy" + ); + ensure!( + get(&store, &space, &hs).await? == sealed, + "strict reads sealed" + ); + + // A tampered sealed object fails closed. + let nk = keys().object_name_key()?; + let key = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &hs); + let mut bytes = raw(&store, &key).await?; + bytes[64 + 3] ^= 1; + store + .client + .put_object() + .bucket(&store.bucket) + .key(&key) + .body(bytes.into()) + .send() + .await?; + let content = store.read(&space, &hs).await?.context("block")?; + let mut out = Vec::new(); + let err = Box::pin(content).read_to_end(&mut out).await.unwrap_err(); + ensure!( + err.kind() == std::io::ErrorKind::InvalidData && out.is_empty(), + "tamper" + ); + + // Removing the sealed block drops it. + ensure!(store.remove(&space, &hs).await? == Some(()), "remove"); + ensure!(store.read(&space, &hs).await?.is_none(), "removed"); + Ok(()) +} From f87eb1c6da5545f1d5398b9acd4dd4a5e658e8be Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 12:29:32 +0000 Subject: [PATCH 2/4] test(TC-887): a double remove floors plaintext sizes at zero; escrow doc for the S3 seal path --- docs/at-rest-escrow.md | 2 +- tinycloud-node-server/src/storage/size.rs | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/docs/at-rest-escrow.md b/docs/at-rest-escrow.md index a083ef3c..71da932a 100644 --- a/docs/at-rest-escrow.md +++ b/docs/at-rest-escrow.md @@ -146,7 +146,7 @@ The same settings as compose env: `TINYCLOUD_AT_REST__MODE`, `TINYCLOUD_AT_REST_ - it keeps a copy of the sealed key, the bundle, the ceremony record and each space's wrapped data key in the block store (`_tc/at-rest/` in S3, `{datadir}/at-rest/_tc/at-rest/` on disk). Data-key objects are named by a keyed hash, not the space; - in this mode a copy that can't be written is logged and the boot goes on (nothing depends on it yet). With `encrypt` or `strict`, every boot-time copy must be written: a block store that refuses the write stops the boot; - nothing is encrypted. -- **`encrypt` and `strict` need a block store with a seal path.** The FileSystem store seals KV blocks (TC-887 S3); with S3 block storage, `encrypt` and `strict` are refused at boot in this build, so the node won't claim to encrypt while it writes plaintext. With a seal path, `encrypt` or `strict` refuses to start unless the recipients equal the bundle's, `verified_fp` equals its EFP, and the stored ceremony record verifies under the key inside the TEE. A Static-root node with no recipients may use `escrow_none = true` instead, which acknowledges that no escrow exists. +- **`encrypt` and `strict` need a block store with a seal path.** Both the FileSystem and the S3 store seal KV blocks (TC-887 S3), so both declare one; without it `encrypt` and `strict` are refused at boot, so the node never claims to encrypt while it writes plaintext. With a seal path, `encrypt` or `strict` refuses to start unless the recipients equal the bundle's, `verified_fp` equals its EFP, and the stored ceremony record verifies under the key inside the TEE. A Static-root node with no recipients may use `escrow_none = true` instead, which acknowledges that no escrow exists. - **In a TEE, the escrow recipients and `verified_fp` must be literal values in the measured compose.** With any at-rest config, the node reads its `app-compose.json` from the guest agent (`/Info`, `tcb_info.app_compose`), checks it against `compose_hash`, and refuses to boot unless `TINYCLOUD_AT_REST__ESCROW__RECIPIENTS` and `TINYCLOUD_AT_REST__ESCROW__VERIFIED_FP` are set there as literals (no `${…}`, no pass-through) equal to the effective config. Values from the encrypted environment, a file or anywhere else are refused. If the app-compose can't be read or parsed, the node refuses to boot. The `/Info` shape must be confirmed on a live CVM before the escrow ceremony. - **The escrow bundle is made or changed only at boot, never by a request.** Every endpoint below only reads it. - **An ARK never gets a second bundle for other holders.** The node refuses to boot (`escrow_holders_changed`) if the configured recipients differ from any existing bundle for its ARK: the active bundle, any stored escrow row, a ceremony record in the block store whose MAC verifies under the ARK, or a bundle object in the block store naming the ARK. This holds after a restore too, when the database lost its escrow rows. The block-store bundle object is unauthenticated, so it can only refuse, never be used. Changing a holder after the ceremony needs an ARK rotation (TC-1345). diff --git a/tinycloud-node-server/src/storage/size.rs b/tinycloud-node-server/src/storage/size.rs index 740bb0cc..81f85ace 100644 --- a/tinycloud-node-server/src/storage/size.rs +++ b/tinycloud-node-server/src/storage/size.rs @@ -123,5 +123,10 @@ mod tests { sizes.decrement_size(&c, Some("tagC"), 40).await; assert_eq!(sizes.get_size(&c, Some("tagC")).await, Some(40)); assert_eq!(sizes.get_size(&c, None).await, Some(40)); + // A second removal of the same bytes (double remove) floors at zero + // under both entries; it never wraps. + sizes.decrement_size(&c, Some("tagC"), 40).await; + sizes.decrement_size(&c, None, 40).await; + assert_eq!(sizes.get_size(&c, Some("tagC")).await, Some(0)); } } From 3268564f6589ba6d84b74af8f9ee53f5b334f28f Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 14:12:02 +0000 Subject: [PATCH 3/4] =?UTF-8?q?fix(TC-887):=20S3=20range=20reads=20pinned?= =?UTF-8?q?=20by=20ETag,=20no=20legacy=20fallback=20mid-read;=20contains?= =?UTF-8?q?=20per=20=C2=A77;=20prepare=20hooks;=20more=20S3=20tests=20(S3?= =?UTF-8?q?=20review)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/at-rest/INTERFACES.md | 4 +- tinycloud-node-server/src/storage/s3.rs | 152 +++++++++----- .../tests/at_rest_s3_store.rs | 194 +++++++++++++++++- 3 files changed, 293 insertions(+), 57 deletions(-) diff --git a/docs/at-rest/INTERFACES.md b/docs/at-rest/INTERFACES.md index d7a4e9ba..0abe38d6 100644 --- a/docs/at-rest/INTERFACES.md +++ b/docs/at-rest/INTERFACES.md @@ -299,7 +299,9 @@ redacted. - **Accepted residual (header trust):** `Content-Length`/`Content-Range`/416 totals come from the unverified header. A forged header can only produce a wrong total or a body that aborts with no unauthenticated bytes (test `storage::sealed::tests::forged_header_length_aborts_without_extra_bytes`), - never wrong bytes. + never wrong bytes. On S3 a range read's segment GET is pinned to the header GET's ETag + (`If-Match`): if the object is replaced or removed in between, the read starts over once, then + answers 503. It never falls back to the legacy name once a sealed header was read. - **Quota:** plaintext bytes. Sealed objects are scanned by `space_tag` and joined to a space when the object-name key is unlocked (lazily, per lookup), so the first quota check after an unlock counts legacy + sealed. While locked, a space reports legacy bytes only (writes are 503 then). diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 27dbaad9..a7100b52 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -19,7 +19,7 @@ use std::{ }; use tinycloud_auth::resource::SpaceId; use tinycloud_core::{ - at_rest::{plaintext_len, DataKeyProvider, DecryptReader, HEADER_LEN}, + at_rest::{plaintext_len, DataKeyProvider, HEADER_LEN}, hash::Hash, storage::*, }; @@ -27,9 +27,10 @@ use tinycloud_core::{ use super::{ file_system, sealed::{ - self, open_sealed, plan_range, read_plan, scanned_object, seal_file, seal_vec, - space_tag_of, write_target, AtRestHandle, AtRestStoreError, AtRestView, ReadPlan, - ScannedObject, SealFileError, SealedName, WriteTarget, + self, contains_name, open_sealed, plan_range, read_plan, scanned_object, seal_file, + seal_vec, space_tag_of, write_target, AtRestHandle, AtRestStoreError, AtRestView, + ContainsName, LoggedDecrypt, ReadPlan, ScannedObject, SealFileError, SealedName, + WriteTarget, }, size::SpaceSizes, }; @@ -52,7 +53,7 @@ type Body = IntoAsyncRead IoError>>; /// Plaintext from a legacy object, or authenticated plaintext from a sealed one. /// The decrypting side is boxed: it carries the AES key schedule and segment /// buffers, and `Content` values move by value through deep request paths. -pub type S3Readable = AsyncEither>>; +pub type S3Readable = AsyncEither>>; fn body(stream: ByteStream) -> Body { stream @@ -70,9 +71,12 @@ enum Staged { enum RangeGet { Missing, InvalidRange, + /// `If-Match` failed: the object was replaced since the ETag was read. + Changed, Got { range: ResolvedByteRange, total: u64, + etag: Option, body: Body, }, } @@ -212,13 +216,19 @@ impl S3BlockStore { } } - async fn get_range(&self, key: &str, range: ByteRangeSpec) -> Result { + async fn get_range( + &self, + key: &str, + range: ByteRangeSpec, + if_match: Option<&str>, + ) -> Result { match self .client .get_object() .bucket(&self.bucket) .key(key) .range(range.to_http_value()) + .set_if_match(if_match.map(str::to_owned)) .send() .await { @@ -230,6 +240,7 @@ impl S3BlockStore { Ok(RangeGet::Got { range, total, + etag: output.e_tag().map(str::to_owned), body: body(output.body), }) } @@ -244,6 +255,9 @@ impl S3BlockStore { Err(SdkError::ServiceError { err, .. }) if err.code() == Some("InvalidRange") => { Ok(RangeGet::InvalidRange) } + Err(SdkError::ServiceError { err, .. }) if err.code() == Some("PreconditionFailed") => { + Ok(RangeGet::Changed) + } Err(error) => Err(S3Error::from(error).into()), } } @@ -342,7 +356,7 @@ impl S3BlockStore { let (size, reader) = open_sealed(provider, space, id, b, len).await?.into_inner(); Ok(Some(Content::new( size, - AsyncEither::Right(Box::new(reader)), + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader))), ))) } @@ -382,13 +396,15 @@ impl S3BlockStore { spec: ByteRangeSpec, ) -> Result>, S3StoreError> { let key = self.key(space, id); - match self.get_range(&key, spec).await? { - RangeGet::Got { range, total, body } => Ok(Some(RangeRead::Content { + match self.get_range(&key, spec, None).await? { + RangeGet::Got { + range, total, body, .. + } => Ok(Some(RangeRead::Content { total_size: total, range, content: Content::new(range.len(), AsyncEither::Left(body)), })), - RangeGet::Missing => Ok(None), + RangeGet::Missing | RangeGet::Changed => Ok(None), RangeGet::InvalidRange => { let head = self .client @@ -416,8 +432,11 @@ impl S3BlockStore { } /// A range of the block's sealed object, or `None` if it is absent: one - /// GET for the header, one for the segments. Totals and the 416 decision - /// use the plaintext length. + /// GET for the header, one (pinned to the first one's ETag) for the + /// segments. Totals and the 416 decision use the plaintext length. If the + /// object is replaced or removed between the two, the read starts over + /// once, then fails (503); once a header was read it never falls back to + /// the legacy name. async fn range_sealed( &self, name: &SealedName, @@ -431,44 +450,64 @@ impl S3BlockStore { start: 0, end: HEADER_LEN as u64 - 1, }; - let (total, mut head_body) = match self.get_range(&object, header_span).await? { - RangeGet::Missing => return Ok(None), - RangeGet::InvalidRange => { - return Err(AtRestStoreError::Integrity("object shorter than its header").into()) - } - RangeGet::Got { total, body, .. } => (total, body), - }; - let mut header = [0u8; HEADER_LEN]; - head_body - .read_exact(&mut header) - .await - .map_err(|_| AtRestStoreError::Integrity("object shorter than its header"))?; - let plan = plan_range(provider, space, &header, total, spec).await?; - let Some(r) = plan.range else { - return Ok(Some(RangeRead::Unsatisfiable { + let mut seen_header = false; + for _ in 0..2 { + let (total, etag, mut head_body) = + match self.get_range(&object, header_span, None).await? { + RangeGet::Missing | RangeGet::Changed if seen_header => break, + RangeGet::Missing | RangeGet::Changed => return Ok(None), + RangeGet::InvalidRange => { + return Err( + AtRestStoreError::Integrity("object shorter than its header").into(), + ) + } + RangeGet::Got { + total, etag, body, .. + } => (total, etag, body), + }; + seen_header = true; + let mut header = [0u8; HEADER_LEN]; + head_body + .read_exact(&mut header) + .await + .map_err(|_| AtRestStoreError::Integrity("object shorter than its header"))?; + let plan = plan_range(provider, space, &header, total, spec).await?; + let Some(r) = plan.range else { + return Ok(Some(RangeRead::Unsatisfiable { + total_size: plan.plaintext_len(), + })); + }; + let span = ByteRangeSpec::Inclusive { + start: r.ciphertext_start(), + end: r.ciphertext_end(), + }; + let body = match self.get_range(&object, span, etag.as_deref()).await? { + RangeGet::Got { range, body, .. } + if range.start() == r.ciphertext_start() + && range.end() == r.ciphertext_end() => + { + body + } + RangeGet::Got { .. } => { + return Err( + AtRestStoreError::Integrity("ranged GET returned another span").into(), + ) + } + // Replaced, removed or shrunk since the header was read. + RangeGet::Changed | RangeGet::Missing | RangeGet::InvalidRange => continue, + }; + let reader = plan.reader(space, id, r, body)?; + let range = r.plaintext_range(); + return Ok(Some(RangeRead::Content { total_size: plan.plaintext_len(), + range, + content: Content::new( + range.len(), + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader))), + ), })); - }; - let span = ByteRangeSpec::Inclusive { - start: r.ciphertext_start(), - end: r.ciphertext_end(), - }; - let body = match self.get_range(&object, span).await? { - RangeGet::Got { range, body, .. } - if range.start() == r.ciphertext_start() && range.end() == r.ciphertext_end() => - { - body - } - RangeGet::Missing => return Ok(None), - _ => return Err(AtRestStoreError::Integrity("object changed during a read").into()), - }; - let reader = plan.reader(space, id, r, body)?; - let range = r.plaintext_range(); - Ok(Some(RangeRead::Content { - total_size: plan.plaintext_len(), - range, - content: Content::new(range.len(), AsyncEither::Right(Box::new(reader))), - })) + } + Err(AtRestStoreError::Unavailable("the object changed during the read").into()) } async fn range_block( @@ -578,15 +617,18 @@ impl ImmutableReadStore for S3BlockStore { type Error = S3StoreError; type Readable = S3Readable; async fn contains(&self, space: &SpaceId, id: &Hash) -> Result { - match read_plan(&self.view(), space, id)? { - ReadPlan::Legacy | ReadPlan::LegacyWhileLocked => { - self.head_exists(&self.key(space, id)).await - } - ReadPlan::SealedThenLegacy { name, .. } => Ok(self.head_exists(&name.key()).await? - || self.head_exists(&self.key(space, id)).await?), - ReadPlan::Sealed { name, .. } => self.head_exists(&name.key()).await, + // The name a write would use (INTERFACES §7). + match contains_name(&self.view(), space, id)? { + ContainsName::Legacy => self.head_exists(&self.key(space, id)).await, + ContainsName::Sealed(name) => self.head_exists(&name.key()).await, } } + fn prepares(&self) -> bool { + super::sealed::prepares(&self.view()) + } + async fn prepare_read(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(super::sealed::prepare_read(&self.view(), space).await?) + } async fn read( &self, diff --git a/tinycloud-node-server/tests/at_rest_s3_store.rs b/tinycloud-node-server/tests/at_rest_s3_store.rs index 3e61c985..727fa8d8 100644 --- a/tinycloud-node-server/tests/at_rest_s3_store.rs +++ b/tinycloud-node-server/tests/at_rest_s3_store.rs @@ -15,6 +15,7 @@ use parking_lot::RwLock; use tinycloud::{ config::AtRestMode, storage::{ + file_system::TempFileSystemStage, s3::{S3BlockConfig, S3BlockStore, S3StoreError}, sealed::{AtRestHandle, AtRestSource, AtRestStoreError, AtRestView}, }, @@ -25,7 +26,7 @@ use tinycloud_core::{ hash::Hash, storage::{ memory::MemoryStaging, ByteRangeSpec, HashBuffer, ImmutableDeleteStore, ImmutableReadStore, - ImmutableWriteStore, RangeRead, StorageConfig, StorageSetup, StoreSize, + ImmutableStaging, ImmutableWriteStore, RangeRead, StorageConfig, StorageSetup, StoreSize, }, }; @@ -271,3 +272,194 @@ async fn legacy_locked_strict_and_tamper_over_s3() -> Result<()> { ensure!(store.read(&space, &hs).await?.is_none(), "removed"); Ok(()) } + +async fn put_file(store: &S3BlockStore, space: &SpaceId, data: &[u8]) -> Result { + ImmutableWriteStore::::prepare_write(store, space).await?; + let mut staged = TempFileSystemStage.stage(space).await?; + staged.write_all(data).await?; + staged.flush().await?; + Ok(ImmutableWriteStore::::persist(store, space, staged).await?) +} + +async fn range( + store: &S3BlockStore, + space: &SpaceId, + h: &Hash, + spec: ByteRangeSpec, +) -> Result<(u64, Vec)> { + match store.read_range(space, h, spec).await?.context("range")? { + RangeRead::Content { + total_size, + content, + .. + } => { + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + Ok((total_size, out)) + } + RangeRead::Unsatisfiable { .. } => anyhow::bail!("{spec:?} unsatisfiable"), + } +} + +fn status(e: &S3StoreError) -> Option { + match e { + S3StoreError::AtRest(e) => Some(e.status()), + _ => None, + } +} + +/// Temp-file staging seals on S3 too; ranges across the tail and From / +/// Suffix ranges at segment edges match the plaintext. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn temp_file_staging_seals_and_ranges_over_s3() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + for n in [3 * SEG, 2 * SEG + 1, 2 * SEG - 1] { + let data = value(n); + let h = put_file(&store, &space, &data).await?; + let nk = keys().object_name_key()?; + let sealed = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + ensure!( + raw(&store, &sealed).await?.starts_with(b"TCE1"), + "{n}: not sealed" + ); + ensure!(get(&store, &space, &h).await? == data, "{n}: read back"); + for at in [0, 1, SEG - 1, SEG, SEG + 1, 2 * SEG - 1, 2 * SEG, n - 1] { + if at >= n { + continue; + } + for spec in [ + ByteRangeSpec::From { start: at as u64 }, + ByteRangeSpec::Suffix { + length: (n - at) as u64, + }, + ] { + let (total, out) = range(&store, &space, &h, spec).await?; + ensure!(total == n as u64 && out == data[at..], "{n}: {spec:?}"); + } + } + } + Ok(()) +} + +/// Back to `off` with keys: sealed blocks still read; `off` with locked keys +/// answers 503 for a sealed-only block. `strict` with locked keys answers 503 +/// for legacy and sealed blocks alike. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn off_after_encrypt_and_strict_while_locked_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let hl = put(&store, &space, &value(70)).await?; + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + let sealed = value(SEG + 2); + let hs = put(&store, &space, &sealed).await?; + + *switch.0.write() = view(AtRestMode::Off, false, Some(keys())); + ensure!( + get(&store, &space, &hs).await? == sealed, + "off reads sealed" + ); + *switch.0.write() = view(AtRestMode::Off, false, Some(keys().locked())); + let e = store.read(&space, &hs).await.map(|_| ()).unwrap_err(); + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "off + locked: {e}" + ); + + *switch.0.write() = view(AtRestMode::Strict, false, Some(keys().locked())); + for h in [hl, hs] { + let e = store.read(&space, &h).await.map(|_| ()).unwrap_err(); + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "strict + locked: {e}" + ); + } + Ok(()) +} + +/// `contains` checks the name a write would use (INTERFACES §7). +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn contains_checks_the_name_a_write_would_use_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let data = value(90); + let h = put(&store, &space, &data).await?; + ensure!(store.contains(&space, &h).await?, "legacy in off"); + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + ensure!(!store.contains(&space, &h).await?, "legacy only in encrypt"); + put(&store, &space, &data).await?; + ensure!(store.contains(&space, &h).await?, "sealed in encrypt"); + Ok(()) +} + +/// A sealed object replaced (same block, another salt) between a range +/// read's header GET and its segment GET: the segment GET is pinned to the +/// header's ETag, so the read starts over or answers 503, and never returns +/// wrong bytes or a false integrity 500. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn a_replaced_object_never_yields_a_false_integrity_error() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + let data = value(3 * SEG + 5); + let h = put(&store, &space, &data).await?; + let nk = keys().object_name_key()?; + let key = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + let space_str = space.to_string(); + let identity = tinycloud_core::at_rest::BlobIdentity::KvBlock { + space: &space_str, + content_hash: &h, + }; + let dek = keys().current_key(&space_str).await?; + let versions = [ + raw(&store, &key).await?, + tinycloud_core::at_rest::seal(&dek, &identity, data.clone())?, + ]; + ensure!(versions[0] != versions[1], "two encryptions differ by salt"); + + let writer = { + let (client, bucket, key) = (store.client.clone(), store.bucket.clone(), key.clone()); + let versions = versions.clone(); + tokio::spawn(async move { + for i in 0..200usize { + client + .put_object() + .bucket(&bucket) + .key(&key) + .body(versions[i % 2].clone().into()) + .send() + .await + .expect("put"); + } + }) + }; + let (mut served, mut retried_out) = (0, 0); + while !writer.is_finished() { + let spec = ByteRangeSpec::Inclusive { + start: (SEG - 3) as u64, + end: (2 * SEG + 3) as u64, + }; + match store.read_range(&space, &h, spec).await { + Ok(Some(RangeRead::Content { content, .. })) => { + let mut out = Vec::new(); + // A body may still abort if the object changes mid-stream + // after the pinned GET started; it never yields wrong bytes. + if Box::pin(content).read_to_end(&mut out).await.is_ok() { + ensure!(out == data[SEG - 3..=2 * SEG + 3], "wrong bytes served"); + served += 1; + } + } + Ok(other) => anyhow::bail!("unexpected {:?}", other.is_some()), + Err(e) => { + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "a replaced object must not be an integrity error: {e}" + ); + retried_out += 1; + } + } + } + writer.await?; + println!("served {served}, gave up after a retry {retried_out}"); + ensure!(served > 0, "no range read completed"); + Ok(()) +} From f5cb0b08dbdf2aa15718b35b809e7be1aa182f6c Mon Sep 17 00:00:00 2001 From: Sam Gbafa Date: Sat, 17 Oct 2026 16:09:15 +0000 Subject: [PATCH 4/4] fix(TC-887): S3 integrity logs name the sealed object; replaced-object test asserts no aborted body; changelog (S3 review deltas) --- CHANGELOG.md | 2 +- tinycloud-node-server/src/storage/s3.rs | 40 ++++++++++++------- .../tests/at_rest_s3_store.rs | 15 ++++--- 3 files changed, 36 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 50a6bbc5..b2e33bb8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## [Unreleased] -- Encryption at rest for KV blocks, FileSystem store (TC-887 S3, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks are sealed (TCE1) under anonymous names and served back as plaintext; legacy blocks stay readable in `encrypt`, are refused in `strict`, and a node with locked keys answers 503 for sealed data. **Behaviour change on nodes with at-rest keys only:** a writing invocation is now authorized (and its spaces checked as hosted) before any key work, so a put to a space the node does not host, or without authority there, answers **401 instead of 404**; nodes without keys are unchanged. See `docs/at-rest/INTERFACES.md` §7. +- Encryption at rest for KV blocks, FileSystem and S3 stores (TC-887 S3, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks are sealed (TCE1) under anonymous names and served back as plaintext; S3 range reads fetch the header and then only the covering segments, pinned to one object version. Legacy blocks stay readable in `encrypt`, are refused in `strict`, and a node with locked keys answers 503 for sealed data. **Behaviour change on nodes with at-rest keys only:** a writing invocation is now authorized (and its spaces checked as hosted) before any key work, so a put to a space the node does not host, or without authority there, answers **401 instead of 404**; nodes without keys are unchanged. See `docs/at-rest/INTERFACES.md` §7. - New migration `m20261017_000000_epoch_order_space_index` adds `idx_epoch_order_space_parent_child ON epoch_order (space, parent, child)` for the head-epoch anti-join that every write runs under its space lock (TC-1274). On a fresh database the build is instant. On PostgreSQL, if a valid index of that name and shape already exists, the migration issues no DDL and takes no lock on `epoch_order`. If the index of that name is INVALID or has a different shape, the boot fails, and nothing is applied, with an error naming the fix. On an existing database **without** the index, the migration builds it inside the boot's migration transaction. That holds a SHARE lock that blocks writes for the whole build, and the lock wait is bounded by the 30 s `lock_timeout` (TC-1176). **Runbook (PostgreSQL, before upgrading a node with history):** in autocommit, run `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child ON epoch_order (space, parent, child);`. Then confirm `SELECT indisvalid FROM pg_index WHERE indexrelid = 'idx_epoch_order_space_parent_child'::regclass;` returns `t`. If it returns `f`, run `DROP INDEX CONCURRENTLY idx_epoch_order_space_parent_child;` and build it again. `down` keeps the index. MySQL is covered only by the generated statement, because the full MySQL migration chain does not run (TC-1314). - Request events include validated/echoed UUID correlation, typed replacement flags, node-keyed target account/space pseudonyms and per-request database acquisition/execution timings and attempt counts. Driver/actor instrumentation includes failures and request-awaited audit batches; forwarding requires explicit first-party origin classification. A shared core `Pseudonymizer` preserves existing account/space hashes, adds the service/space-scoped database hash primitive for F6, and is derived once at startup for injection into consumers. See `docs/request-telemetry.md` for the wire contract, key custody, restart joinability and coverage/exclusions (TC-959 N2). - CACAO/SIWE delegation ingestion normalizes ECDSA recovery IDs 0/1 to 27/28 before hashing and persisting the canonical DAG-CBOR bytes (TC-1071). Equivalent low-s signatures now share one delegation CID, including the CID used for revocation lookup. High-s signatures are rejected during decoding, matching the existing EIP-191 verifier and EIP-2 low-s convention; UCAN/JWS bytes remain unchanged. Existing stored delegation IDs are not rewritten. diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index a7100b52..a131868d 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -353,10 +353,13 @@ impl S3BlockStore { let Some((len, b)) = self.get_full(&name.key()).await? else { return Ok(None); }; - let (size, reader) = open_sealed(provider, space, id, b, len).await?.into_inner(); + let (size, reader) = open_sealed(provider, space, id, b, len) + .await + .inspect_err(|e| e.log_at(name))? + .into_inner(); Ok(Some(Content::new( size, - AsyncEither::Right(Box::new(LoggedDecrypt::new(reader))), + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader, name))), ))) } @@ -446,6 +449,10 @@ impl S3BlockStore { spec: ByteRangeSpec, ) -> Result>, S3StoreError> { let object = name.key(); + let logged = |e: AtRestStoreError| { + e.log_at(name); + e + }; let header_span = ByteRangeSpec::Inclusive { start: 0, end: HEADER_LEN as u64 - 1, @@ -457,9 +464,10 @@ impl S3BlockStore { RangeGet::Missing | RangeGet::Changed if seen_header => break, RangeGet::Missing | RangeGet::Changed => return Ok(None), RangeGet::InvalidRange => { - return Err( - AtRestStoreError::Integrity("object shorter than its header").into(), - ) + return Err(logged(AtRestStoreError::Integrity( + "object shorter than its header", + )) + .into()) } RangeGet::Got { total, etag, body, .. @@ -467,11 +475,14 @@ impl S3BlockStore { }; seen_header = true; let mut header = [0u8; HEADER_LEN]; - head_body - .read_exact(&mut header) + head_body.read_exact(&mut header).await.map_err(|_| { + logged(AtRestStoreError::Integrity( + "object shorter than its header", + )) + })?; + let plan = plan_range(provider, space, &header, total, spec) .await - .map_err(|_| AtRestStoreError::Integrity("object shorter than its header"))?; - let plan = plan_range(provider, space, &header, total, spec).await?; + .map_err(logged)?; let Some(r) = plan.range else { return Ok(Some(RangeRead::Unsatisfiable { total_size: plan.plaintext_len(), @@ -489,21 +500,22 @@ impl S3BlockStore { body } RangeGet::Got { .. } => { - return Err( - AtRestStoreError::Integrity("ranged GET returned another span").into(), - ) + return Err(logged(AtRestStoreError::Integrity( + "ranged GET returned another span", + )) + .into()) } // Replaced, removed or shrunk since the header was read. RangeGet::Changed | RangeGet::Missing | RangeGet::InvalidRange => continue, }; - let reader = plan.reader(space, id, r, body)?; + let reader = plan.reader(space, id, r, body).map_err(logged)?; let range = r.plaintext_range(); return Ok(Some(RangeRead::Content { total_size: plan.plaintext_len(), range, content: Content::new( range.len(), - AsyncEither::Right(Box::new(LoggedDecrypt::new(reader))), + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader, name))), ), })); } diff --git a/tinycloud-node-server/tests/at_rest_s3_store.rs b/tinycloud-node-server/tests/at_rest_s3_store.rs index 727fa8d8..0577d5dd 100644 --- a/tinycloud-node-server/tests/at_rest_s3_store.rs +++ b/tinycloud-node-server/tests/at_rest_s3_store.rs @@ -440,13 +440,16 @@ async fn a_replaced_object_never_yields_a_false_integrity_error() -> Result<()> }; match store.read_range(&space, &h, spec).await { Ok(Some(RangeRead::Content { content, .. })) => { + // The segment GET is pinned to one version: its body is that + // version's bytes even if the object is replaced meanwhile, + // so a body never aborts and never yields wrong bytes. let mut out = Vec::new(); - // A body may still abort if the object changes mid-stream - // after the pinned GET started; it never yields wrong bytes. - if Box::pin(content).read_to_end(&mut out).await.is_ok() { - ensure!(out == data[SEG - 3..=2 * SEG + 3], "wrong bytes served"); - served += 1; - } + Box::pin(content) + .read_to_end(&mut out) + .await + .context("a served range aborted mid-stream")?; + ensure!(out == data[SEG - 3..=2 * SEG + 3], "wrong bytes served"); + served += 1; } Ok(other) => anyhow::bail!("unexpected {:?}", other.is_some()), Err(e) => {