diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 34fee11b..1d07235d 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -426,6 +426,12 @@ 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) + timeout-minutes: 10 + run: | + (set -a; . "$TC887_RUN_DIR/harness.env"; set +a + cargo test --locked -p tinycloud-node --test at_rest_s3_store -- --ignored) + - name: SQL object store (against the harness MinIO) timeout-minutes: 10 run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 62377ddc..c367f3ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] ### Changes +- Encryption at rest for KV blocks, S3 store (TC-887, off by default). The S3 block store seals new KV blocks under `v1/{space_tag}/…` names like the FileSystem store, with the same mode rules; a sealed range read fetches the header, then only the covering segments, pinned to the same object version (`If-Match`), and answers `503` if the object changes twice during the read. Orphan collection deletes both copies of a block and waits while keys are locked. Both block stores now declare the blocks seal path. Mode `off` is unchanged. - Encryption at rest for KV blocks, FileSystem store (TC-887, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks on FileSystem block storage are sealed (TCE1) under `{blocks}/v1/{space_tag}/…` and served back as plaintext, ranges included; legacy blocks stay readable in `encrypt`, are refused in `strict`, and locked keys answer `503` for sealed data. A writing invocation prepares its spaces' data keys after its preupload authorization and before any upload or write lock, so a refused request creates no key. Orphan collection deletes both copies of a block and waits while keys are locked. The FileSystem store declares the blocks seal path; `encrypt`/`strict` still boot only once SQL artifacts declare theirs. Mode `off` is unchanged. See `docs/at-rest/INTERFACES.md` §7. - Encryption at rest groundwork for KV blocks (TC-887, nothing is sealed yet): the S3 and FileSystem block stores count sealed objects (`v1/{space_tag}/…`) by plaintext size at boot, per space tag, and a space's quota joins them once the at-rest keys can name the tag. The S3 store makes one more `ListObjectsV2` listing at boot, under `v1/`. At-rest storage failures answer `503` (keys locked or key store unavailable) or `500` (integrity, unsupported format, legacy refused in `strict`) on `/invoke` and the KV GET routes. Sealed names carry the same `@N` generation suffix as legacy names (`docs/at-rest/INTERFACES.md` §5). - Deploys: the workflow supports an explicit, reason-recorded rollback to an earlier released dstack image on the unchanged N0 compose. It checks image labels, release tag, ancestry, and identical migration sources while preserving the normal forward-release guard. diff --git a/docs/at-rest/INTERFACES.md b/docs/at-rest/INTERFACES.md index 65f073fb..2ad86a09 100644 --- a/docs/at-rest/INTERFACES.md +++ b/docs/at-rest/INTERFACES.md @@ -304,7 +304,14 @@ 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`) and its `Content-Range` must equal the computed ciphertext span: 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. +- **S3 deletion charges:** a lost DELETE response is settled per copy (legacy and sealed are + tracked apart). The boot listing can't name sealed objects (keys arrive after storage opens), so + a sealed object whose DELETE lands after a restart's listing stays counted until the next + restart (a conservative over-count, never an under-count). - **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/mod.rs b/tinycloud-node-server/src/storage/mod.rs index c0dca765..d627e113 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -19,20 +19,16 @@ use crate::{ BlockStores, }; -/// 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 the blocks seal path. The S3 store does not seal in this build -/// and declares none, so `at_rest::setup` refuses a non-`off` mode with S3 -/// block 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 attaching either +/// declares the blocks seal path. pub fn attach_at_rest(stores: &mut BlockStores, runtime: &Arc) { + let handle = sealed::AtRestHandle::new(runtime.clone()); match stores { - Either::B(fs) => { - runtime.declare_seal_path(SealPath::Blocks); - fs.attach_at_rest(sealed::AtRestHandle::new(runtime.clone())); - } - Either::A(_) => {} + Either::A(s3) => s3.attach_at_rest(handle), + Either::B(fs) => fs.attach_at_rest(handle), } + runtime.declare_seal_path(SealPath::Blocks); } /// One physical blob lifetime; a later generation must never share its diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 01856eaf..88a6dd52 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -11,6 +11,7 @@ use aws_sdk_s3::{ use aws_smithy_http_client::tls; use futures::{ future::Either as AsyncEither, + io::AsyncReadExt, stream::{IntoAsyncRead, Stream, TryStreamExt}, }; use rocket::{async_trait, http::hyper::Uri}; @@ -28,7 +29,9 @@ use std::{ }; use tinycloud_auth::resource::{SpaceId, SPACE_ID_PREFIX}; use tinycloud_core::{ - at_rest::{plaintext_len, SEALED_NAME_PREFIX}, + at_rest::{ + plaintext_len, AtRestView, DataKey, DataKeyProvider, HEADER_LEN, SEALED_NAME_PREFIX, + }, hash::Hash, storage::*, }; @@ -36,7 +39,16 @@ use tokio_util::bytes::Bytes; #[cfg(test)] use super::TestPause; -use super::{file_system, sealed, size::SpaceSizes, BlobGenerationKey, BlobOperationLocks}; +use super::{ + file_system, + sealed::{ + self, contains_name, open_sealed, plan_range, read_plan, removal_name, seal_file, seal_vec, + space_tag_of, write_target, AtRestHandle, AtRestStoreError, ContainsName, LoggedDecrypt, + ReadPlan, SealFileError, SealedName, WriteTarget, + }, + size::SpaceSizes, + BlobGenerationKey, BlobOperationLocks, +}; use tokio::sync::Mutex; #[cfg(test)] use tokio::sync::Notify; @@ -78,7 +90,13 @@ pub struct S3BlockStore { // size until absence is confirmed, so retry refunds exactly once. On // restart, the initial bucket listing reconstructs only physical bytes. deleting: Arc>>, + // The same for sealed copies (TC-887). Kept apart: a boot listing can't + // name sealed objects (keys arrive after storage opens), so `deleting` + // holds only legacy charges, and a block may have both copies. + deleting_sealed: Arc>>, operations: BlobOperationLocks, + // The node's at-rest state (TC-887); without it the store is mode `off`. + at_rest: Option, #[cfg(test)] after_delete_head: TestPause, #[cfg(test)] @@ -117,7 +135,8 @@ impl S3BlockConfig { impl StorageSetup for S3BlockStore { type Error = std::convert::Infallible; async fn create(&self, space: &SpaceId) -> Result<(), Self::Error> { - self.sizes.init_size(space.clone(), None).await; + let tag = space_tag_of(&self.view(), space); + self.sizes.init_size(space.clone(), tag.as_deref()).await; Ok(()) } } @@ -205,7 +224,9 @@ impl S3BlockStore { sizes, pending: Arc::new(Mutex::new(HashMap::new())), deleting: Arc::new(Mutex::new(counted_deletes)), + deleting_sealed: Arc::new(Mutex::new(HashMap::new())), operations: BlobOperationLocks::default(), + at_rest: None, #[cfg(test)] after_delete_head: Arc::new(Mutex::new(None)), #[cfg(test)] @@ -230,11 +251,159 @@ impl S3BlockStore { Self::key_for(space, id, generation) } - 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); + } + + fn view(&self) -> AtRestView { + AtRestHandle::view_of(self.at_rest.as_ref()) + } + + /// `tag`: the space tag of sealed bytes (TC-887), `None` for legacy ones. + async fn increment_size(&self, space: &SpaceId, tag: Option<&str>, size: u64) { + self.sizes.increment_size(space, tag, size).await; + } + async fn decrement_size(&self, space: &SpaceId, tag: Option<&str>, size: u64) { + self.sizes.decrement_size(space, tag, size).await; + } + + /// The stored size of an object, or `None` if absent (HeadObject). + async fn head_size(&self, key: &str) -> Result, S3StoreError> { + match self + .client + .head_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + { + Ok(object) => Ok(Some( + object.content_length().unwrap_or_default().try_into()?, + )), + Err(SdkError::ServiceError(e)) if matches!(e.err(), HeadObjectError::NotFound(_)) => { + Ok(None) + } + Err(e) => Err(S3Error::from(e).into()), + } + } + + /// 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().unwrap_or_default().try_into()?, + body_stream(o.body).into_async_read(), + ))), + Err(SdkError::ServiceError(e)) if matches!(e.err(), GetObjectError::NoSuchKey(_)) => { + Ok(None) + } + Err(e) => Err(S3Error::from(e).into()), + } + } + + /// A ranged GET, optionally pinned to an ETag (`If-Match`). + 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 + { + 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, + etag: output.e_tag().map(str::to_owned), + body: body_stream(output.body).into_async_read(), + }) + } + Err(SdkError::ServiceError(e)) if matches!(e.err(), GetObjectError::NoSuchKey(_)) => { + Ok(RangeGet::Missing) + } + Err(SdkError::ServiceError(e)) if e.err().code() == Some("InvalidRange") => { + Ok(RangeGet::InvalidRange) + } + Err(SdkError::ServiceError(e)) + if e.err().code() == Some("PreconditionFailed") + || e.raw().status().as_u16() == 412 => + { + Ok(RangeGet::Changed) + } + Err(error) => Err(S3Error::from(error).into()), + } } - async fn decrement_size(&self, space: &SpaceId, size: u64) { - self.sizes.decrement_size(space, None, size).await; + + /// Uploads a staged block under the name the at-rest state picks: the + /// legacy name in mode `off`, else the sealed name (TC-887). Only that + /// name dedups: a legacy copy never suppresses a sealed write. The quota + /// charge is the plaintext size either way. + async fn persist_staged( + &self, + space: &SpaceId, + hash: Hash, + generation: i64, + staged: Staged, + ) -> Result<(), S3StoreError> { + match write_target(&self.view(), space, &hash, generation).await? { + WriteTarget::Legacy => { + let object = self.key(space, &hash, generation); + match staged { + Staged::File(file) => { + self.put_if_absent(space, hash, generation, &object, None, || { + file_body(file) + }) + .await + } + Staged::Bytes(bytes) => { + self.put_if_absent(space, hash, generation, &object, None, || async move { + Ok((bytes.len() as u64, ByteStream::from(bytes), ())) + }) + .await + } + } + } + WriteTarget::Sealed { name, key } => { + let (object, tag) = (name.key(), Some(name.space_tag.as_str())); + match staged { + Staged::File(file) => { + self.put_if_absent(space, hash, generation, &object, tag, || { + sealed_file_body(key, space, hash, file) + }) + .await + } + Staged::Bytes(bytes) => { + self.put_if_absent(space, hash, generation, &object, tag, || async move { + let size = bytes.len() as u64; + let sealed = seal_vec(key, space, &hash, bytes).await?; + Ok((size, ByteStream::from(sealed), ())) + }) + .await + } + } + } + } } async fn put_if_absent( @@ -242,6 +411,8 @@ impl S3BlockStore { space: &SpaceId, hash: Hash, generation: i64, + object: &str, + tag: Option<&str>, body: F, ) -> Result<(), S3StoreError> where @@ -252,7 +423,7 @@ impl S3BlockStore { // Dedup, reservation, PUT and settlement are one physical generation // operation. A second KV path cannot charge the same bytes again. let _operation = self.operations.lock(key.clone()).await; - if self.contains(space, &hash, generation).await? { + if self.head_size(object).await?.is_some() { // A retry of an ambiguous PUT can find the object it uploaded. self.pending.lock().await.remove(&key); return Ok(()); @@ -268,14 +439,14 @@ impl S3BlockStore { let mut pending = self.pending.lock().await; if let Entry::Vacant(slot) = pending.entry(key.clone()) { slot.insert(size); - self.increment_size(space, size).await; + self.increment_size(space, tag, size).await; } drop(pending); let upload = self .client .put_object() .bucket(&self.bucket) - .key(self.key(space, &hash, generation)) + .key(object) .body(body) .send() .await; @@ -285,33 +456,365 @@ impl S3BlockStore { Ok(()) } - async fn refund_missing_pending(&self, space: &SpaceId, id: &Hash, generation: i64) { + async fn refund_missing_pending( + &self, + space: &SpaceId, + id: &Hash, + generation: i64, + tag: Option<&str>, + ) { let key = (space.clone(), *id, generation); let mut pending = self.pending.lock().await; if let Some(size) = pending.get(&key).copied() { - self.decrement_size(space, size).await; + self.decrement_size(space, tag, size).await; pending.remove(&key); } } - async fn refund_confirmed_absence(&self, space: &SpaceId, id: &Hash, generation: i64) { + /// Settles the deletion charge `deleting` holds for this generation, if + /// any, together with its upload reservation. Returns whether there was + /// one. + async fn refund_confirmed_absence( + &self, + deleting: &Mutex>, + space: &SpaceId, + id: &Hash, + generation: i64, + tag: Option<&str>, + ) -> bool { let key = (space.clone(), *id, generation); - let mut deleting = self.deleting.lock().await; - if let Some(size) = deleting.get(&key).copied() { - let mut pending = self.pending.lock().await; - // Hold both reservations until the quota update finishes. Once - // it runs, both removals are synchronous and cancellation cannot - // cause a second refund on the next HEAD 404. - self.decrement_size(space, size).await; - deleting.remove(&key); - pending.remove(&key); - } else { - drop(deleting); - self.refund_missing_pending(space, id, generation).await; + let mut deleting = deleting.lock().await; + let Some(size) = deleting.get(&key).copied() else { + return false; + }; + let mut pending = self.pending.lock().await; + // Hold both reservations until the quota update finishes. Once it + // runs, both removals are synchronous and cancellation cannot cause a + // second refund on the next HEAD 404. + self.decrement_size(space, tag, size).await; + deleting.remove(&key); + pending.remove(&key); + true + } + + /// Deletes one stored copy of a generation (`object`), charged `tag`'s + /// bytes. Returns (deleted now, its charge was settled). Callers hold + /// the generation's operation lock. + #[allow(clippy::too_many_arguments)] + async fn remove_object( + &self, + deleting: &Mutex>, + space: &SpaceId, + id: &Hash, + generation: i64, + object: &str, + tag: Option<&str>, + sealed: bool, + ) -> Result<(bool, bool), S3StoreError> { + let size = match self.head_size(object).await? { + // A sealed object is charged its plaintext bytes, as counted. + Some(stored) if sealed => plaintext_len(stored).unwrap_or(0), + Some(stored) => stored, + None => { + let settled = self + .refund_confirmed_absence(deleting, space, id, generation, tag) + .await; + return Ok((false, settled)); + } + }; + #[cfg(test)] + let pause = self.after_delete_head.lock().await.take(); + #[cfg(test)] + if let Some((entered, release)) = pause { + entered.notify_one(); + release.notified().await; + } + deleting + .lock() + .await + .entry((space.clone(), *id, generation)) + .or_insert(size); + self.client + .delete_object() + .bucket(&self.bucket) + .key(object) + .send() + .await + .map_err(S3Error::from)?; + self.refund_confirmed_absence(deleting, space, id, generation, tag) + .await; + Ok((true, true)) + } + + async fn read_legacy( + &self, + space: &SpaceId, + id: &Hash, + generation: i64, + ) -> Result>, S3StoreError> { + Ok(self + .get_full(&self.key(space, id, generation)) + .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 + .inspect_err(|e| e.log_at(name))? + .into_inner(); + Ok(Some(Content::new( + size, + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader, name))), + ))) + } + + async fn read_block( + &self, + space: &SpaceId, + id: &Hash, + generation: i64, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id, generation)? { + ReadPlan::Legacy => self.read_legacy(space, id, generation).await, + ReadPlan::LegacyWhileLocked => match self.read_legacy(space, id, generation).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, generation).await, + } + } + ReadPlan::Sealed { name, provider } => { + match self.read_sealed(&name, &*provider, space, id).await? { + Some(c) => Ok(Some(c)), + None if self + .head_size(&self.key(space, id, generation)) + .await? + .is_some() => + { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } + } + } + + async fn range_legacy( + &self, + space: &SpaceId, + id: &Hash, + generation: i64, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + let key = self.key(space, id, generation); + 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 | RangeGet::Changed => Ok(None), + RangeGet::InvalidRange => Ok(self + .head_size(&key) + .await? + .map(|total_size| RangeRead::Unsatisfiable { total_size })), + } + } + + /// A range of the block's sealed object, or `None` if it is absent: one + /// 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, + provider: &dyn DataKeyProvider, + space: &SpaceId, + id: &Hash, + 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, + }; + 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(logged(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(|_| { + logged(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(), + })); + }; + 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(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).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, name))), + ), + })); + } + Err(AtRestStoreError::Unavailable("the object changed during the read").into()) + } + + async fn range_block( + &self, + space: &SpaceId, + id: &Hash, + generation: i64, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id, generation)? { + ReadPlan::Legacy => self.range_legacy(space, id, generation, spec).await, + ReadPlan::LegacyWhileLocked => { + match self.range_legacy(space, id, generation, 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, generation, 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_size(&self.key(space, id, generation)) + .await? + .is_some() => + { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } } } } +/// An S3 object body as an `AsyncRead`. +type Body = IntoAsyncRead; + +/// 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>>; + +/// 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, + /// `If-Match` failed: the object was replaced since the ETag was read. + Changed, + Got { + range: ResolvedByteRange, + total: u64, + etag: Option, + body: Body, + }, +} + +/// Seals a staged plaintext file into a temporary file and returns its +/// plaintext size and upload body. The plaintext stage is removed once +/// sealed; the sealed file lives until the returned guard drops. +async fn sealed_file_body( + key: DataKey, + space: &SpaceId, + hash: Hash, + file: file_system::TempFileStage, +) -> Result<(u64, ByteStream, tempfile::TempPath), S3StoreError> { + let size = file.size().await?; + let (_file, src) = file.into_inner(); + let out = tempfile::NamedTempFile::new()?; + seal_file(&key, space, &hash, &src, size, out.reopen()?).await?; + drop(src); + let path = out.into_temp_path(); + let body = ByteStream::from_path(&path).await?; + Ok((size, body, path)) +} + /// Plaintext bytes per space tag of the sealed objects (`v1/{space_tag}/…`, /// TC-887), from a second listing restricted to that prefix: the KV listing /// above only sees legacy names. @@ -378,30 +881,38 @@ impl From for S3StoreError { } } +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; + type Readable = S3Readable; + /// Whether the block exists under the name a write would use (INTERFACES + /// §7): in `encrypt`/`strict` only the sealed name counts. async fn contains( &self, space: &SpaceId, id: &Hash, generation: i64, ) -> Result { - match self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id, generation)) - .send() - .await - { - Ok(_) => Ok(true), - Err(SdkError::ServiceError(e)) if matches!(e.err(), HeadObjectError::NotFound(_)) => { - Ok(false) - } - Err(e) => Err(S3Error::from(e).into()), - } + let object = match contains_name(&self.view(), space, id, generation)? { + ContainsName::Legacy => self.key(space, id, generation), + ContainsName::Sealed(name) => name.key(), + }; + Ok(self.head_size(&object).await?.is_some()) + } + fn prepares(&self) -> bool { + sealed::prepares(&self.view()) + } + async fn prepare_read(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(sealed::prepare_read(&self.view(), space).await?) } async fn read( @@ -411,23 +922,9 @@ impl ImmutableReadStore for S3BlockStore { generation: i64, ) -> Result>, Self::Error> { let start = Instant::now(); - let res = self - .client - .get_object() - .bucket(&self.bucket) - .key(self.key(space, id, generation)) - .send() - .await; - let result = match res { - Ok(o) => Ok(Some(Content::new( - o.content_length().unwrap_or_default().try_into()?, - body_stream(o.body).into_async_read(), - ))), - Err(SdkError::ServiceError(e)) if matches!(e.err(), GetObjectError::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, generation)).await; crate::prometheus::observe_stage( crate::prometheus::InvocationStage::BlockRead, crate::prometheus::StageOutcome::from(result.is_ok()), @@ -443,53 +940,7 @@ impl ImmutableReadStore for S3BlockStore { generation: i64, range: ByteRangeSpec, ) -> Result>, Self::Error> { - let key = self.key(space, id, generation); - 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(), body_stream(output.body).into_async_read()); - Ok(Some(RangeRead::Content { - total_size, - range, - content, - })) - } - Err(SdkError::ServiceError(e)) if matches!(e.err(), GetObjectError::NoSuchKey(_)) => { - return Ok(None) - } - Err(SdkError::ServiceError(e)) if e.err().code() == Some("InvalidRange") => { - let head = self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id, generation)) - .send() - .await; - match head { - Ok(output) => Ok(Some(RangeRead::Unsatisfiable { - total_size: output.content_length().unwrap_or_default().try_into()?, - })), - Err(SdkError::ServiceError(e)) - if matches!(e.err(), HeadObjectError::NotFound(_)) => - { - Ok(None) - } - Err(error) => Err(S3Error::from(error).into()), - } - } - Err(error) => Err(S3Error::from(error).into()), - } + Box::pin(self.range_block(space, id, generation, range)).await } } @@ -519,61 +970,62 @@ 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>, generation: i64, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - self.put_if_absent(space, hash, generation, || async move { - Ok((f.len() as u64, ByteStream::from(f), ())) - }) - .await?; + Box::pin(self.persist_staged(space, hash, generation, 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>, generation: i64, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - self.put_if_absent(space, hash, generation, || file_body(f)) - .await?; + Box::pin(self.persist_staged(space, hash, generation, 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 } } @@ -582,94 +1034,73 @@ impl ImmutableWriteStore Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer< as ImmutableStaging>::Writable>, generation: i64, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - match f { - AsyncEither::Left(t_file) => { - self.put_if_absent(space, hash, generation, || file_body(t_file)) - .await?; - } - AsyncEither::Right(b) => { - self.put_if_absent(space, hash, generation, || async move { - Ok((b.len() as u64, ByteStream::from(b), ())) - }) - .await?; - } - } + let staged = match f { + AsyncEither::Left(file) => Staged::File(file), + AsyncEither::Right(bytes) => Staged::Bytes(bytes), + }; + Box::pin(self.persist_staged(space, hash, generation, staged)).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 ImmutableDeleteStore for S3BlockStore { type Error = S3StoreError; + /// Deletes both copies of the generation: the sealed name whenever keys + /// are set up, then the legacy name. With keys set up but locked nothing + /// is deleted (503) and the collector retries later. async fn remove( &self, space: &SpaceId, id: &Hash, generation: i64, ) -> Result, Self::Error> { + let sealed = removal_name(&self.view(), space, id, generation)?; + let tag = sealed.as_ref().map(|name| name.space_tag.as_str()); // A stale positive HEAD must not recreate a charge after another // deleter has already confirmed absence and refunded it. let _operation = self.operations.lock((space.clone(), *id, generation)).await; - let size: u64 = match self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id, generation)) - .send() - .await - { - Ok(object) => object.content_length().unwrap_or_default().try_into()?, - Err(SdkError::ServiceError(e)) if matches!(e.err(), HeadObjectError::NotFound(_)) => { - self.refund_confirmed_absence(space, id, generation).await; - return Ok(None); - } - Err(error) => return Err(S3Error::from(error).into()), - }; - #[cfg(test)] - let pause = self.after_delete_head.lock().await.take(); - #[cfg(test)] - if let Some((entered, release)) = pause { - entered.notify_one(); - release.notified().await; + let (mut removed, mut settled) = (false, false); + if let Some(name) = &sealed { + let (r, s) = self + .remove_object( + &self.deleting_sealed, + space, + id, + generation, + &name.key(), + tag, + true, + ) + .await?; + (removed, settled) = (r, s); } - self.deleting - .lock() - .await - .entry((space.clone(), *id, generation)) - .or_insert(size); - match self - .client - .delete_object() - .bucket(&self.bucket) - .key(self.key(space, id, generation)) - .send() - .await - { - Ok(_) => { - self.refund_confirmed_absence(space, id, generation).await; - Ok(Some(())) - } - Err(e) => Err(S3Error::from(e).into()), + let legacy = self.key(space, id, generation); + let (r, s) = self + .remove_object(&self.deleting, space, id, generation, &legacy, None, false) + .await?; + (removed, settled) = (removed || r, settled || s); + if !settled { + // Neither copy exists, nor did a deletion charge: an upload whose + // PUT never landed still holds its reservation. + self.refund_missing_pending(space, id, generation, tag) + .await; } + Ok(removed.then_some(())) } async fn cleanup_stages( &self, @@ -686,10 +1117,12 @@ 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 tag = space_tag_of(&self.view(), space); + Ok(self.sizes.get_size(space, tag.as_deref()).await) } async fn init_size(&self, space: &SpaceId) -> Result { - Ok(self.sizes.init_size(space.clone(), None).await) + let tag = space_tag_of(&self.view(), space); + Ok(self.sizes.init_size(space.clone(), tag.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..29784f2d --- /dev/null +++ b/tinycloud-node-server/tests/at_rest_s3_store.rs @@ -0,0 +1,594 @@ +//! TC-887: 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::storage::{ + file_system::TempFileSystemStage, + s3::{S3BlockConfig, S3BlockStore, S3StoreError}, + sealed::{AtRestHandle, AtRestStoreError}, +}; +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::{ + at_rest::{ + ciphertext_len, sealed_object_name, AtRestMode, AtRestSource, AtRestView, DataKeyProvider, + FixedKeyProvider, SEGMENT_LEN, + }, + hash::Hash, + storage::{ + memory::MemoryStaging, ByteRangeSpec, HashBuffer, ImmutableDeleteStore, ImmutableReadStore, + ImmutableStaging, 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 encrypt() -> AtRestView { + view(AtRestMode::Encrypt, true, Some(keys())) +} + +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_gen( + store: &S3BlockStore, + space: &SpaceId, + data: &[u8], + generation: i64, +) -> 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, generation).await +} + +async fn put(store: &S3BlockStore, space: &SpaceId, data: &[u8]) -> Result { + put_gen(store, space, data, 0).await +} + +async fn get_gen( + store: &S3BlockStore, + space: &SpaceId, + h: &Hash, + generation: i64, +) -> Result> { + let content = store + .read(space, h, generation) + .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 get(store: &S3BlockStore, space: &SpaceId, h: &Hash) -> Result> { + get_gen(store, space, h, 0).await +} + +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()) +} + +async fn exists(store: &S3BlockStore, key: &str) -> Result { + match store + .client + .head_object() + .bucket(&store.bucket) + .key(key) + .send() + .await + { + Ok(_) => Ok(true), + Err(e) if e.raw_response().is_some_and(|r| r.status().as_u16() == 404) => Ok(false), + Err(e) => Err(e.into()), + } +} + +fn sealed_key(space: &SpaceId, h: &Hash) -> String { + sealed_object_name(&keys().object_name_key().unwrap(), &space.to_string(), h) +} + +fn legacy_key(space: &SpaceId, h: &Hash) -> String { + format!( + "{space}/{}", + base64::encode_config(h.as_ref(), base64::URL_SAFE) + ) +} + +fn status(e: &S3StoreError) -> Option { + match e { + S3StoreError::AtRest(e) => Some(e.status()), + _ => None, + } +} + +#[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(encrypt()).await?; + 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 bytes = raw(&store, &sealed_key(&space, &h)).await?; + ensure!( + bytes.len() as u64 == ciphertext_len(n as u64)?, + "{n}: stored size" + ); + ensure!(bytes.starts_with(b"TCE1"), "{n}: not TCE1"); + ensure!( + !exists(&store, &legacy_key(&space, &h)).await?, + "{n}: legacy copy" + ); + 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, 0, 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, 0, 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 (the `v1/` listing), by space tag, with no legacy objects + // for the space; boot seeding keeps them. + let (reopened, _) = open_store(encrypt()).await?; + ensure!( + reopened.total_size(&space).await? == Some(total), + "boot-scan plaintext quota" + ); + ensure!( + reopened.init_size(&space).await? == total, + "seeding keeps sealed bytes" + ); + Ok(()) +} + +/// Each generation has its own sealed name, with the legacy name's `@N` +/// suffix; removing one generation leaves the other. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn generations_get_their_own_sealed_names_over_s3() -> Result<()> { + let (store, _, space) = store(encrypt()).await?; + let data = value(SEG + 3); + let h = put_gen(&store, &space, &data, 0).await?; + put_gen(&store, &space, &data, 2).await?; + let base = sealed_key(&space, &h); + ensure!(raw(&store, &base).await?.starts_with(b"TCE1")); + ensure!(raw(&store, &format!("{base}@2")) + .await? + .starts_with(b"TCE1")); + ensure!(store.total_size(&space).await? == Some(2 * data.len() as u64)); + ensure!(store.remove(&space, &h, 0).await? == Some(())); + ensure!(store.read(&space, &h, 0).await?.is_none()); + ensure!(get_gen(&store, &space, &h, 2).await? == data); + ensure!(store.total_size(&space).await? == Some(data.len() as u64)); + ensure!(store.remove(&space, &h, 2).await? == Some(())); + ensure!(store.total_size(&space).await? == Some(0)); + 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() = encrypt(); + 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| status(e) == Some(rocket::http::Status::ServiceUnavailable); + ensure!( + put(&store, &space, &value(9)) + .await + .as_ref() + .is_err_and(unavailable), + "locked write" + ); + ensure!( + store + .read(&space, &hs, 0) + .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, 0).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 key = sealed_key(&space, &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, 0).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, 0).await? == Some(()), "remove"); + ensure!(store.read(&space, &hs, 0).await?.is_none(), "removed"); + Ok(()) +} + +/// Orphan removal deletes both copies of a block; while keys are locked it +/// deletes nothing (it cannot name the sealed copy) and fails with 503, so +/// the collector retries later. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn removal_deletes_both_copies_and_waits_while_locked_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let data = value(900); + let h = put(&store, &space, &data).await?; + *switch.0.write() = encrypt(); + put(&store, &space, &data).await?; + ensure!(exists(&store, &legacy_key(&space, &h)).await?); + ensure!(exists(&store, &sealed_key(&space, &h)).await?); + ensure!(store.total_size(&space).await? == Some(1800)); + + *switch.0.write() = view(AtRestMode::Encrypt, false, Some(keys().locked())); + let e = store.remove(&space, &h, 0).await.unwrap_err(); + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "{e}" + ); + ensure!(exists(&store, &legacy_key(&space, &h)).await?); + ensure!(exists(&store, &sealed_key(&space, &h)).await?); + + *switch.0.write() = view(AtRestMode::Off, false, Some(keys())); + ensure!(store.remove(&space, &h, 0).await? == Some(())); + ensure!(!exists(&store, &legacy_key(&space, &h)).await?); + ensure!(!exists(&store, &sealed_key(&space, &h)).await?); + ensure!(store.total_size(&space).await? == Some(0)); + ensure!(store.remove(&space, &h, 0).await?.is_none()); + ensure!(store.total_size(&space).await? == Some(0)); + 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, 0).await?) +} + +async fn range( + store: &S3BlockStore, + space: &SpaceId, + h: &Hash, + spec: ByteRangeSpec, +) -> Result<(u64, Vec)> { + match store + .read_range(space, h, 0, 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"), + } +} + +/// 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(encrypt()).await?; + for n in [3 * SEG, 2 * SEG + 1, 2 * SEG - 1] { + let data = value(n); + let h = put_file(&store, &space, &data).await?; + ensure!( + raw(&store, &sealed_key(&space, &h)) + .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() = encrypt(); + 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, 0).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, 0).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, 0).await?, "legacy in off"); + *switch.0.write() = encrypt(); + ensure!( + !store.contains(&space, &h, 0).await?, + "legacy only in encrypt" + ); + put(&store, &space, &data).await?; + ensure!(store.contains(&space, &h, 0).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(encrypt()).await?; + let data = value(3 * SEG + 5); + let h = put(&store, &space, &data).await?; + let key = sealed_key(&space, &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, 0, 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(); + 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) => { + 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(()) +}