diff --git a/tinycloud-node-server/Cargo.toml b/tinycloud-node-server/Cargo.toml index 1f47ea62..0b1a61c0 100644 --- a/tinycloud-node-server/Cargo.toml +++ b/tinycloud-node-server/Cargo.toml @@ -117,7 +117,8 @@ path = "../tinycloud-at-rest-escrow/" [dev-dependencies.tinycloud-core] path = "../tinycloud-core/" default-features = false -features = ["sqlite", "postgres", "mysql", "tokio", "verification-count-test-hook"] +# TC-887: `test-util` provides at_rest::FixedKeyProvider for sealed-store tests. +features = ["sqlite", "postgres", "mysql", "tokio", "verification-count-test-hook", "test-util"] [dev-dependencies] # TC-887: escrow holders decrypting their shares in tests/at_rest_ceremony.rs. diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index a6c8ad83..2a27634f 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -137,6 +137,36 @@ fn database_failure( } } +/// TC-887: the status for an invocation whose block-store step failed for +/// at-rest reasons (503 keys unavailable, 500 integrity/unsupported), if so. +fn at_rest_invoke_status( + e: &TxStoreError, +) -> Option { + match e { + TxStoreError::StoreRead(err) | TxStoreError::StoreWrite(err) => { + crate::storage::sealed::block_store_at_rest_error(err).map(|e| e.status()) + } + _ => None, + } +} + +/// A KV read failure: 503/500 for at-rest causes (TC-887), 500 otherwise. +pub(crate) fn kv_read_failure( + e: tinycloud_core::storage::either::EitherError< + DbErr, + ::Error, + >, +) -> (Status, String) { + use tinycloud_core::storage::either::EitherError; + let status = match &e { + EitherError::B(err) => crate::storage::sealed::block_store_at_rest_error(err) + .map(|e| e.status()) + .unwrap_or(Status::InternalServerError), + EitherError::A(_) => Status::InternalServerError, + }; + (status, e.to_string()) +} + /// `/invoke` failure: plain text for everything except storage rejections, /// which carry the structured TC-619 JSON body. #[derive(Debug)] @@ -426,7 +456,7 @@ pub async fn signed_kv_get( if kv_result.is_ok() { "ok" } else { "error" }, kv_start.elapsed(), ); - return match kv_result.map_err(|e| (Status::InternalServerError, e.to_string()))? { + return match kv_result.map_err(kv_read_failure)? { Some((metadata, hash, content)) => { validate_signed_kv_hash_binding(&ticket, &hash)?; Ok(SignedKvReadResponse::from_range(metadata, hash, content)) @@ -441,7 +471,7 @@ pub async fn signed_kv_get( if kv_result.is_ok() { "ok" } else { "error" }, kv_start.elapsed(), ); - match kv_result.map_err(|e| (Status::InternalServerError, e.to_string()))? { + match kv_result.map_err(kv_read_failure)? { Some((md, hash, content)) => { validate_signed_kv_hash_binding(&ticket, &hash)?; Ok(SignedKvReadResponse::Full { @@ -2229,6 +2259,9 @@ async fn invoke_impl( database_failure(error, write_may_have_applied, e.to_string()).into() ); } + if let Some(status) = at_rest_invoke_status(&e) { + return Err((status, e.to_string()).into()); + } Err(( match &e { TxStoreError::Tx(TxError::SpaceNotFound) => Status::NotFound, diff --git a/tinycloud-node-server/src/routes/public.rs b/tinycloud-node-server/src/routes/public.rs index 2aac4626..8465eb78 100644 --- a/tinycloud-node-server/src/routes/public.rs +++ b/tinycloud-node-server/src/routes/public.rs @@ -238,7 +238,7 @@ pub async fn public_kv_get( let result = tinycloud .public_kv_get(&space_id, &key) .await - .map_err(|e| (Status::InternalServerError, e.to_string()))?; + .map_err(super::kv_read_failure)?; match result { Some((md, hash, content)) => { @@ -286,7 +286,7 @@ pub async fn public_kv_head( let result = tinycloud .public_kv_get(&space_id, &key) .await - .map_err(|e| (Status::InternalServerError, e.to_string()))?; + .map_err(super::kv_read_failure)?; match result { Some((md, hash, _content)) => { diff --git a/tinycloud-node-server/src/storage/file_system.rs b/tinycloud-node-server/src/storage/file_system.rs index 1896dbb5..3250857c 100644 --- a/tinycloud-node-server/src/storage/file_system.rs +++ b/tinycloud-node-server/src/storage/file_system.rs @@ -16,7 +16,7 @@ use std::{ }; use tempfile::{NamedTempFile, PathPersistError}; use tinycloud_auth::{resource::SpaceId, ssi::dids::DIDBuf}; -use tinycloud_core::{hash::Hash, storage::*}; +use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*}; use tokio::{ fs::{create_dir_all, metadata, remove_file, File}, io::{AsyncReadExt, AsyncSeekExt}, @@ -34,7 +34,10 @@ pub struct FileSystemStore { impl FileSystemStore { async fn new(path: PathBuf) -> Result { // get the size of the directory - let sizes = store_sizes(&path).await?.into(); + let sizes = SpaceSizes::from_scan( + store_sizes(&path).await?, + sealed_sizes(&path.join(SEALED_DIR)).await?, + ); Ok(Self { path, sizes }) } @@ -46,10 +49,10 @@ impl FileSystemStore { } async fn increment_size(&self, space: &SpaceId, size: u64) { - self.sizes.increment_size(space, size).await; + self.sizes.increment_size(space, None, size).await; } async fn decrement_size(&self, space: &SpaceId, size: u64) { - self.sizes.decrement_size(space, size).await; + self.sizes.decrement_size(space, None, size).await; } } @@ -107,6 +110,8 @@ pub enum FileSystemStoreError { Io(#[from] IoError), #[error(transparent)] Persist(#[from] PathPersistError), + #[error(transparent)] + AtRest(#[from] super::sealed::AtRestStoreError), } #[async_trait] @@ -168,7 +173,7 @@ impl ImmutableReadStore for FileSystemStore { impl StoreSize for FileSystemStore { type Error = FileSystemStoreError; async fn total_size(&self, space: &SpaceId) -> Result, Self::Error> { - Ok(self.sizes.get_size(space).await) + Ok(self.sizes.get_size(space, None).await) } } @@ -177,11 +182,15 @@ async fn store_sizes>(path: &P) -> Result, ReadDirStream::new(tokio::fs::read_dir(path).await?) // for every entry in the store dir .try_fold(HashMap::new(), |mut acc, entry| async move { - // if its a directory and the suffix is a valid string + // if its a directory and the suffix is a valid string (the sealed + // `v1` directory is scanned by `sealed_sizes`) if let (true, Ok(ref suffix)) = ( entry.metadata().await?.is_dir(), entry.file_name().into_string(), ) { + if suffix == SEALED_DIR { + return Ok(acc); + } let mut ds = ReadDirStream::new(tokio::fs::read_dir(entry.path()).await?); let did: DIDBuf = ["did:", suffix.as_str()].concat().parse().map_err(|_| { IoError::new(ErrorKind::InvalidData, format!("Invalid DID: {suffix}")) @@ -207,6 +216,42 @@ async fn store_sizes>(path: &P) -> Result, .await } +/// The sealed-object directory under the store root: `v1/{space_tag}/{object_tag}`. +const SEALED_DIR: &str = "v1"; + +/// Plaintext bytes per space tag under `v1/` (TC-887). Missing directory: none. +async fn sealed_sizes(path: &Path) -> Result, IoError> { + let mut sizes = HashMap::new(); + let mut invalid = super::sealed::InvalidLengthLog::default(); + let mut tags = match tokio::fs::read_dir(path).await { + Ok(dir) => ReadDirStream::new(dir), + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(sizes), + Err(e) => return Err(e), + }; + while let Some(tag) = tags.try_next().await? { + let (true, Ok(name)) = ( + tag.metadata().await?.is_dir(), + tag.file_name().into_string(), + ) else { + continue; + }; + let mut total = 0u64; + let mut objects = ReadDirStream::new(tokio::fs::read_dir(tag.path()).await?); + while let Some(object) = objects.try_next().await? { + let meta = object.metadata().await?; + if meta.is_file() { + total += plaintext_len(meta.len()).unwrap_or_else(|| { + invalid.record(meta.len()); + 0 + }); + } + } + sizes.insert(name, total); + } + invalid.finish(); + Ok(sizes) +} + async fn space_size>(path: &P) -> Result { // get the sum size of all files in this directory (do not recurse into subdirectories) ReadDirStream::new(tokio::fs::read_dir(path).await?) @@ -492,4 +537,34 @@ mod test { None ); } + + /// A store holding sealed objects under `v1/` still opens, and counts + /// their plaintext bytes per space tag, not under a space. + #[tokio::test] + async fn boot_scan_counts_sealed_objects_by_tag() { + use tinycloud_core::at_rest::ciphertext_len; + let dir = tempfile::tempdir().unwrap(); + let space_id: SpaceId = "tinycloud:key:test:default".parse().unwrap(); + let legacy = dir + .path() + .join(space_id.suffix()) + .join(space_id.name().as_str()); + std::fs::create_dir_all(&legacy).unwrap(); + std::fs::write(legacy.join("obj"), [1u8; 11]).unwrap(); + let tag = dir.path().join(SEALED_DIR).join("tagA"); + std::fs::create_dir_all(&tag).unwrap(); + for (name, plain) in [("o1", 100u64), ("o2", 70_000)] { + let len = ciphertext_len(plain).unwrap() as usize; + std::fs::write(tag.join(name), vec![0u8; len]).unwrap(); + } + // Not a valid TCE1 length: counted as 0. + std::fs::write(tag.join("bad"), [0u8; 70]).unwrap(); + + let store = FileSystemConfig::new(dir.path()).open().await.unwrap(); + assert_eq!(store.total_size(&space_id).await.unwrap(), Some(11)); + assert_eq!( + store.sizes.get_size(&space_id, Some("tagA")).await, + Some(11 + 100 + 70_000) + ); + } } diff --git a/tinycloud-node-server/src/storage/mod.rs b/tinycloud-node-server/src/storage/mod.rs index 80d92074..e0cad057 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -1,4 +1,5 @@ pub mod at_rest_mirror; pub mod file_system; pub mod s3; +pub mod sealed; pub mod size; diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 932baf84..5935a343 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -15,9 +15,13 @@ use serde::{Deserialize, Serialize}; use serde_with::{serde_as, DisplayFromStr}; use std::{collections::HashMap, io::Error as IoError, ops::AddAssign, time::Instant}; use tinycloud_auth::resource::SpaceId; -use tinycloud_core::{hash::Hash, storage::*}; +use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*}; -use super::{file_system, size::SpaceSizes}; +use super::{ + file_system, + sealed::{self, scanned_object, ScannedObject}, + size::SpaceSizes, +}; async fn aws_config() -> SdkConfig { aws_config::from_env().load().await @@ -70,31 +74,48 @@ pub(crate) async fn new_client(config: &S3BlockConfig) -> Client { impl S3BlockStore { async fn new_(config: &S3BlockConfig) -> Result { let client = new_client(config).await; - let sizes = client + let (spaces, sealed, invalid) = client .list_objects_v2() .bucket(&config.bucket) .into_paginator() .send() - // get the sum of all objects in each page - .try_fold(HashMap::new(), |mut acc, page| async move { - // get the sum of all objects per space in this particular page - for (space, obj_size) in page.contents.into_iter().flatten().filter_map(|content| { - content.key().and_then(|key| { - let (o, _) = key.rsplit_once('/')?; - let space: SpaceId = o.parse().ok()?; - if content.size() > 0 { - Some((space, content.size() as u64)) - } else { - None + // sum stored plaintext bytes per space (legacy names) and per + // space tag (sealed names) in each page + .try_fold( + ( + HashMap::new(), + HashMap::new(), + sealed::InvalidLengthLog::default(), + ), + |(mut spaces, mut sealed, mut invalid), page| async move { + for content in page.contents.into_iter().flatten() { + let (Some(key), size) = (content.key(), content.size().max(0) as u64) + else { + continue; + }; + match scanned_object(key) { + ScannedObject::Legacy(space) if size > 0 => { + spaces.entry(space).or_insert(0).add_assign(size); + } + ScannedObject::Sealed(tag) => { + let plaintext = plaintext_len(size).unwrap_or_else(|| { + invalid.record(size); + 0 + }); + sealed + .entry(tag.to_owned()) + .or_insert(0) + .add_assign(plaintext); + } + _ => {} } - }) - }) { - acc.entry(space).or_insert(0).add_assign(obj_size); - } - Ok(acc) - }) - .await? - .into(); + } + Ok((spaces, sealed, invalid)) + }, + ) + .await?; + invalid.finish(); + let sizes = SpaceSizes::from_scan(spaces, sealed); Ok(S3BlockStore { client, bucket: config.bucket.clone(), @@ -111,10 +132,10 @@ impl S3BlockStore { } async fn increment_size(&self, space: &SpaceId, size: u64) { - self.sizes.increment_size(space, size).await; + self.sizes.increment_size(space, None, size).await; } async fn decrement_size(&self, space: &SpaceId, size: u64) { - self.sizes.decrement_size(space, size).await; + self.sizes.decrement_size(space, None, size).await; } /// The stored size of an object, or `None` if absent. HeadObject, not @@ -170,6 +191,8 @@ pub enum S3StoreError { Length(#[from] std::num::TryFromIntError), #[error("invalid S3 Content-Range response: {0}")] InvalidContentRange(String), + #[error(transparent)] + AtRest(#[from] sealed::AtRestStoreError), } #[async_trait] @@ -482,7 +505,7 @@ 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).await) + Ok(self.sizes.get_size(space, None).await) } } diff --git a/tinycloud-node-server/src/storage/sealed.rs b/tinycloud-node-server/src/storage/sealed.rs new file mode 100644 index 00000000..c289cd6a --- /dev/null +++ b/tinycloud-node-server/src/storage/sealed.rs @@ -0,0 +1,456 @@ +//! Encryption at rest for KV blocks (TC-887 S3): the decisions and the +//! envelope plumbing shared by the S3 and FileSystem block stores. +//! +//! Rules (agreed with the key-continuity work; `docs/at-rest/INTERFACES.md`): +//! - mode `off` writes plaintext under the legacy name, exactly as before; +//! any other mode seals iff [`AtRestView::encrypting`], else 503, never a +//! plaintext fallback; +//! - reads try the sealed name whenever keys exist, then the legacy name, +//! unless `strict` (legacy refused); a sealed read under locked keys is a +//! 503; integrity and unsupported-format failures are hard errors, never +//! treated as legacy. +//! +//! Legacy vs sealed is decided by name (`{space}/…` vs `v1/{space_tag}/…`), +//! never by the stored bytes. + +use crate::at_rest::AtRestRuntime; +use crate::config::AtRestMode; +use futures::io::{AsyncRead, AsyncReadExt}; +use std::{path::Path, sync::Arc}; +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::{ + at_rest::{ + space_tag, BlobIdentity, DataKey, DataKeyError, DataKeyProvider, DecryptReader, + EncryptReader, EnvelopeError, Header, ObjectNameKey, SegmentRange, HEADER_LEN, + }, + hash::Hash, + storage::{ByteRangeSpec, Content}, +}; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +/// Why a KV block operation failed for at-rest reasons. +#[derive(Debug, thiserror::Error)] +pub enum AtRestStoreError { + /// Keys are locked, not set up, or the encrypt gate is closed. Retryable + /// once an operator acts: answer 503. + #[error("encrypted storage is unavailable: {0}")] + Unavailable(&'static str), + /// The key store could not be read (database or mirror). Answer 503. + #[error("at-rest key store failed: {0}")] + KeyStore(String), + /// The sealed object failed authentication or does not match its header. + #[error("stored object failed integrity checks: {0}")] + Integrity(&'static str), + /// A well-formed header of a later format generation. Unauthenticated, so + /// it is a hard error like an integrity failure. + #[error("unsupported at-rest format version {0}")] + Unsupported(u8), + /// `strict` mode refuses legacy plaintext objects. + #[error("legacy plaintext object refused in strict mode")] + LegacyRefused, +} + +impl AtRestStoreError { + /// The HTTP status a request failing with this error should get. + pub fn status(&self) -> rocket::http::Status { + use rocket::http::Status; + match self { + Self::Unavailable(_) | Self::KeyStore(_) => Status::ServiceUnavailable, + Self::Integrity(_) | Self::Unsupported(_) | Self::LegacyRefused => { + Status::InternalServerError + } + } + } +} + +/// The at-rest cause of a node block-store error, if it has one. +pub fn block_store_at_rest_error( + e: &tinycloud_core::storage::either::EitherError< + super::s3::S3StoreError, + super::file_system::FileSystemStoreError, + >, +) -> Option<&AtRestStoreError> { + use tinycloud_core::storage::either::EitherError; + match e { + EitherError::A(super::s3::S3StoreError::AtRest(e)) + | EitherError::B(super::file_system::FileSystemStoreError::AtRest(e)) => Some(e), + _ => None, + } +} + +impl From for AtRestStoreError { + fn from(e: DataKeyError) -> Self { + match e { + DataKeyError::Locked => Self::Unavailable("at-rest keys are locked"), + DataKeyError::UnknownKey { .. } => Self::Integrity("header names an unknown data key"), + other => Self::KeyStore(other.to_string()), + } + } +} + +impl From for AtRestStoreError { + fn from(e: EnvelopeError) -> Self { + match e { + EnvelopeError::Unsupported { version } => Self::Unsupported(version), + EnvelopeError::Header(_) => Self::Integrity("invalid header"), + EnvelopeError::KeyMismatch => Self::Integrity("header names another data key"), + EnvelopeError::TooLarge => Self::Integrity("object too large"), + EnvelopeError::LengthMismatch => Self::Integrity("length mismatch"), + _ => Self::Integrity("authentication failed"), + } + } +} + +/// A snapshot of the node's at-rest state for one storage operation. +#[derive(Clone)] +pub struct AtRestView { + pub mode: AtRestMode, + pub encrypting: bool, + pub provider: Option>, +} + +impl AtRestView { + /// The state a store without at-rest wiring behaves as: `off`, no keys. + pub fn off() -> Self { + Self { + mode: AtRestMode::Off, + encrypting: false, + provider: None, + } + } + + pub fn of(runtime: Option<&AtRestRuntime>) -> Self { + match runtime { + None => Self::off(), + Some(rt) => Self { + mode: rt.mode(), + encrypting: rt.encrypting(), + provider: rt.provider(), + }, + } + } + + /// The object-name key, if keys are set up and unlocked. + pub fn object_name_key(&self) -> Option { + self.provider.as_ref()?.object_name_key().ok() + } +} + +/// The sealed name of a block: backends lay it out as `v1/{space_tag}/{object_tag}`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SealedName { + pub space_tag: String, + pub object_tag: String, +} + +impl SealedName { + pub fn new(nk: &ObjectNameKey, space: &SpaceId, hash: &Hash) -> Self { + let full = tinycloud_core::at_rest::sealed_object_name(nk, &space.to_string(), hash); + let mut parts = full.splitn(3, '/').skip(1); + let (space_tag, object_tag) = (parts.next(), parts.next()); + Self { + space_tag: space_tag.expect("v1/{tag}/{obj}").to_owned(), + object_tag: object_tag.expect("v1/{tag}/{obj}").to_owned(), + } + } + + /// The S3 object key / FS relative path. + pub fn key(&self) -> String { + format!( + "{}{}/{}", + tinycloud_core::at_rest::SEALED_NAME_PREFIX, + self.space_tag, + self.object_tag + ) + } +} + +/// The pseudonymous size-accounting tag of a space, if keys are unlocked. +pub fn space_tag_of(view: &AtRestView, space: &SpaceId) -> Option { + view.object_name_key() + .map(|nk| space_tag(&nk, &space.to_string())) +} + +/// Where a new block is written. +pub enum WriteTarget { + Legacy, + Sealed { name: SealedName, key: DataKey }, +} + +/// What a stored S3 object name is, for the boot size scan. +#[derive(Debug, PartialEq, Eq)] +pub enum ScannedObject<'a> { + /// `{space}/{b64url digest}`. + Legacy(SpaceId), + /// `v1/{space_tag}/{object_tag}`: counted under its tag. + Sealed(&'a str), + /// Anything else, including the key mirror under `_tc/`. + Other, +} + +/// The key mirror's prefix in the block bucket (`_tc/at-rest/...`). +pub const RESERVED_PREFIX: &str = "_tc/"; + +pub fn scanned_object(key: &str) -> ScannedObject<'_> { + if key.starts_with(RESERVED_PREFIX) { + return ScannedObject::Other; + } + if let Some(rest) = key.strip_prefix(tinycloud_core::at_rest::SEALED_NAME_PREFIX) { + return match rest.split_once('/') { + Some((tag, obj)) if !tag.is_empty() && !obj.is_empty() && !obj.contains('/') => { + ScannedObject::Sealed(tag) + } + _ => ScannedObject::Other, + }; + } + match key.rsplit_once('/').and_then(|(o, _)| o.parse().ok()) { + Some(space) => ScannedObject::Legacy(space), + None => ScannedObject::Other, + } +} + +pub async fn write_target( + view: &AtRestView, + space: &SpaceId, + hash: &Hash, +) -> Result { + if view.mode == AtRestMode::Off { + return Ok(WriteTarget::Legacy); + } + if !view.encrypting { + return Err(AtRestStoreError::Unavailable("encryption is not enabled")); + } + let provider = view + .provider + .as_ref() + .ok_or(AtRestStoreError::Unavailable("at-rest keys are not set up"))?; + let nk = provider.object_name_key()?; + let key = provider.current_key(&space.to_string()).await?; + Ok(WriteTarget::Sealed { + name: SealedName::new(&nk, space, hash), + key, + }) +} + +/// Which names a read tries, in order. +pub enum ReadPlan { + /// No keys: the legacy name only. + Legacy, + /// Keys locked: the legacy name only, and a miss there is a 503 (a sealed + /// copy may exist but cannot be named or opened). + LegacyWhileLocked, + /// The sealed name, then the legacy name. + SealedThenLegacy { + name: SealedName, + provider: Arc, + }, + /// `strict`: the sealed name only. + Sealed { + name: SealedName, + provider: Arc, + }, +} + +pub fn read_plan( + view: &AtRestView, + space: &SpaceId, + hash: &Hash, +) -> Result { + let strict = view.mode == AtRestMode::Strict; + let Some(provider) = view.provider.clone() else { + return if strict { + Err(AtRestStoreError::Unavailable("at-rest keys are not set up")) + } else { + Ok(ReadPlan::Legacy) + }; + }; + match provider.object_name_key() { + Ok(nk) => { + let name = SealedName::new(&nk, space, hash); + Ok(if strict { + ReadPlan::Sealed { name, provider } + } else { + ReadPlan::SealedThenLegacy { name, provider } + }) + } + Err(DataKeyError::Locked) if !strict => Ok(ReadPlan::LegacyWhileLocked), + Err(e) => Err(e.into()), + } +} + +fn identity<'a>(space: &'a str, hash: &'a Hash) -> BlobIdentity<'a> { + BlobIdentity::KvBlock { + space, + content_hash: hash, + } +} + +/// Reads and parses the header at the start of a sealed object. +pub async fn read_header(reader: &mut R) -> Result { + let mut b = [0u8; HEADER_LEN]; + reader + .read_exact(&mut b) + .await + .map_err(|_| AtRestStoreError::Integrity("object shorter than its header"))?; + Ok(Header::parse(&b)?) +} + +/// The header's declared size must match the stored object's size. Still a +/// hint (the header is authenticated only when a segment decrypts), but it +/// rejects truncated or padded objects before any byte is served. +fn check_object_len(header: &Header, object_len: u64) -> Result<(), AtRestStoreError> { + if header.ciphertext_len() == object_len { + Ok(()) + } else { + Err(AtRestStoreError::Integrity( + "object size does not match its header", + )) + } +} + +/// Opens a whole sealed object. `reader` yields the object from byte 0 and +/// `object_len` is the stored size. The returned content reports the +/// plaintext length and yields only authenticated bytes; a corrupt object +/// fails part way with an `InvalidData` I/O error, which the HTTP layer turns +/// into an aborted body. +pub async fn open_sealed( + provider: &dyn DataKeyProvider, + space: &SpaceId, + hash: &Hash, + mut reader: R, + object_len: u64, +) -> Result>, AtRestStoreError> { + let header = read_header(&mut reader).await?; + check_object_len(&header, object_len)?; + let space = space.to_string(); + let key = provider.key(&space, header.dek_id()).await?; + let plaintext_len = header.plaintext_len(); + let decrypt = DecryptReader::new(&key, &identity(&space, hash), &header, reader)?; + Ok(Content::new(plaintext_len, decrypt)) +} + +/// A sealed range read, planned from the object's header. +pub struct SealedRange { + header: Header, + key: DataKey, + /// `None`: unsatisfiable against the plaintext length. + pub range: Option, +} + +impl SealedRange { + pub fn plaintext_len(&self) -> u64 { + self.header.plaintext_len() + } + + /// Wraps the fetched ciphertext span (`range.ciphertext_start()..=end`). + pub fn reader( + &self, + space: &SpaceId, + hash: &Hash, + range: SegmentRange, + inner: R, + ) -> Result, AtRestStoreError> { + let space = space.to_string(); + Ok(DecryptReader::range( + &self.key, + &identity(&space, hash), + &self.header, + range, + inner, + )?) + } +} + +/// Plans a range read from the object's first [`HEADER_LEN`] bytes. +pub async fn plan_range( + provider: &dyn DataKeyProvider, + space: &SpaceId, + header_bytes: &[u8], + object_len: u64, + spec: ByteRangeSpec, +) -> Result { + let header = Header::parse(header_bytes)?; + check_object_len(&header, object_len)?; + let key = provider.key(&space.to_string(), header.dek_id()).await?; + let range = header.map_range(spec); + Ok(SealedRange { header, key, range }) +} + +/// Seals an in-memory block, off the async executor. +pub async fn seal_vec( + key: DataKey, + space: &SpaceId, + hash: &Hash, + plaintext: Vec, +) -> Result, AtRestStoreError> { + let space = space.to_string(); + let hash = *hash; + tokio::task::spawn_blocking(move || { + tinycloud_core::at_rest::seal(&key, &identity(&space, &hash), plaintext) + }) + .await + .map_err(|_| AtRestStoreError::Integrity("sealing task failed"))? + .map_err(Into::into) +} + +/// Seals the staged plaintext file `src` (`plaintext_len` bytes) into a new +/// temporary file in the same directory, streaming one segment at a time. +pub async fn seal_file( + key: &DataKey, + space: &SpaceId, + hash: &Hash, + src: &Path, + plaintext_len: u64, +) -> Result { + let dir = src.parent().unwrap_or_else(|| Path::new(".")); + let out = tempfile::NamedTempFile::new_in(dir)?; + let space = space.to_string(); + let input = tokio::fs::File::open(src).await?.compat(); + let mut reader = EncryptReader::new(key, &identity(&space, hash), plaintext_len, input) + .map_err(AtRestStoreError::from)?; + let mut writer = tokio::fs::File::from_std(out.reopen()?).compat_write(); + futures::io::copy(&mut reader, &mut writer).await?; + futures::io::AsyncWriteExt::close(&mut writer).await?; + Ok(out) +} + +#[derive(Debug, thiserror::Error)] +pub enum SealFileError { + #[error(transparent)] + AtRest(#[from] AtRestStoreError), + #[error(transparent)] + Io(#[from] std::io::Error), +} + +/// Rate-limits the boot-scan warning for sealed objects whose size is not a +/// valid TCE1 length: the first few are logged, the rest only counted. +#[derive(Default)] +pub struct InvalidLengthLog { + seen: u64, +} + +impl InvalidLengthLog { + const LOGGED: u64 = 5; + + pub fn record(&mut self, object_len: u64) { + self.seen += 1; + if self.seen <= Self::LOGGED { + tracing::warn!( + object_len, + "sealed object size is not a valid TCE1 length; counted as 0 bytes" + ); + } + } + + pub fn finish(self) { + if self.seen > Self::LOGGED { + tracing::warn!( + total = self.seen, + "sealed objects with invalid TCE1 lengths (only the first were logged)" + ); + } + } +} + +#[cfg(test)] +mod tests; diff --git a/tinycloud-node-server/src/storage/sealed/tests.rs b/tinycloud-node-server/src/storage/sealed/tests.rs new file mode 100644 index 00000000..bd895c34 --- /dev/null +++ b/tinycloud-node-server/src/storage/sealed/tests.rs @@ -0,0 +1,400 @@ +use super::*; +use futures::io::{AsyncReadExt, Cursor}; +use rocket::http::Status; +use tinycloud_core::at_rest::{ciphertext_len, DekId, FixedKeyProvider, SEGMENT_LEN}; + +const SEG: usize = SEGMENT_LEN; + +fn space() -> SpaceId { + "tinycloud:pkh:eip155:1:0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266:default" + .parse() + .unwrap() +} + +fn view(mode: AtRestMode, encrypting: bool, provider: Option) -> AtRestView { + AtRestView { + mode, + encrypting, + provider: provider.map(|p| Arc::new(p) as Arc), + } +} + +fn keys() -> FixedKeyProvider { + FixedKeyProvider::new([7; 32]) +} + +fn bytes(n: usize) -> Vec { + (0..n).map(|i| (i * 31 % 251) as u8).collect() +} + +async fn sealed(p: &FixedKeyProvider, data: &[u8]) -> (Hash, Vec) { + let hash = tinycloud_core::hash::hash(data); + let key = p.current_key(&space().to_string()).await.unwrap(); + ( + hash, + seal_vec(key, &space(), &hash, data.to_vec()).await.unwrap(), + ) +} + +async fn read_to_end(mut r: R) -> (Vec, Option) { + let mut out = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + match r.read(&mut buf).await { + Ok(0) => return (out, None), + Ok(n) => out.extend_from_slice(&buf[..n]), + Err(e) => return (out, Some(e)), + } + } +} + +#[tokio::test] +async fn write_target_follows_the_mode_and_gate() { + let s = space(); + let h = tinycloud_core::hash::hash(b"x"); + // off writes legacy even when keys exist. + let off = view(AtRestMode::Off, true, Some(keys())); + assert!(matches!( + write_target(&off, &s, &h).await, + Ok(WriteTarget::Legacy) + )); + // encrypt with the gate open seals under the current key and the sealed name. + let on = view(AtRestMode::Encrypt, true, Some(keys())); + match write_target(&on, &s, &h).await.unwrap() { + WriteTarget::Sealed { name, key } => { + let nk = keys().object_name_key().unwrap(); + assert_eq!(name, SealedName::new(&nk, &s, &h)); + assert_eq!( + name.key(), + tinycloud_core::at_rest::sealed_object_name(&nk, &s.to_string(), &h) + ); + assert_eq!(key.id(), DekId::new(1)); + } + WriteTarget::Legacy => panic!("encrypt mode wrote legacy"), + } + // encrypt with the gate closed, keys missing or locked: 503, never plaintext. + for v in [ + view(AtRestMode::Encrypt, false, Some(keys())), + view(AtRestMode::Strict, false, None), + view(AtRestMode::Encrypt, true, None), + view(AtRestMode::Encrypt, true, Some(keys().locked())), + ] { + let e = write_target(&v, &s, &h).await.err().expect("must refuse"); + assert_eq!(e.status(), Status::ServiceUnavailable, "{e}"); + } +} + +#[test] +fn read_plan_follows_the_mode_and_key_state() { + let s = space(); + let h = tinycloud_core::hash::hash(b"x"); + assert!(matches!( + read_plan(&AtRestView::off(), &s, &h), + Ok(ReadPlan::Legacy) + )); + assert!(matches!( + read_plan(&view(AtRestMode::Off, false, Some(keys())), &s, &h), + Ok(ReadPlan::SealedThenLegacy { .. }) + )); + assert!(matches!( + read_plan(&view(AtRestMode::Encrypt, true, Some(keys())), &s, &h), + Ok(ReadPlan::SealedThenLegacy { .. }) + )); + assert!(matches!( + read_plan( + &view(AtRestMode::Encrypt, false, Some(keys().locked())), + &s, + &h + ), + Ok(ReadPlan::LegacyWhileLocked) + )); + assert!(matches!( + read_plan(&view(AtRestMode::Strict, true, Some(keys())), &s, &h), + Ok(ReadPlan::Sealed { .. }) + )); + for v in [ + view(AtRestMode::Strict, false, Some(keys().locked())), + view(AtRestMode::Strict, false, None), + ] { + let e = read_plan(&v, &s, &h) + .err() + .expect("strict without keys refuses"); + assert_eq!(e.status(), Status::ServiceUnavailable); + } +} + +#[tokio::test] +async fn sealed_objects_round_trip_with_plaintext_length() { + let p = keys(); + for n in [0, 1, SEG - 1, SEG, SEG + 1, 2 * SEG + 7] { + let data = bytes(n); + let (h, ct) = sealed(&p, &data).await; + assert_eq!(ct.len() as u64, ciphertext_len(n as u64).unwrap()); + let len = ct.len() as u64; + let content = open_sealed(&p, &space(), &h, Cursor::new(ct), len) + .await + .unwrap(); + assert_eq!(content.len(), n as u64); + let (out, err) = read_to_end(content).await; + assert!(err.is_none()); + assert_eq!(out, data); + } +} + +#[tokio::test] +async fn sealed_open_fails_closed() { + let p = keys(); + let data = bytes(SEG + 10); + let (h, ct) = sealed(&p, &data).await; + let s = space(); + let len = ct.len() as u64; + + // Stored size differs from the header: refused before any byte. + let e = open_sealed( + &p, + &s, + &h, + Cursor::new(ct[..ct.len() - 1].to_vec()), + len - 1, + ) + .await + .err() + .unwrap(); + assert!(matches!(e, AtRestStoreError::Integrity(_)), "{e}"); + + // A later generation: unsupported, a hard error. + let mut v2 = ct.clone(); + v2[3] = b'2'; + v2[4] = 2; + let e = open_sealed(&p, &s, &h, Cursor::new(v2), len) + .await + .err() + .unwrap(); + assert!(matches!(e, AtRestStoreError::Unsupported(2)), "{e}"); + assert_eq!(e.status(), Status::InternalServerError); + + // Header names a key the space never had. + let newer = FixedKeyProvider::new([7; 32]).with_current(DekId::new(2)); + let (h2, ct2) = sealed(&newer, &data).await; + let len2 = ct2.len() as u64; + let e = open_sealed(&p, &s, &h2, Cursor::new(ct2), len2) + .await + .err() + .unwrap(); + assert!(matches!(e, AtRestStoreError::Integrity(_)), "{e}"); + + // Locked keys: 503. + let e = open_sealed(&keys().locked(), &s, &h, Cursor::new(ct.clone()), len) + .await + .err() + .unwrap(); + assert_eq!(e.status(), Status::ServiceUnavailable); + + // Bound to the content hash: another hash's name cannot serve it. + let other = tinycloud_core::hash::hash(b"other"); + let content = open_sealed(&p, &s, &other, Cursor::new(ct.clone()), len) + .await + .unwrap(); + let (out, err) = read_to_end(content).await; + assert!(out.is_empty()); + assert_eq!(err.unwrap().kind(), std::io::ErrorKind::InvalidData); + + // Not TCE at all (a legacy blob under a sealed name): never served as plaintext. + let e = open_sealed(&p, &s, &h, Cursor::new(data.clone()), data.len() as u64) + .await + .err() + .unwrap(); + assert!(matches!(e, AtRestStoreError::Integrity(_)), "{e}"); +} + +/// The header's length is unverified until a segment decrypts. A forged +/// length can mis-state the total, but the body aborts and yields no bytes +/// that were not authenticated. +#[tokio::test] +async fn forged_header_length_aborts_without_extra_bytes() { + let p = keys(); + let data = bytes(3 * SEG); + let (h, ct) = sealed(&p, &data).await; + for forged in [SEG as u64, 5 * SEG as u64, 1] { + let mut f = ct.clone(); + f[12..20].copy_from_slice(&forged.to_be_bytes()); + // Pad or cut so the stored size matches the forged header. + f.resize(ciphertext_len(forged).unwrap() as usize, 0xAB); + let len = f.len() as u64; + let content = open_sealed(&p, &space(), &h, Cursor::new(f), len) + .await + .unwrap(); + assert_eq!( + content.len(), + forged, + "the total is the (unverified) header's" + ); + let (out, err) = read_to_end(content).await; + assert!( + out.is_empty(), + "forged {forged}: {} bytes released", + out.len() + ); + assert_eq!(err.unwrap().kind(), std::io::ErrorKind::InvalidData); + } +} + +#[tokio::test] +async fn sealed_ranges_equal_plaintext_slices() { + let p = keys(); + let n = 3 * SEG + 5; + let data = bytes(n); + let (h, ct) = sealed(&p, &data).await; + let s = space(); + let len = ct.len() as u64; + let edges = [0, 1, SEG - 1, SEG, SEG + 1, 2 * SEG, n - 1]; + for &a in &edges { + for &b in &edges { + if a > b { + continue; + } + let spec = ByteRangeSpec::Inclusive { + start: a as u64, + end: b as u64, + }; + let plan = plan_range(&p, &s, &ct[..HEADER_LEN], len, spec) + .await + .unwrap(); + let r = plan.range.expect("satisfiable"); + assert_eq!(plan.plaintext_len(), n as u64); + let span = ct[r.ciphertext_start() as usize..=r.ciphertext_end() as usize].to_vec(); + let (out, err) = read_to_end(plan.reader(&s, &h, r, Cursor::new(span)).unwrap()).await; + assert!(err.is_none()); + assert_eq!(out, data[a..=b], "{a}..={b}"); + } + } + let plan = plan_range( + &p, + &s, + &ct[..HEADER_LEN], + len, + ByteRangeSpec::From { start: n as u64 }, + ) + .await + .unwrap(); + assert!(plan.range.is_none(), "416"); + assert_eq!( + plan.plaintext_len(), + n as u64, + "416 total is the plaintext length" + ); +} + +#[tokio::test] +async fn sealed_files_stream_to_the_same_envelope() { + let p = keys(); + let s = space(); + let dir = tempfile::tempdir().unwrap(); + let data = bytes(2 * SEG + 123); + let h = tinycloud_core::hash::hash(&data); + let src = dir.path().join("staged"); + std::fs::write(&src, &data).unwrap(); + let key = p.current_key(&s.to_string()).await.unwrap(); + + let out = seal_file(&key, &s, &h, &src, data.len() as u64) + .await + .unwrap(); + assert_eq!( + out.path().parent(), + Some(dir.path()), + "sealed beside the staged file" + ); + let ct = std::fs::read(out.path()).unwrap(); + let len = ct.len() as u64; + let (plain, err) = + read_to_end(open_sealed(&p, &s, &h, Cursor::new(ct), len).await.unwrap()).await; + assert!(err.is_none()); + assert_eq!(plain, data); + + // A staged file that is not the declared length is refused. + assert!(seal_file(&key, &s, &h, &src, data.len() as u64 + 1) + .await + .is_err()); + assert!(seal_file(&key, &s, &h, &src, data.len() as u64 - 1) + .await + .is_err()); +} + +#[test] +fn boot_scan_classifies_names() { + let s = space(); + assert_eq!( + scanned_object(&format!("{s}/QU63=")), + ScannedObject::Legacy(s) + ); + assert_eq!(scanned_object("v1/tagA/obj"), ScannedObject::Sealed("tagA")); + for other in [ + "_tc/at-rest/dek-x-1.json", + "_tc/at-rest/ark-1.sealed", + "v1/tagA", + "v1//obj", + "v1/tagA/obj/extra", + "not-a-space/obj", + "loose", + ] { + assert_eq!(scanned_object(other), ScannedObject::Other, "{other}"); + } +} + +#[test] +fn space_tags_need_unlocked_keys() { + let s = space(); + assert_eq!( + space_tag_of(&view(AtRestMode::Encrypt, false, Some(keys().locked())), &s), + None + ); + assert_eq!(space_tag_of(&AtRestView::off(), &s), None); + let nk = keys().object_name_key().unwrap(); + assert_eq!( + space_tag_of(&view(AtRestMode::Off, false, Some(keys())), &s), + Some(space_tag(&nk, &s.to_string())) + ); +} + +#[test] +fn at_rest_causes_map_to_503_or_500_on_kv_reads() { + use crate::storage::{file_system::FileSystemStoreError, s3::S3StoreError}; + use tinycloud_core::storage::either::EitherError; + let read = |e: AtRestStoreError, s3: bool| { + let store = if s3 { + EitherError::A(S3StoreError::AtRest(e)) + } else { + EitherError::B(FileSystemStoreError::AtRest(e)) + }; + crate::routes::kv_read_failure(EitherError::B(store)).0 + }; + for s3 in [true, false] { + assert_eq!( + read(AtRestStoreError::Unavailable("locked"), s3), + Status::ServiceUnavailable + ); + assert_eq!( + read(AtRestStoreError::KeyStore("pg".into()), s3), + Status::ServiceUnavailable + ); + assert_eq!( + read(AtRestStoreError::Integrity("x"), s3), + Status::InternalServerError + ); + assert_eq!( + read(AtRestStoreError::Unsupported(2), s3), + Status::InternalServerError + ); + assert_eq!( + read(AtRestStoreError::LegacyRefused, s3), + Status::InternalServerError + ); + } + let other = EitherError::B(EitherError::B(FileSystemStoreError::Io( + std::io::Error::other("disk"), + ))); + assert_eq!( + crate::routes::kv_read_failure(other).0, + Status::InternalServerError + ); +} diff --git a/tinycloud-node-server/src/storage/size.rs b/tinycloud-node-server/src/storage/size.rs index 9fac85d6..740bb0cc 100644 --- a/tinycloud-node-server/src/storage/size.rs +++ b/tinycloud-node-server/src/storage/size.rs @@ -1,38 +1,127 @@ -use std::{ - collections::HashMap, - ops::{AddAssign, SubAssign}, - sync::Arc, -}; +use std::{collections::HashMap, sync::Arc}; use tinycloud_auth::resource::SpaceId; use tokio::sync::RwLock; +/// Per-space stored plaintext bytes, for quota and usage. +/// +/// Legacy objects are named by space, so a boot scan attributes them to a +/// [`SpaceId`]. Sealed objects (TC-887) are named by a pseudonymous +/// `space_tag` that only the unlocked object-name key can compute, so the +/// scan files them by tag and a lookup joins the two once the caller can +/// name the tag. While keys are locked a space reports its legacy bytes only. #[derive(Debug, Clone, Default)] -pub struct SpaceSizes(Arc>>); +pub struct SpaceSizes(Arc>); + +#[derive(Debug, Default)] +struct Sizes { + spaces: HashMap, + sealed: HashMap, +} impl SpaceSizes { pub fn new() -> Self { - Self(Arc::new(RwLock::new(HashMap::new()))) + Self::default() } + + pub fn from_scan(spaces: HashMap, sealed: HashMap) -> Self { + Self(Arc::new(RwLock::new(Sizes { spaces, sealed }))) + } + pub async fn init_size(&self, space: SpaceId) { - self.0.write().await.insert(space, 0); + self.0.write().await.spaces.insert(space, 0); } - pub async fn increment_size(&self, space: &SpaceId, size: u64) { - if let Some(s) = self.0.write().await.get_mut(space) { - s.add_assign(size) + + /// Adds `size` bytes to the space's entry, or to its sealed-tag entry if + /// only that is known (a space whose objects are all sealed after a + /// restart). A space known under neither is left untouched, as before. + pub async fn increment_size(&self, space: &SpaceId, tag: Option<&str>, size: u64) { + let mut s = self.0.write().await; + if let Some(n) = s.spaces.get_mut(space) { + *n = n.saturating_add(size); + } else if let Some(n) = tag.and_then(|t| s.sealed.get_mut(t)) { + *n = n.saturating_add(size); } } - pub async fn decrement_size(&self, space: &SpaceId, size: u64) { - if let Some(s) = self.0.write().await.get_mut(space) { - s.sub_assign(size) + + /// Subtracts `size` bytes, from the sealed-tag entry first when the + /// removed object was sealed (that is where a boot scan counted it). + pub async fn decrement_size(&self, space: &SpaceId, tag: Option<&str>, size: u64) { + let mut s = self.0.write().await; + let mut left = size; + if let Some(n) = tag.and_then(|t| s.sealed.get_mut(t)) { + let take = left.min(*n); + *n -= take; + left -= take; + } + if let Some(n) = s.spaces.get_mut(space) { + *n = n.saturating_sub(left); } } - pub async fn get_size(&self, space: &SpaceId) -> Option { - self.0.read().await.get(space).copied() + + /// The space's bytes, joining its sealed-tag entry when `tag` is given. + /// `None` when the space is known under neither. + pub async fn get_size(&self, space: &SpaceId, tag: Option<&str>) -> Option { + let s = self.0.read().await; + let legacy = s.spaces.get(space).copied(); + let sealed = tag.and_then(|t| s.sealed.get(t)).copied(); + match (legacy, sealed) { + (None, None) => None, + (a, b) => Some(a.unwrap_or(0).saturating_add(b.unwrap_or(0))), + } } } impl From> for SpaceSizes { fn from(map: HashMap) -> Self { - Self(Arc::new(RwLock::new(map))) + Self::from_scan(map, HashMap::new()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn space(n: &str) -> SpaceId { + format!("tinycloud:pkh:eip155:1:0x0000000000000000000000000000000000000001:{n}") + .parse() + .unwrap() + } + + #[tokio::test] + async fn sealed_bytes_join_only_when_the_tag_is_known() { + let a = space("a"); + let sizes = SpaceSizes::from_scan( + HashMap::from([(a.clone(), 100)]), + HashMap::from([("tagA".to_string(), 40)]), + ); + // Locked: no tag, legacy only. + assert_eq!(sizes.get_size(&a, None).await, Some(100)); + // Unlocked: legacy + sealed, without any rescan. + assert_eq!(sizes.get_size(&a, Some("tagA")).await, Some(140)); + assert_eq!(sizes.get_size(&a, Some("other")).await, Some(100)); + } + + #[tokio::test] + async fn a_sealed_only_space_is_known_and_grows_under_its_tag() { + let b = space("b"); + let sizes = SpaceSizes::from_scan(HashMap::new(), HashMap::from([("tagB".into(), 10)])); + assert_eq!(sizes.get_size(&b, None).await, None); + assert_eq!(sizes.get_size(&b, Some("tagB")).await, Some(10)); + sizes.increment_size(&b, Some("tagB"), 5).await; + assert_eq!(sizes.get_size(&b, Some("tagB")).await, Some(15)); + sizes.decrement_size(&b, Some("tagB"), 20).await; + assert_eq!(sizes.get_size(&b, Some("tagB")).await, Some(0)); + } + + #[tokio::test] + async fn sealed_removal_drains_the_tag_entry_first() { + let c = space("c"); + let sizes = SpaceSizes::from_scan( + HashMap::from([(c.clone(), 50)]), + HashMap::from([("tagC".into(), 30)]), + ); + 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)); } }